diff --git a/packages/client/src/effect/api/api.ts b/packages/client/src/effect/api/api.ts index bf114ca1f239..68ae125a3c44 100644 --- a/packages/client/src/effect/api/api.ts +++ b/packages/client/src/effect/api/api.ts @@ -2374,7 +2374,9 @@ export type ExperimentalPersistentPtyConnectTokenOperation = ( export type ExperimentalScheduleCreateInput = { readonly sessionID: Session.ID readonly name?: string | undefined - readonly when: Schedule.WhenInput + readonly when?: Schedule.WhenInput | undefined + readonly if?: Schedule.ConditionInput | undefined + readonly until?: Schedule.ConditionInput | undefined readonly ends?: Schedule.Ends | undefined readonly prompt: string } diff --git a/packages/client/src/effect/generated/client.ts b/packages/client/src/effect/generated/client.ts index bd3096140ed2..b57d2c379aaa 100644 --- a/packages/client/src/effect/generated/client.ts +++ b/packages/client/src/effect/generated/client.ts @@ -1576,6 +1576,8 @@ const EndpointExperimentalScheduleCreate = sessionID: input["sessionID"], name: input["name"], when: input["when"], + if: input["if"], + until: input["until"], ends: input["ends"], prompt: input["prompt"], }, diff --git a/packages/client/src/promise/generated/client.ts b/packages/client/src/promise/generated/client.ts index 0c59497f859e..cadaa8b85a3f 100644 --- a/packages/client/src/promise/generated/client.ts +++ b/packages/client/src/promise/generated/client.ts @@ -1499,6 +1499,8 @@ export function make(options: ClientOptions) { sessionID: input["sessionID"], name: input["name"], when: input["when"], + if: input["if"], + until: input["until"], ends: input["ends"], prompt: input["prompt"], }, diff --git a/packages/client/src/promise/generated/types.ts b/packages/client/src/promise/generated/types.ts index 8bd2108c4ea3..64ba3f90b124 100644 --- a/packages/client/src/promise/generated/types.ts +++ b/packages/client/src/promise/generated/types.ts @@ -327,11 +327,13 @@ export type ScheduleWhenAt = { type: "at"; at: string; tz?: string } export type ScheduleWhenCron = { type: "cron"; cron: string; tz?: string } +export type ScheduleConditionShell = { type: "shell"; run: string; is?: string; matches?: string; changes?: boolean } + export type ScheduleActionPrompt = { type: "prompt"; prompt: string } export type ScheduleStatus = "active" | "paused" | "done" -export type ScheduleOutcome = "fired" | "late" | "skipped" | "disabled" | "failed" +export type ScheduleOutcome = "fired" | "late" | "checked" | "expired" | "skipped" | "disabled" | "failed" export type ReferenceGitSource = { type: "git"; repository: string; branch?: string } @@ -905,6 +907,8 @@ export type ScheduleWhenIn = { type: "in"; in: ScheduleDuration; tz?: string } export type ScheduleWhenEvery = { type: "every"; every: ScheduleDuration; tz?: string } +export type ScheduleCondition = ScheduleConditionShell + export type ScheduleAction = ScheduleActionPrompt export type VcsInfo = { provider?: string; branch: VcsBranch } @@ -2538,6 +2542,8 @@ export type ScheduleInfo = { name?: string when: ScheduleWhen tz: string + if?: ScheduleCondition + until?: ScheduleCondition action: ScheduleAction status: ScheduleStatus nextDueAt?: number @@ -2546,6 +2552,7 @@ export type ScheduleInfo = { runCount: number lastFiredAt?: number lastOutcome?: ScheduleOutcome + lastOutput?: string time: { created: number; updated: number } } @@ -4405,11 +4412,44 @@ export type ExperimentalMediaGetOutput = globalThis.Uint8Array export type ExperimentalScheduleCreateInput = { readonly sessionID: string readonly name?: string | undefined - readonly when: - | { readonly in: string; readonly at?: never; readonly every?: never; readonly cron?: never; readonly tz?: string } - | { readonly at: string; readonly in?: never; readonly every?: never; readonly cron?: never; readonly tz?: string } - | { readonly every: string; readonly in?: never; readonly at?: never; readonly cron?: never; readonly tz?: string } - | { readonly cron: string; readonly in?: never; readonly at?: never; readonly every?: never; readonly tz?: string } + readonly when?: + | ( + | { + readonly in: string + readonly at?: never + readonly every?: never + readonly cron?: never + readonly tz?: string + } + | { + readonly at: string + readonly in?: never + readonly every?: never + readonly cron?: never + readonly tz?: string + } + | { + readonly every: string + readonly in?: never + readonly at?: never + readonly cron?: never + readonly tz?: string + } + | { + readonly cron: string + readonly in?: never + readonly at?: never + readonly every?: never + readonly tz?: string + } + ) + | undefined + readonly if?: + | { readonly run: string; readonly is?: string; readonly matches?: string; readonly changes?: boolean } + | undefined + readonly until?: + | { readonly run: string; readonly is?: string; readonly matches?: string; readonly changes?: boolean } + | undefined readonly ends?: { readonly after?: string; readonly at?: string; readonly times?: number } | undefined readonly prompt: string } diff --git a/packages/core/schema.json b/packages/core/schema.json index bb629130d75f..9f524b6020b9 100644 --- a/packages/core/schema.json +++ b/packages/core/schema.json @@ -1,8 +1,8 @@ { "version": "7", "dialect": "sqlite", - "id": "e5daa0af-cae7-4057-919a-ebf44dfced0a", - "prevIds": ["dc16dd12-cb76-432f-8108-0c3aadaf54bb"], + "id": "0ac88673-a402-45a0-b931-e694d59ccefa", + "prevIds": ["e5daa0af-cae7-4057-919a-ebf44dfced0a"], "ddl": [ { "name": "account_state", @@ -792,6 +792,16 @@ "entityType": "columns", "table": "schedule" }, + { + "type": "text", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "check", + "entityType": "columns", + "table": "schedule" + }, { "type": "text", "notNull": true, @@ -862,6 +872,26 @@ "entityType": "columns", "table": "schedule" }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": "0", + "generated": null, + "name": "checks", + "entityType": "columns", + "table": "schedule" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": "0", + "generated": null, + "name": "failures", + "entityType": "columns", + "table": "schedule" + }, { "type": "integer", "notNull": false, @@ -902,6 +932,16 @@ "entityType": "columns", "table": "schedule" }, + { + "type": "text", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "state", + "entityType": "columns", + "table": "schedule" + }, { "type": "integer", "notNull": true, diff --git a/packages/core/src/database/migration.gen.ts b/packages/core/src/database/migration.gen.ts index f5afbd4fb6d5..7503530ea4b0 100644 --- a/packages/core/src/database/migration.gen.ts +++ b/packages/core/src/database/migration.gen.ts @@ -56,6 +56,7 @@ import m53 from "./migration/20261025154125_drop_session_pending.js" import m54 from "./migration/20261026223005_schedule.js" import m55 from "./migration/20261029002625_session_part.js" import m56 from "./migration/20261029073502_session_time_busy.js" +import m57 from "./migration/20261029165540_schedule_checks.js" export const migrations = [ m00, @@ -115,4 +116,5 @@ export const migrations = [ m54, m55, m56, + m57, ] satisfies DatabaseMigration.Migration[] diff --git a/packages/core/src/database/migration/20261029165540_schedule_checks.ts b/packages/core/src/database/migration/20261029165540_schedule_checks.ts new file mode 100644 index 000000000000..fabb991b29dc --- /dev/null +++ b/packages/core/src/database/migration/20261029165540_schedule_checks.ts @@ -0,0 +1,16 @@ +import { Effect } from "effect" +import type { DatabaseMigration } from "../migration.js" + +const migration: DatabaseMigration.Migration = { + id: "20261029165540_schedule_checks", + up(tx) { + return Effect.gen(function* () { + yield* tx.run(`ALTER TABLE \`schedule\` ADD \`check\` text;`) + yield* tx.run(`ALTER TABLE \`schedule\` ADD \`checks\` integer DEFAULT 0 NOT NULL;`) + yield* tx.run(`ALTER TABLE \`schedule\` ADD \`failures\` integer DEFAULT 0 NOT NULL;`) + yield* tx.run(`ALTER TABLE \`schedule\` ADD \`state\` text;`) + }) + }, +} + +export default migration diff --git a/packages/core/src/database/schema.gen.ts b/packages/core/src/database/schema.gen.ts index 2cb123ff319b..db107c32aad7 100644 --- a/packages/core/src/database/schema.gen.ts +++ b/packages/core/src/database/schema.gen.ts @@ -122,6 +122,7 @@ const schema: Omit = { \`session_id\` text, \`name\` text, \`action\` text NOT NULL, + \`check\` text, \`when\` text NOT NULL, \`tz\` text NOT NULL, \`status\` text DEFAULT 'active' NOT NULL, @@ -129,10 +130,13 @@ const schema: Omit = { \`ends_at\` integer, \`remaining_runs\` integer, \`run_count\` integer DEFAULT 0 NOT NULL, + \`checks\` integer DEFAULT 0 NOT NULL, + \`failures\` integer DEFAULT 0 NOT NULL, \`last_fired_at\` integer, \`last_message_id\` text, \`last_outcome\` text, \`retry_at\` integer, + \`state\` text, \`time_created\` integer NOT NULL, \`time_updated\` integer NOT NULL, CONSTRAINT \`fk_schedule_session_id_session_v2_id_fk\` FOREIGN KEY (\`session_id\`) REFERENCES \`session_v2\`(\`id\`) ON DELETE CASCADE diff --git a/packages/core/src/permission.ts b/packages/core/src/permission.ts index 2192c0e07a26..a2899fbf2c50 100644 --- a/packages/core/src/permission.ts +++ b/packages/core/src/permission.ts @@ -48,11 +48,23 @@ export interface AskResult { export { Event } from "@opencode/schema/permission" +/** + * Rules that settle a request which would otherwise ask, for work nobody is present to answer, such as a schedule's + * check. A request they do not allow is denied. Configured denies still win, as they are evaluated first. + */ +export const Unattended = Context.Reference("@opencode/Permission/Unattended", { + defaultValue: () => undefined, +}) + export class DeclinedError extends Schema.TaggedError()("Permission.DeclinedError", {}) {} export class CorrectedError extends Schema.TaggedError()("Permission.CorrectedError", { feedback: Schema.String, -}) {} +}) { + override get message() { + return this.feedback + } +} export class BlockedError extends Schema.TaggedError()("Permission.BlockedError", { rules: Permission.Ruleset, @@ -174,7 +186,17 @@ const layer = Layer.effect( source: input.source, effect, }) - return { effect: event.effect, message: event.message, rules: all } + const unattended = yield* Unattended + if (event.effect !== "ask" || unattended === undefined) + return { effect: event.effect, message: event.message, rules: all } + const settled = [...all, ...unattended] + const allowed = input.resources.every((resource) => evaluate(input.action, resource, settled).effect === "allow") + if (allowed) return { effect: "allow" as const, rules: all } + return { + effect: "deny" as const, + message: `Permission ${input.action} needs approval, and nobody is present to give it`, + rules: all, + } }) function request(input: AssertInput, message?: string): Request { diff --git a/packages/core/src/plugin/command/schedule.txt b/packages/core/src/plugin/command/schedule.txt index b003c0fb12ea..42ac4847d084 100644 --- a/packages/core/src/plugin/command/schedule.txt +++ b/packages/core/src/plugin/command/schedule.txt @@ -4,7 +4,7 @@ Request: $ARGUMENTS Use the schedule tools in the `opencode` namespace. They only see this session's schedules. -- To schedule something, call `schedule_create`. Choose exactly one `when`: `{ in: "20 minutes" }`, `{ at: "2026-10-27T09:00" }`, `{ every: "1 hour" }` or `{ cron: "0 9 * * 1-5" }`. Wall-clock times use the configured time zone unless you pass `tz`. To stop a repeating schedule, add `ends`: `{ after: "8 hours" }`, `{ at: "..." }` or `{ times: 3 }`. Write `prompt` as the instruction you will receive when it fires, with enough context to act on it without this conversation. +- To schedule something, call `schedule_create`. Choose `when` (`{ in: "20 minutes" }`, `{ at: "2026-10-27T09:00" }`, `{ every: "1 hour" }` or `{ cron: "0 9 * * 1-5" }`, optional with `until` which defaults to every 30s). Add `until: { run: "..." }` to wait once for a shell condition (exit 0, `is`, `matches`, or `changes: true`, defaulting to a 24h timeout), or `if: { run: "..." }` to gate each recurring slot without a model call. To stop a repeating schedule, add `ends`: `{ after: "8 hours" }`, `{ at: "..." }` or `{ times: 3 }`. Write `prompt` as the instruction you will receive when it wakes, with enough context to act on it without this conversation. - To pause, resume or delete one, find its id with `schedule_list`, then call `schedule_pause`, `schedule_resume` or `schedule_delete`. - If the request is empty, call `schedule_list` and show this session's schedules: name, when, status and next run. diff --git a/packages/core/src/schedule.ts b/packages/core/src/schedule.ts index 3d0bd0659371..96e096caf80b 100644 --- a/packages/core/src/schedule.ts +++ b/packages/core/src/schedule.ts @@ -4,29 +4,35 @@ import { and, asc, eq, gt, isNull, lte, min, or } from "drizzle-orm" import { Cause, Clock, Context, Cron, DateTime, Effect, Latch, Layer, Option, Result, Schema } from "effect" import { Node, makeGlobalNode } from "@opencode/util/effect/app-node" import { LayerNode } from "@opencode/util/effect/layer-node" -import { Schedule as ScheduleSchema } from "@opencode/schema/schedule" +import { + Condition, + ConditionInput, + CreateInput, + durationMillis, + type Ends, + ID, + Info, + Outcome, + Request, + Status, + type StoredCheck, + toCreateInput, + When, +} from "@opencode/schema/schedule" import { Config } from "./config.js" import { Database } from "./database/database.js" +import { FileSystem } from "./filesystem.js" import { Instance } from "./instance/service.js" +import type { Permission } from "./permission.js" import { Session } from "./session.js" import { SessionExecution } from "./session/execution.js" import { SessionMessage } from "./session/message.js" import { SessionSchema } from "./session/schema.js" import { SessionInboxTable } from "./session/sql.js" +import { ScheduleCheck } from "./schedule/check.js" import { ScheduleTable } from "./schedule/sql.js" -export const ID = ScheduleSchema.ID -export type ID = ScheduleSchema.ID -export const Info = ScheduleSchema.Info -export type Info = ScheduleSchema.Info -export const CreateInput = ScheduleSchema.CreateInput -export type CreateInput = ScheduleSchema.CreateInput -export const Request = ScheduleSchema.Request -export type Request = ScheduleSchema.Request -export const When = ScheduleSchema.When -export const Status = ScheduleSchema.Status -export const Outcome = ScheduleSchema.Outcome -export const toCreateInput = ScheduleSchema.toCreateInput +export { ID, Info, CreateInput, Request, Condition, ConditionInput, When, Status, Outcome, toCreateInput } export class NotFoundError extends Schema.TaggedError()("Schedule.NotFoundError", { scheduleID: ID, @@ -51,7 +57,9 @@ export class DisabledError extends Schema.TaggedError()("Schedule } export interface Interface { - readonly create: (input: CreateInput) => Effect.Effect + readonly create: ( + input: CreateInput, + ) => Effect.Effect readonly list: (input?: { readonly sessionID?: SessionSchema.ID }) => Effect.Effect readonly get: (id: ID) => Effect.Effect readonly pause: (id: ID) => Effect.Effect @@ -75,6 +83,9 @@ type Row = typeof ScheduleTable.$inferSelect const MINUTE = 60_000 const HOUR = 60 * MINUTE const MIN_INTERVAL = MINUTE +const MIN_CHECK_INTERVAL = 15_000 +const DEFAULT_UNTIL_EVERY: When = { type: "every", every: "30 seconds" } +const DEFAULT_UNTIL_AFTER = 24 * HOUR /** A firing this far past its due time is marked late. */ const LATE_AFTER = 60_000 /** The loop re-reads the table at least this often, so clock jumps and missed doorbells are caught. */ @@ -83,9 +94,13 @@ const MAX_SLEEP = 30_000 const DISABLED_RETRY = HOUR /** How long a schedule whose prompt failed waits before trying again. */ const FAILED_RETRY = 5 * MINUTE +/** How long a schedule whose check failed waits before trying again, up to MAX_CHECK_FAILURES. */ +const CHECK_FAILED_RETRY = MINUTE +const MAX_CHECK_FAILURES = 3 +const TICK_CONCURRENCY = 4 const duration = (value: string) => - Option.match(ScheduleSchema.durationMillis(value), { + Option.match(durationMillis(value), { onNone: () => Effect.fail(new InvalidError({ reason: `Invalid duration: ${value}` })), onSome: (millis) => millis > 0 @@ -95,7 +110,7 @@ const duration = (value: string) => /** Stored durations were validated on create. */ const millis = (value: string) => - Option.getOrThrowWith(ScheduleSchema.durationMillis(value), () => new Error(`Invalid duration: ${value}`)) + Option.getOrThrowWith(durationMillis(value), () => new Error(`Invalid duration: ${value}`)) const OFFSET = /\d(?:z|[+-]\d{2}:?\d{2})$/i @@ -142,12 +157,34 @@ const finished = (row: Pick, next: number | undefined, remaining const optionalTime = (value: number | null) => (value === null ? undefined : DateTime.makeUnsafe(value)) -const toInfo = (row: Row): Info => ({ +const promptText = (prompt: string, message: string | undefined) => { + if (!message?.trim()) return prompt + return `${prompt}\n\n${message}` +} + +const resolveStoredCheck = (input: CreateInput): Effect.Effect => { + if (input.if !== undefined && input.until !== undefined) + return Effect.fail(new InvalidError({ reason: "Choose either if or until, not both" })) + const check: StoredCheck | undefined = + input.if !== undefined + ? { mode: "if", condition: input.if } + : input.until !== undefined + ? { mode: "until", condition: input.until } + : undefined + if (check === undefined) return Effect.succeed(undefined) + const validated = ScheduleCheck.validate(check.condition) + if (Result.isFailure(validated)) return Effect.fail(new InvalidError({ reason: validated.failure })) + return Effect.succeed(check) +} + +const toInfo = (row: Row, lastOutput?: string): Info => ({ id: row.id, sessionID: row.session_id ?? undefined, name: row.name ?? undefined, when: row.when, tz: row.tz, + if: row.check?.mode === "if" ? row.check.condition : undefined, + until: row.check?.mode === "until" ? row.check.condition : undefined, action: row.action, status: row.status, nextDueAt: optionalTime(row.next_due_at), @@ -156,13 +193,15 @@ const toInfo = (row: Row): Info => ({ runCount: row.run_count, lastFiredAt: optionalTime(row.last_fired_at), lastOutcome: row.last_outcome ?? undefined, + ...(lastOutput === undefined ? {} : { lastOutput }), time: { created: DateTime.makeUnsafe(row.time_created), updated: DateTime.makeUnsafe(row.time_updated) }, }) const layer = Layer.effect( Service, Effect.gen(function* () { - const { db } = yield* Database.Service + const database = yield* Database.Service + const db = database.db const sessions = yield* Session.Service const instances = yield* Instance.Service const execution = yield* SessionExecution.Service @@ -210,22 +249,172 @@ const layer = Layer.effect( Effect.map((row) => row !== undefined), ) - const fire = Effect.fnUntraced(function* (row: Row, sessionID: SessionSchema.ID, due: number, now: number) { - const run = row.run_count + 1 - // The id is derived from the run, so retrying after a crash reconciles with the earlier admission. + const admitWake = Effect.fnUntraced(function* ( + row: Row, + sessionID: SessionSchema.ID, + run: number, + due: number, + late: boolean, + text: string, + checkMeta?: Record, + ) { + // The id is derived from `run_count`, so retrying after a crash reconciles with the earlier admission. const id = SessionMessage.ID.make(`msg_${row.id}_${run}`) - const late = now - due >= LATE_AFTER yield* sessions.prompt({ sessionID, id, - text: row.action.prompt, + text, delivery: "steer", - metadata: { schedule: { id: row.id, name: row.name ?? undefined, run, dueAt: due, late } }, + metadata: { + schedule: { + id: row.id, + name: row.name ?? undefined, + run, + dueAt: due, + late, + ...(checkMeta === undefined ? {} : { check: checkMeta }), + }, + }, }) + return id + }) + + const fire = Effect.fnUntraced(function* (row: Row, session: SessionSchema.Info, due: number, now: number) { + const late = now - due >= LATE_AFTER const next = following(row, now) + + if (row.check !== null) { + const check = row.check + const outcome = yield* ScheduleCheck.evaluate(session, check, row.state ?? null).pipe( + instances.provide(session), + ) + + if (!outcome.ok) { + const strikes = row.failures + 1 + if (strikes < MAX_CHECK_FAILURES) { + yield* update(row, { + failures: strikes, + last_outcome: "failed", + retry_at: now + CHECK_FAILED_RETRY, + time_updated: now, + }) + return + } + const run = row.run_count + 1 + const id = yield* admitWake( + row, + session.id, + run, + due, + late, + `${row.action.prompt}\n\n[Schedule paused after ${strikes} failed checks: ${outcome.error}]`, + { error: outcome.error, failures: strikes }, + ) + yield* db + .update(ScheduleTable) + .set({ + status: "paused", + failures: strikes, + run_count: run, + last_message_id: id, + last_fired_at: now, + last_outcome: "failed", + retry_at: null, + time_updated: now, + }) + .where(eq(ScheduleTable.id, row.id)) + .run() + .pipe(Effect.orDie) + return + } + + if (!outcome.wake) { + const checks = row.checks + 1 + const ended = next === undefined || (row.ends_at !== null && next > row.ends_at) + if (check.mode === "until" && ended) { + const run = row.run_count + 1 + const id = yield* admitWake( + row, + session.id, + run, + due, + late, + `${row.action.prompt}\n\n[Schedule timed out without matching: ${ScheduleCheck.describe(check.condition)}]`, + { expired: true }, + ) + yield* db + .update(ScheduleTable) + .set({ + status: "done", + checks, + failures: 0, + state: outcome.state, + run_count: run, + last_message_id: id, + last_fired_at: now, + last_outcome: "expired", + next_due_at: null, + remaining_runs: 0, + retry_at: null, + time_updated: now, + }) + .where(eq(ScheduleTable.id, row.id)) + .run() + .pipe(Effect.orDie) + return + } + yield* update(row, { + checks, + failures: 0, + state: outcome.state, + last_outcome: "checked", + retry_at: null, + ...(ended ? { status: "done" as const, next_due_at: null } : { next_due_at: next }), + time_updated: now, + }) + return + } + + const run = row.run_count + 1 + const id = yield* admitWake( + row, + session.id, + run, + due, + late, + promptText(row.action.prompt, outcome.message), + outcome.message === undefined ? {} : { message: outcome.message }, + ) + const remaining = + check.mode === "until" ? 0 : row.remaining_runs === null ? null : row.remaining_runs - 1 + const done = check.mode === "until" || finished(row, next, remaining) + // Record the run even if a pause landed during the prompt, or the next firing would reuse this message id. + yield* db + .update(ScheduleTable) + .set({ + run_count: run, + checks: row.checks + 1, + failures: 0, + state: outcome.state, + last_message_id: id, + last_fired_at: now, + last_outcome: late ? "late" : "fired", + retry_at: null, + remaining_runs: remaining, + ...(done ? { status: "done" as const } : {}), + next_due_at: done ? null : next, + time_updated: now, + }) + .where(eq(ScheduleTable.id, row.id)) + .run() + .pipe(Effect.orDie) + return + } + + const run = row.run_count + 1 + const id = yield* admitWake(row, session.id, run, due, late, row.action.prompt) const remaining = row.remaining_runs === null ? null : row.remaining_runs - 1 const done = finished(row, next, remaining) - // Record the run even if a pause landed during the prompt, or the next firing would reuse this message id. yield* db .update(ScheduleTable) .set({ @@ -268,7 +457,7 @@ const layer = Layer.effect( }) return } - yield* fire(row, session.id, due, now) + yield* fire(row, session, due, now) }) const tickAt = Effect.fnUntraced(function* (now: number) { @@ -289,18 +478,26 @@ const layer = Layer.effect( due, (row) => step(row, now).pipe( - Effect.catchCause((cause) => - Cause.hasInterruptsOnly(cause) - ? Effect.interrupt - : // Back off rather than retrying a failing prompt on every tick. - Effect.logWarning("schedule failed to fire", { scheduleID: row.id, cause }).pipe( - Effect.andThen( - update(row, { last_outcome: "failed", retry_at: now + FAILED_RETRY, time_updated: now }), - ), - ), - ), + Effect.catchCause((cause) => { + if (Cause.hasInterruptsOnly(cause)) return Effect.interrupt + if (Cause.squash(cause) instanceof FileSystem.DirectoryNotFoundError) { + // The Session's folder (e.g. a removed worktree) is gone; pause rather than retrying every 5 minutes. + return update(row, { + status: "paused", + last_outcome: "failed", + retry_at: null, + time_updated: now, + }) + } + // Back off rather than retrying a failing prompt on every tick. + return Effect.logWarning("schedule failed to fire", { scheduleID: row.id, cause }).pipe( + Effect.andThen( + update(row, { last_outcome: "failed", retry_at: now + FAILED_RETRY, time_updated: now }), + ), + ) + }), ), - { discard: true }, + { discard: true, concurrency: due.length > 1 ? TICK_CONCURRENCY : 1 }, ) }, Effect.withSpan("Schedule.tick")) @@ -356,7 +553,12 @@ const layer = Layer.effect( }) } - const firstDue = Effect.fnUntraced(function* (when: ScheduleSchema.When, zone: DateTime.TimeZone, now: number) { + const firstDue = Effect.fnUntraced(function* ( + when: When, + hasCheck: boolean, + zone: DateTime.TimeZone, + now: number, + ) { switch (when.type) { case "in": return now + (yield* duration(when.in)) @@ -368,7 +570,12 @@ const layer = Layer.effect( } case "every": { const every = yield* duration(when.every) - if (every < MIN_INTERVAL) return yield* new InvalidError({ reason: "every must be at least 1 minute" }) + const minInterval = hasCheck ? MIN_CHECK_INTERVAL : MIN_INTERVAL + if (every < minInterval) { + return yield* new InvalidError({ + reason: hasCheck ? "every must be at least 15 seconds" : "every must be at least 1 minute", + }) + } return now + every } case "cron": { @@ -384,12 +591,16 @@ const layer = Layer.effect( }) const endsAt = Effect.fnUntraced(function* ( - ends: ScheduleSchema.Ends | undefined, + ends: Ends | undefined, + isUntil: boolean, zone: DateTime.TimeZone, now: number, ) { const after = ends?.after === undefined ? undefined : now + (yield* duration(ends.after)) - if (ends?.at === undefined) return after ?? null + if (ends?.at === undefined) { + if (after !== undefined) return after + return isUntil ? now + DEFAULT_UNTIL_AFTER : null + } const at = parseAt(ends.at, zone) if (Option.isNone(at)) return yield* new InvalidError({ reason: `Invalid time: ${ends.at}` }) return Math.min(at.value, after ?? at.value) @@ -398,28 +609,101 @@ const layer = Layer.effect( return Service.of({ create: Effect.fn("Schedule.create")(function* (input) { const session = yield* sessions.get(input.sessionID) - const config = yield* settings(session).pipe(Effect.orDie) + const config = yield* settings(session).pipe( + Effect.catchTag("FileSystem.DirectoryNotFoundError", () => + Effect.fail(new InvalidError({ reason: `Session directory does not exist: ${session.location.directory}` })), + ), + ) if (!config.enabled) return yield* new DisabledError() - const zone = yield* resolveZone(input.when.tz ?? config.timezone) + const check = yield* resolveStoredCheck(input) + const when = input.when ?? (check?.mode === "until" ? DEFAULT_UNTIL_EVERY : undefined) + if (when === undefined) + return yield* new InvalidError({ reason: "when is required unless until is set" }) + const zone = yield* resolveZone(when.tz ?? config.timezone) const now = yield* Clock.currentTimeMillis - const due = yield* firstDue(input.when, zone, now) + const due = yield* firstDue(when, check !== undefined, zone, now) if (input.ends?.times !== undefined && input.ends.times < 1) return yield* new InvalidError({ reason: "ends.times must be at least 1" }) - const ends = yield* endsAt(input.ends, zone, now) + const ends = yield* endsAt(input.ends, check?.mode === "until", zone, now) if (ends !== null && due > ends) return yield* new InvalidError({ reason: "The schedule ends before it first fires" }) + + const seeded = + check === undefined + ? { matched: false as const, state: null as Schema.Json, checks: 0 } + : yield* Effect.gen(function* () { + yield* ScheduleCheck.authorize(session, input.name, check).pipe( + instances.provide(session), + Effect.mapError((error) => (typeof error === "string" ? new InvalidError({ reason: error }) : error)), + Effect.catchTag("FileSystem.DirectoryNotFoundError", () => + Effect.fail( + new InvalidError({ reason: `Session directory does not exist: ${session.location.directory}` }), + ), + ), + ) + const shouldSeed = + check.mode === "until" || (check.condition.type === "shell" && check.condition.changes === true) + if (!shouldSeed) return { matched: false as const, state: null as Schema.Json, checks: 0 } + const init = yield* ScheduleCheck.evaluate(session, check, null).pipe( + instances.provide(session), + Effect.catchTag("FileSystem.DirectoryNotFoundError", () => + Effect.fail( + new InvalidError({ reason: `Session directory does not exist: ${session.location.directory}` }), + ), + ), + ) + // A target that is not up yet (e.g. curl exiting 7 while a server boots) is "not yet", not an invalid schedule. + if (!init.ok) return { matched: false as const, state: null as Schema.Json, checks: 0 } + if (check.mode === "until" && init.wake) + return { matched: true as const, state: init.state, message: init.message } + return { matched: false as const, state: init.state, checks: 1 } + }) + + if (seeded.matched) { + const row = yield* db + .insert(ScheduleTable) + .values({ + id: ID.create(), + session_id: session.id, + name: input.name ?? null, + check: check ?? null, + state: seeded.state, + checks: 1, + action: input.action, + when, + tz: DateTime.zoneToString(zone), + status: "done", + next_due_at: null, + ends_at: ends, + remaining_runs: 0, + run_count: 1, + last_fired_at: now, + last_outcome: "fired", + time_created: now, + time_updated: now, + }) + .returning() + .get() + .pipe(Effect.orDie) + return toInfo(row, seeded.message) + } + + const remaining = check?.mode === "until" ? 1 : (input.ends?.times ?? null) const row = yield* db .insert(ScheduleTable) .values({ id: ID.create(), session_id: session.id, name: input.name ?? null, + check: check ?? null, + state: seeded.state, + checks: seeded.checks, action: input.action, - when: input.when, + when, tz: DateTime.zoneToString(zone), next_due_at: due, ends_at: ends, - remaining_runs: input.ends?.times ?? null, + remaining_runs: remaining, time_created: now, time_updated: now, }) @@ -437,7 +721,7 @@ const layer = Layer.effect( .orderBy(asc(ScheduleTable.time_created), asc(ScheduleTable.id)) .all() .pipe(Effect.orDie) - return rows.map(toInfo) + return rows.map((row) => toInfo(row)) }), get: Effect.fn("Schedule.get")(function* (id) { return toInfo(yield* find(id)) @@ -459,7 +743,13 @@ const layer = Layer.effect( const done = next === null || finished(row, next, row.remaining_runs) yield* db .update(ScheduleTable) - .set({ status: done ? "done" : "active", next_due_at: done ? null : next, retry_at: null, time_updated: now }) + .set({ + status: done ? "done" : "active", + next_due_at: done ? null : next, + failures: 0, + retry_at: null, + time_updated: now, + }) .where(and(eq(ScheduleTable.id, id), eq(ScheduleTable.status, "paused"))) .run() .pipe(Effect.orDie) @@ -471,7 +761,11 @@ const layer = Layer.effect( yield* db.delete(ScheduleTable).where(eq(ScheduleTable.id, id)).run().pipe(Effect.orDie) }), enabled: Effect.fn("Schedule.enabled")(function* (sessionID) { - return (yield* settings(yield* sessions.get(sessionID)).pipe(Effect.orDie)).enabled + const session = yield* sessions.get(sessionID) + return yield* settings(session).pipe( + Effect.map((value) => value.enabled), + Effect.catchTag("FileSystem.DirectoryNotFoundError", () => Effect.succeed(false)), + ) }), tick, run, diff --git a/packages/core/src/schedule/check.ts b/packages/core/src/schedule/check.ts new file mode 100644 index 000000000000..0ebf78a3b103 --- /dev/null +++ b/packages/core/src/schedule/check.ts @@ -0,0 +1,265 @@ +export * as ScheduleCheck from "./check.js" + +import { isRecord } from "@opencode/util/record" +import { collectStream } from "@opencode/util/process" +import type { Schedule as ScheduleSchema } from "@opencode/schema/schedule" +import { Array, Duration, Effect, Option, Result, type Schema } from "effect" +import { ChildProcess } from "effect/process" +import { Agent } from "../agent.js" +import { Environment } from "../environment/index.js" +import { FileAccess } from "../file-access.js" +import { Location } from "../location.js" +import { Permission } from "../permission.js" +import { Plugin } from "../plugin.js" +import type { SessionSchema } from "../session/schema.js" +import { ShellParse } from "../shell/parse.js" +import { ShellSelect } from "../shell/select.js" + +export const TIMEOUT_MS = 30_000 +const MAX_OUTPUT_BYTES = 32 * 1024 + +export type Outcome = + | { + readonly ok: true + readonly wake: boolean + readonly state: Schema.Json + readonly message?: string + } + | { + readonly ok: false + readonly error: string + } + +/** Validates condition syntax synchronously before asking for permissions or spawning anything. */ +export function validate(condition: ScheduleSchema.Condition): Result.Result { + switch (condition.type) { + case "shell": { + if (condition.run.trim() === "") return Result.fail("A shell check needs a command") + const predicates = [ + condition.is !== undefined, + condition.matches !== undefined, + condition.changes !== undefined, + ].filter(Boolean).length + if (predicates > 1) return Result.fail("Choose at most one of is, matches or changes") + if (condition.matches === undefined) return Result.succeed(undefined) + return Result.try({ + try: () => { + new RegExp(condition.matches!) + }, + catch: (cause) => + `Invalid regular expression ${JSON.stringify(condition.matches)}: ${cause instanceof Error ? cause.message : String(cause)}`, + }) + } + } +} + +export function describe(condition: ScheduleSchema.Condition): string { + switch (condition.type) { + case "shell": + return condition.run + } +} + +/** Runs in the Session's Location while the user is present; approves required permissions once without saving. */ +export const authorize = Effect.fnUntraced(function* ( + session: SessionSchema.Info, + name: string | undefined, + check: ScheduleSchema.StoredCheck, +) { + const plugins = yield* Plugin.Service + yield* plugins.awaitActivation + switch (check.condition.type) { + case "shell": + return yield* authorizeShell(session, name, check.mode, check.condition) + } +}) + +/** Runs one check in the Session's Location under narrow unattended permissions and a 30s wall-clock cap. */ +export const evaluate = Effect.fnUntraced(function* ( + session: SessionSchema.Info, + check: ScheduleSchema.StoredCheck, + prevState: Schema.Json, +) { + const run = Effect.gen(function* () { + const plugins = yield* Plugin.Service + yield* plugins.awaitActivation + switch (check.condition.type) { + case "shell": + return yield* runShell(session, check.condition, prevState) + } + }) + return yield* run.pipe( + Effect.timeoutOption(TIMEOUT_MS), + Effect.map((result): Outcome => + Option.match(result, { + onNone: () => ({ ok: false, error: `Check timed out after ${TIMEOUT_MS}ms` }), + onSome: (outcome) => outcome, + }), + ), + ) +}) + +const scanShell = Effect.fnUntraced(function* (command: string) { + const location = yield* Location.Service + const access = yield* FileAccess.Service + const shellSelect = yield* ShellSelect.Service + const target = yield* access + .resolve({ path: location.directory, kind: "directory" }) + .pipe(Effect.mapError((error) => error.message)) + const shell = yield* shellSelect.resolve({ priority: "compat" }) + const parsed = yield* ShellParse.scanPortable(command, shell, target.absolute).pipe( + Effect.mapError((error) => error.message), + ) + const directories = yield* Effect.forEach(parsed.directories, (directory) => + access + .resolve({ path: FileAccess.resolvePath(target.absolute, directory), kind: "directory" }) + .pipe(Effect.mapError((error) => error.message)), + ) + const external = Array.dedupe( + [target, ...directories].flatMap((item) => (item.externalDirectory ? [item.externalDirectory.resource] : [])), + ) + const commands = Array.dedupe(parsed.commands.map((item) => item.resource)) + return { shell, cwd: target.absolute, workspaceID: location.workspaceID, external, commands } +}) + +const assertShellPermissions = Effect.fnUntraced(function* ( + session: SessionSchema.Info, + scanned: { readonly external: ReadonlyArray; readonly commands: ReadonlyArray }, + metadata?: Record, +) { + const permission = yield* Permission.Service + const agents = yield* Agent.Service + const agent = yield* agents.select(session.agent) + if (scanned.external.length > 0) { + yield* permission.assert({ + action: "external_directory", + resources: [...scanned.external], + ...(metadata === undefined ? {} : { metadata }), + sessionID: session.id, + agent: agent.id, + }) + } + if (scanned.commands.length > 0) { + yield* permission.assert({ + action: "shell", + resources: [...scanned.commands], + ...(metadata === undefined ? {} : { metadata }), + sessionID: session.id, + agent: agent.id, + }) + } +}) + +const authorizeShell = Effect.fnUntraced(function* ( + session: SessionSchema.Info, + name: string | undefined, + mode: ScheduleSchema.CheckMode, + condition: ScheduleSchema.ShellCondition, +) { + const scanned = yield* scanShell(condition.run) + yield* assertShellPermissions(session, scanned, { + schedule: name, + mode, + run: condition.run, + }) +}) + +const runShell = Effect.fnUntraced(function* ( + session: SessionSchema.Info, + condition: ScheduleSchema.ShellCondition, + prevState: Schema.Json, +) { + const environment = yield* Environment.Service + const scannedExit = yield* Effect.result(scanShell(condition.run)) + if (Result.isFailure(scannedExit)) return { ok: false, error: scannedExit.failure } satisfies Outcome + const scanned = scannedExit.success + const unattended: Permission.Ruleset = [ + ...scanned.external.map((resource) => ({ action: "external_directory", resource, effect: "allow" as const })), + ...scanned.commands.map((resource) => ({ action: "shell", resource, effect: "allow" as const })), + ] + const authorized = yield* assertShellPermissions(session, scanned).pipe( + Effect.provideService(Permission.Unattended, unattended), + Effect.result, + ) + if (Result.isFailure(authorized)) return { ok: false as const, error: authorized.failure.message } + + const executed = yield* Effect.scoped( + Effect.gen(function* () { + const proc = ChildProcess.make(scanned.shell, ShellSelect.args(scanned.shell, condition.run), { + cwd: scanned.cwd, + ...(scanned.workspaceID === undefined ? { env: process.env } : {}), + stdin: "ignore", + detached: process.platform !== "win32", + forceKillAfter: Duration.seconds(3), + }) + const handle = yield* environment.spawner.spawn(proc) + const [stdout, stderr, exitCode] = yield* Effect.all( + [ + collectStream(handle.stdout, MAX_OUTPUT_BYTES), + collectStream(handle.stderr, MAX_OUTPUT_BYTES), + handle.exitCode, + ], + { concurrency: "unbounded" }, + ) + return { + exitCode, + stdout: utf8Decode(stdout.buffer), + stderr: utf8Decode(stderr.buffer), + } + }), + ).pipe(Effect.result) + + if (Result.isFailure(executed)) { + return { + ok: false as const, + error: executed.failure instanceof Error ? executed.failure.message : String(executed.failure), + } + } + + const output = executed.success + const trimmed = output.stdout.trim() + const message = trimmed === "" ? undefined : trimmed + const hasPredicate = + condition.is !== undefined || condition.matches !== undefined || condition.changes === true + + if (!hasPredicate) { + return { + ok: true as const, + wake: output.exitCode === 0, + state: null, + ...(output.exitCode === 0 && message !== undefined ? { message } : {}), + } + } + + if (output.exitCode !== 0) { + const detail = output.stderr.trim() || trimmed + return { + ok: false as const, + error: `Command exited with code ${output.exitCode}${detail ? `: ${detail}` : ""}`, + } + } + + if (condition.is !== undefined) { + const wake = trimmed === condition.is.trim() + return { ok: true as const, wake, state: null, ...(wake && message !== undefined ? { message } : {}) } + } + + if (condition.matches !== undefined) { + const wake = new RegExp(condition.matches).test(output.stdout) + return { ok: true as const, wake, state: null, ...(wake && message !== undefined ? { message } : {}) } + } + + const prevStdout = isRecord(prevState) && typeof prevState.stdout === "string" ? prevState.stdout : null + const wake = prevStdout !== null && output.stdout !== prevStdout + return { + ok: true as const, + wake, + state: { stdout: output.stdout }, + ...(wake && message !== undefined ? { message } : {}), + } +}) + +function utf8Decode(bytes: Uint8Array): string { + const text = new TextDecoder("utf-8").decode(bytes) + return text.endsWith("\uFFFD") ? text.slice(0, -1) : text +} diff --git a/packages/core/src/schedule/sql.ts b/packages/core/src/schedule/sql.ts index bd24fbb968fd..20d839f09e2b 100644 --- a/packages/core/src/schedule/sql.ts +++ b/packages/core/src/schedule/sql.ts @@ -1,5 +1,6 @@ import { sql } from "drizzle-orm" import { index, integer, sqliteTable, text } from "drizzle-orm/sqlite-core" +import type { Schema } from "effect" import type { Schedule } from "@opencode/schema/schedule" import { Timestamps } from "../database/schema.sql.js" import type { SessionMessage } from "../session/message.js" @@ -16,6 +17,7 @@ export const ScheduleTable = sqliteTable( .references(() => SessionTable.id, { onDelete: "cascade" }), name: text(), action: text({ mode: "json" }).$type().notNull(), + check: text({ mode: "json" }).$type(), when: text({ mode: "json" }).$type().notNull(), tz: text().notNull(), status: text().$type().notNull().default("active"), @@ -23,11 +25,15 @@ export const ScheduleTable = sqliteTable( ends_at: integer(), remaining_runs: integer(), run_count: integer().notNull().default(0), + checks: integer().notNull().default(0), + failures: integer().notNull().default(0), last_fired_at: integer(), last_message_id: text().$type(), last_outcome: text().$type(), /** The loop leaves a due row alone until then, e.g. while its location has schedules turned off. */ retry_at: integer(), + /** Condition state persisted across checks (e.g. previous stdout for shell `changes`). */ + state: text({ mode: "json" }).$type(), ...Timestamps, }, (table) => [ diff --git a/packages/core/src/tool/plugin/schedule.ts b/packages/core/src/tool/plugin/schedule.ts index 89c5a8780fc5..e8ef41a373b9 100644 --- a/packages/core/src/tool/plugin/schedule.ts +++ b/packages/core/src/tool/plugin/schedule.ts @@ -16,6 +16,8 @@ export const View = Schema.Struct({ name: Schema.String.pipe(optional), when: Schedule.When, tz: Schema.String, + if: Schedule.Condition.pipe(optional), + until: Schedule.Condition.pipe(optional), prompt: Schema.String, status: Schedule.Status, nextDueAt: Schema.String.pipe(optional), @@ -24,6 +26,7 @@ export const View = Schema.Struct({ runCount: Schema.Number, lastFiredAt: Schema.String.pipe(optional), lastOutcome: Schedule.Outcome.pipe(optional), + lastOutput: Schema.String.pipe(optional), }) export type View = typeof View.Type @@ -73,7 +76,7 @@ export const Plugin = { editor.add({ name: "schedule_create", description: - "Schedule a prompt that wakes this session later: once after a delay, at a time, every interval, or on a cron. Returns the schedule with its next run.", + "Schedule a prompt that wakes this session later: once after a delay, at a time, every interval, on a cron, or gated by an `if` / `until` shell check. Returns the schedule with its next run (or `done` immediately if `until` already matches).", input: Schedule.Request, output: View, options: { namespace: "opencode", codemode: true }, @@ -81,7 +84,9 @@ export const Plugin = { schedules.create(Schedule.toCreateInput(context.sessionID, input)).pipe( Effect.map((info) => { const view = toView(info) - return { output: view, content: `Scheduled: ${summary(view)}` } + const prefix = + view.status === "done" && view.until !== undefined ? "Matched immediately" : "Scheduled" + return { output: view, content: `${prefix}: ${summary(view)}` } }), Effect.mapError(failure), ), @@ -165,6 +170,13 @@ export const Plugin = { }), } +function formatCondition(condition: Schedule.Condition): string { + switch (condition.type) { + case "shell": + return JSON.stringify(condition.run) + } +} + /** One line per schedule, for models that read tool content rather than output (Code Mode off). */ function summary(view: View) { const when = view.when @@ -176,12 +188,20 @@ function summary(view: View) { : when.type === "every" ? `every ${when.every}` : `cron "${when.cron}"` + const gate = + view.until !== undefined + ? `until ${formatCondition(view.until)}` + : view.if !== undefined + ? `if ${formatCondition(view.if)}` + : undefined const prompt = view.prompt.length > 80 ? `${view.prompt.slice(0, 79)}…` : view.prompt return [ `${view.id}${view.name ? ` "${view.name}"` : ""}: ${trigger} (${view.tz})`, + gate, view.status, view.nextDueAt && `next ${view.nextDueAt}`, `prompt: ${JSON.stringify(prompt)}`, + view.lastOutput && `output: ${JSON.stringify(view.lastOutput)}`, ] .filter(Boolean) .join(", ") @@ -200,6 +220,8 @@ function toView(info: Schedule.Info): View { name: info.name, when: info.when, tz: info.tz, + if: info.if, + until: info.until, prompt: info.action.prompt, status: info.status, nextDueAt: local(info.nextDueAt), @@ -208,5 +230,6 @@ function toView(info: Schedule.Info): View { runCount: info.runCount, lastFiredAt: local(info.lastFiredAt), lastOutcome: info.lastOutcome, + lastOutput: info.lastOutput, } } diff --git a/packages/core/test/schedule.test.ts b/packages/core/test/schedule.test.ts index 09515e859b94..7314928036b9 100644 --- a/packages/core/test/schedule.test.ts +++ b/packages/core/test/schedule.test.ts @@ -1,14 +1,22 @@ import { describe, expect } from "bun:test" -import type { FileSystem } from "@opencode/core/filesystem" -import { DateTime, Effect, Fiber, Layer, LayerMap, Stream } from "effect" +import { FileSystem } from "@opencode/core/filesystem" +import { DateTime, Effect, Fiber, Latch, Layer, LayerMap, Queue, Sink, Stream } from "effect" +import { ChildProcessSpawner } from "effect/process" +import { ExitCode, makeHandle, ProcessId } from "effect/process/ChildProcessSpawner" import { TestClock } from "effect/testing" import { eq } from "drizzle-orm" +import { Agent } from "@opencode/core/agent" import { Database } from "@opencode/core/database/database" import { AppNodeBuilder } from "@opencode/core/effect/app-node-builder" +import { Environment } from "@opencode/core/environment/index" +import { FileAccess } from "@opencode/core/file-access" +import { Instance } from "@opencode/core/instance/service" import { LayerNode } from "@opencode/util/effect/layer-node" import { makeGlobalNode } from "@opencode/util/effect/app-node" import { Bus } from "@opencode/core/bus" -import { Location } from "@opencode/schema/location" +import { Location } from "@opencode/core/location" +import { Permission } from "@opencode/core/permission" +import { PermissionSaved } from "@opencode/core/permission/saved" import { Project } from "@opencode/core/project" import { ProjectTable } from "@opencode/core/project/sql" import { AbsolutePath } from "@opencode/core/schema" @@ -21,6 +29,7 @@ import { SessionExecution } from "@opencode/core/session/execution" import { SessionInbox } from "@opencode/core/session/inbox" import { SessionInboxTable, SessionTable } from "@opencode/core/session/sql" import { SessionStore } from "@opencode/core/session/store" +import { ShellSelect } from "@opencode/core/shell/select" import { LocationServiceMap } from "@opencode/core/location-service-map" import type { LocationServices } from "@opencode/core/location-services" import { Plugin } from "@opencode/core/plugin" @@ -30,11 +39,13 @@ import { Skill } from "@opencode/core/skill" import { Config } from "@opencode/core/config" import { Document, Info } from "@opencode/schema/config" import { ConfigExperimental } from "@opencode/schema/config/experimental" +import { location } from "./fixture/location" import { testEffect } from "./lib/effect" const enabledDirectory = "/enabled" const disabledDirectory = "/disabled" const toggleDirectory = "/toggle" +const missingDirectory = "/missing" const wakes: Session.ID[] = [] /** Runs inside the next wake, which happens while a schedule is firing. */ @@ -79,31 +90,141 @@ const toggled = Layer.succeed( changes: () => Stream.empty, }), ) +/** Fake shell state and permission rules reset by `setup`. */ +const shellState = { + exitCode: 0, + stdout: "", + stderr: "", + hangMs: 0, + runs: [] as string[], + missingNow: false, +} +let agentRules: Permission.Ruleset = [{ action: "*", resource: "*", effect: "allow" }] +let activationGate: Effect.Effect = Effect.void + +const testEnvironment = Layer.effect( + Environment.Service, + Effect.gen(function* () { + const memory = Environment.makeMemoryDriver() + const files = Environment.makeFiles(memory) + yield* files.mkdir(enabledDirectory).pipe(Effect.orDie) + yield* files.mkdir(toggleDirectory).pipe(Effect.orDie) + const spawner = ChildProcessSpawner.make((command) => + Effect.gen(function* () { + const script = command._tag === "StandardCommand" ? (command.args.at(-1) ?? command.command) : "pipeline" + shellState.runs.push(script) + if (shellState.hangMs > 0) yield* Effect.sleep(shellState.hangMs) + const outBytes = new TextEncoder().encode(shellState.stdout) + const errBytes = new TextEncoder().encode(shellState.stderr) + return makeHandle({ + pid: ProcessId(1), + exitCode: Effect.succeed(ExitCode(shellState.exitCode)), + isRunning: Effect.succeed(false), + kill: () => Effect.void, + stdin: Sink.drain, + stdout: outBytes.byteLength > 0 ? Stream.make(outBytes) : Stream.empty, + stderr: errBytes.byteLength > 0 ? Stream.make(errBytes) : Stream.empty, + all: outBytes.byteLength > 0 ? Stream.make(outBytes) : Stream.empty, + getInputFd: () => Sink.drain, + getOutputFd: () => Stream.empty, + unref: Effect.succeed(Effect.void), + }) + }), + ) + return Environment.Service.of({ files, spawner }) + }), +) + +const configureAgent = Layer.effectDiscard( + Effect.gen(function* () { + const agents = yield* Agent.Service + yield* agents.transform((editor) => + editor.update(Agent.defaultID, (agent) => { + agent.permissions = [...agentRules] + }), + ) + }), +) + const locations = makeGlobalNode({ service: LocationServiceMap.Service, layer: Layer.effect( LocationServiceMap.Service, Effect.gen(function* () { const bus = yield* Bus.Service + const store = yield* SessionStore.Service + const saved = yield* PermissionSaved.Service return yield* LayerMap.make( - (ref: Location.Ref) => + (ref: Location.Ref): Layer.Layer => { + if (ref.directory === missingDirectory && shellState.missingNow) { + return Layer.effectDiscard( + Effect.fail( + new FileSystem.DirectoryNotFoundError({ directory: ref.directory, cause: new Error("missing") }), + ), + ) as unknown as Layer.Layer + } // oxlint-disable-next-line typescript-eslint/no-unsafe-type-assertion - Layer.mergeAll( - ref.directory === enabledDirectory ? enabled : ref.directory === toggleDirectory ? toggled : Layer.empty, - LayerNode.compile(LayerNode.group([PluginHooks.node, Skill.node]), { - replacements: [Bus.node.replace(Layer.succeed(Bus.Service, bus))], - }), + return Layer.mergeAll( + ref.directory === enabledDirectory || ref.directory === missingDirectory + ? enabled + : ref.directory === toggleDirectory + ? toggled + : Layer.empty, + configureAgent.pipe( + Layer.provideMerge( + LayerNode.compile( + LayerNode.group([ + Location.node, + Environment.node, + PluginHooks.node, + Skill.node, + Agent.node, + Permission.node, + FileAccess.node, + ]), + { + replacements: [ + Bus.node.replace(Layer.succeed(Bus.Service, bus)), + SessionStore.node.replace(Layer.succeed(SessionStore.Service, store)), + PermissionSaved.node.replace(Layer.succeed(PermissionSaved.Service, saved)), + Location.node.replace(Layer.succeed(Location.Service, location(ref))), + Environment.node.replace(testEnvironment), + ], + }, + ), + ), + ), + Layer.succeed( + ShellSelect.Service, + ShellSelect.Service.of({ + transform: () => Effect.die("unused"), + reload: () => Effect.void, + resolve: () => Effect.succeed("/bin/bash"), + }), + ), Layer.mock(Snapshot.Service, { capture: () => Effect.undefined, restore: () => Effect.void }), - Layer.mock(Plugin.Service, { awaitActivation: Effect.void }), - ).pipe(Layer.fresh) as unknown as Layer.Layer, + Layer.succeed(Plugin.Service, { + awaitActivation: Effect.suspend(() => activationGate), + } as unknown as Plugin.Interface), + ).pipe(Layer.fresh) as unknown as Layer.Layer + }, ) }), ), - deps: [Bus.node], + deps: [Bus.node, SessionStore.node, PermissionSaved.node], }) const it = testEffect( AppNodeBuilder.build( - LayerNode.group([Database.node, Bus.node, SessionProjector.node, SessionStore.node, Session.node, Schedule.node]), + LayerNode.group([ + Database.node, + Bus.node, + SessionProjector.node, + SessionStore.node, + Session.node, + Schedule.node, + Instance.node, + LocationServiceMap.node, + ]), [ Bus.node.replace(Bus.configured({ persist: true })), SessionExecution.node.replace(execution), @@ -132,6 +253,14 @@ const setup = (directory = enabledDirectory) => duringWake = Effect.void toggle.on = false toggle.reads = 0 + shellState.exitCode = 0 + shellState.stdout = "" + shellState.stderr = "" + shellState.hangMs = 0 + shellState.runs = [] + shellState.missingNow = false + agentRules = [{ action: "*", resource: "*", effect: "allow" }] + activationGate = Effect.void return yield* Schedule.Service }) @@ -690,4 +819,344 @@ describe("Schedule", () => { expect(yield* schedules.list({ sessionID })).toEqual([]) }), ) + + describe("shell checks", () => { + it.effect("returns done immediately without queuing a prompt when an until check already passes at creation", () => + Effect.gen(function* () { + const schedules = yield* setup() + shellState.exitCode = 0 + shellState.stdout = "all green\n" + const schedule = yield* schedules.create({ + sessionID, + until: { type: "shell", run: "gh pr checks 123" }, + action: { type: "prompt", prompt: "CI passed." }, + }) + expect(schedule).toMatchObject({ + status: "done", + runCount: 1, + lastOutcome: "fired", + lastOutput: "all green", + nextDueAt: undefined, + when: { type: "every", every: "30 seconds" }, + }) + expect(yield* pending).toEqual([]) + expect(wakes).toEqual([]) + }), + ) + + it.effect("polls until exit 0 with default 30s interval and 24h deadline, then wakes once and stops", () => + Effect.gen(function* () { + const schedules = yield* setup() + shellState.exitCode = 1 + const schedule = yield* schedules.create({ + sessionID, + until: { type: "shell", run: "gh pr checks 123" }, + action: { type: "prompt", prompt: "CI passed." }, + }) + expect(schedule).toMatchObject({ + status: "active", + when: { type: "every", every: "30 seconds" }, + remainingRuns: 1, + }) + expect(iso(schedule.endsAt)).toBe(iso(DateTime.makeUnsafe(24 * 60 * 60_000))) + + yield* TestClock.adjust("30 seconds") + yield* schedules.tick + expect(yield* pending).toEqual([]) + expect(yield* schedules.get(schedule.id)).toMatchObject({ + status: "active", + runCount: 0, + remainingRuns: 1, + lastOutcome: "checked", + }) + + shellState.exitCode = 0 + shellState.stdout = "20 pass\n" + yield* TestClock.adjust("30 seconds") + yield* schedules.tick + expect(yield* pending).toEqual([messageID(schedule, 1)]) + const { db } = yield* Database.Service + expect(yield* SessionInbox.find(db, messageID(schedule, 1))).toMatchObject({ + payload: { + text: "CI passed.\n\n20 pass", + metadata: { schedule: { id: schedule.id, run: 1, check: { message: "20 pass" } } }, + }, + }) + expect(yield* schedules.get(schedule.id)).toMatchObject({ + status: "done", + runCount: 1, + remainingRuns: 0, + lastOutcome: "fired", + nextDueAt: undefined, + }) + }), + ) + + it.effect("wakes once with expired when an until check hits its deadline without matching", () => + Effect.gen(function* () { + const schedules = yield* setup() + shellState.exitCode = 1 + const schedule = yield* schedules.create({ + sessionID, + when: { type: "every", every: "30 seconds" }, + until: { type: "shell", run: "curl -sf http://localhost:3000" }, + ends: { after: "1 minute" }, + action: { type: "prompt", prompt: "Server check finished." }, + }) + + yield* TestClock.adjust("30 seconds") + yield* schedules.tick + expect(yield* pending).toEqual([]) + + yield* TestClock.adjust("30 seconds") + yield* schedules.tick + expect(yield* pending).toEqual([messageID(schedule, 1)]) + const { db } = yield* Database.Service + expect(yield* SessionInbox.find(db, messageID(schedule, 1))).toMatchObject({ + payload: { + text: expect.stringContaining("[Schedule timed out without matching: curl -sf http://localhost:3000]"), + metadata: { schedule: { id: schedule.id, run: 1, check: { expired: true } } }, + }, + }) + expect(yield* schedules.get(schedule.id)).toMatchObject({ + status: "done", + runCount: 1, + lastOutcome: "expired", + nextDueAt: undefined, + }) + }), + ) + + it.effect("seeds changes baseline at t=0 and counts ends.times by wakes rather than quiet checks", () => + Effect.gen(function* () { + const schedules = yield* setup() + shellState.stdout = "sha-1\n" + const schedule = yield* schedules.create({ + sessionID, + when: { type: "every", every: "15 seconds" }, + if: { type: "shell", run: "git rev-parse HEAD", changes: true }, + ends: { times: 2 }, + action: { type: "prompt", prompt: "HEAD moved." }, + }) + + // Quiet check: same output as t=0 baseline -> remainingRuns stays 2. + yield* TestClock.adjust("15 seconds") + yield* schedules.tick + expect(yield* pending).toEqual([]) + expect(yield* schedules.get(schedule.id)).toMatchObject({ + status: "active", + runCount: 0, + remainingRuns: 2, + lastOutcome: "checked", + }) + + // Change before second tick: wakes once -> remainingRuns becomes 1. + shellState.stdout = "sha-2\n" + yield* TestClock.adjust("15 seconds") + yield* schedules.tick + expect(yield* pending).toEqual([messageID(schedule, 1)]) + expect(yield* schedules.get(schedule.id)).toMatchObject({ + status: "active", + runCount: 1, + remainingRuns: 1, + lastOutcome: "fired", + }) + yield* deliver + + // Another quiet check at sha-2 -> remainingRuns stays 1. + yield* TestClock.adjust("15 seconds") + yield* schedules.tick + expect(yield* pending).toEqual([]) + expect(yield* schedules.get(schedule.id)).toMatchObject({ status: "active", remainingRuns: 1 }) + + // Second change -> wakes a second time and finishes. + shellState.stdout = "sha-3\n" + yield* TestClock.adjust("15 seconds") + yield* schedules.tick + expect(yield* pending).toEqual([messageID(schedule, 2)]) + expect(yield* schedules.get(schedule.id)).toMatchObject({ + status: "done", + runCount: 2, + remainingRuns: 0, + lastOutcome: "fired", + }) + }), + ) + + it.effect("supports is and matches predicates, even when the command exits non-zero at creation", () => + Effect.gen(function* () { + const schedules = yield* setup() + // Service is still booting at creation (curl exits 7); creating an `until` with `is` still succeeds. + shellState.exitCode = 7 + shellState.stdout = "" + const exact = yield* schedules.create({ + sessionID, + when: { type: "every", every: "20 seconds" }, + until: { type: "shell", run: "curl -sf http://localhost:3000/status", is: "MERGED" }, + action: { type: "prompt", prompt: "Merged." }, + }) + shellState.exitCode = 0 + shellState.stdout = "OPEN\n" + const regex = yield* schedules.create({ + sessionID, + when: { type: "every", every: "20 seconds" }, + until: { type: "shell", run: "gh pr view 2", matches: "^MERGED" }, + action: { type: "prompt", prompt: "Regex merged." }, + }) + + yield* TestClock.adjust("20 seconds") + yield* schedules.tick + expect(yield* pending).toEqual([]) + + shellState.stdout = "MERGED\n" + yield* TestClock.adjust("20 seconds") + yield* schedules.tick + expect(yield* pending).toEqual([messageID(exact, 1), messageID(regex, 1)]) + }), + ) + + it.effect("asks once without save at creation, runs unattended on ticks, and pauses after 3 failures", () => + Effect.gen(function* () { + const schedules = yield* setup() + agentRules = [{ action: "shell", resource: "*", effect: "ask" }] + shellState.exitCode = 1 + + const bus = yield* Bus.Service + const instances = yield* Instance.Service + const sessions = yield* Session.Service + const session = yield* sessions.get(sessionID) + const asks = yield* Queue.unbounded() + const unsubscribe = yield* bus.listen((event) => + event.type === Permission.Event.Asked.type + ? Queue.offer(asks, event.data as Permission.Request).pipe(Effect.asVoid) + : Effect.void, + ) + yield* Effect.addFinalizer(() => unsubscribe) + const asked: Permission.Request[] = [] + const answerLoop = yield* Queue.take(asks).pipe( + Effect.tap((request) => Effect.sync(() => asked.push(request))), + Effect.flatMap((request) => + Permission.Service.use((permission) => permission.reply({ requestID: request.id, reply: "once" })).pipe( + instances.provide(session), + ), + ), + Effect.forever, + Effect.forkChild, + ) + + const schedule = yield* schedules.create({ + sessionID, + name: "ci", + when: { type: "every", every: "1 minute" }, + if: { type: "shell", run: "gh pr checks 99" }, + action: { type: "prompt", prompt: "CI green." }, + }) + yield* Fiber.interrupt(answerLoop) + expect(asked).toHaveLength(1) + expect(asked[0]?.save).toBeUndefined() + expect(asked[0]?.resources).toEqual(["gh pr checks 99"]) + + // Unattended tick runs without asking again. + yield* TestClock.adjust("1 minute") + yield* schedules.tick + expect(asked).toHaveLength(1) + expect(yield* schedules.get(schedule.id)).toMatchObject({ lastOutcome: "checked" }) + + // Revoking shell permission causes 3 consecutive check failures -> auto-pause + error wake. + agentRules = [{ action: "shell", resource: "*", effect: "deny" }] + const locationsMap = yield* LocationServiceMap.Service + yield* locationsMap.invalidate(Location.Ref.make({ directory: AbsolutePath.make(enabledDirectory) })) + + for (const strike of [1, 2]) { + yield* TestClock.adjust("1 minute") + yield* schedules.tick + expect(yield* schedules.get(schedule.id)).toMatchObject({ + status: "active", + lastOutcome: "failed", + runCount: 0, + }) + expect(strike).toBeLessThan(3) + } + yield* TestClock.adjust("1 minute") + yield* schedules.tick + expect(yield* schedules.get(schedule.id)).toMatchObject({ + status: "paused", + lastOutcome: "failed", + runCount: 1, + }) + expect(yield* pending).toEqual([messageID(schedule, 1)]) + const { db } = yield* Database.Service + expect(yield* SessionInbox.find(db, messageID(schedule, 1))).toMatchObject({ + payload: { text: expect.stringContaining("[Schedule paused after 3 failed checks:") }, + }) + }), + ) + + it.effect("waits for Plugin.awaitActivation and auto-pauses when the Session directory disappears", () => + Effect.gen(function* () { + const schedules = yield* setup(missingDirectory) + const gate = yield* Latch.make() + activationGate = gate.await + shellState.exitCode = 1 + const creating = yield* Effect.forkChild( + schedules.create({ + sessionID, + when: { type: "every", every: "1 minute" }, + until: { type: "shell", run: "gh pr checks 1" }, + action: { type: "prompt", prompt: "done" }, + }), + ) + yield* Effect.repeat(Effect.yieldNow, { times: 5 }) + expect(shellState.runs).toEqual([]) + yield* gate.open + const schedule = yield* Fiber.join(creating) + expect(shellState.runs).toEqual(["gh pr checks 1"]) + + // Simulate the worktree directory being deleted before the next tick. + const session = yield* Session.Service.use((sessions) => sessions.get(sessionID)) + const locationsMap = yield* LocationServiceMap.Service + shellState.missingNow = true + yield* locationsMap.invalidate(session.location) + + yield* TestClock.adjust("1 minute") + yield* schedules.tick + expect(yield* schedules.get(schedule.id)).toMatchObject({ + status: "paused", + lastOutcome: "failed", + }) + }), + ) + + it.effect("validates check combinations, intervals, and regexes at creation", () => + Effect.gen(function* () { + const schedules = yield* setup() + const bad = (input: Omit) => + Effect.flip(schedules.create({ sessionID, action: { type: "prompt", prompt: "x" }, ...input })) + + expect(yield* bad({})).toMatchObject({ reason: "when is required unless until is set" }) + expect( + yield* bad({ + if: { type: "shell", run: "true" }, + until: { type: "shell", run: "true" }, + }), + ).toMatchObject({ reason: "Choose either if or until, not both" }) + expect( + yield* bad({ + when: { type: "every", every: "10 seconds" }, + until: { type: "shell", run: "true" }, + }), + ).toMatchObject({ reason: "every must be at least 15 seconds" }) + expect( + yield* bad({ + until: { type: "shell", run: "echo hi", is: "hi", matches: "hi" }, + }), + ).toMatchObject({ reason: "Choose at most one of is, matches or changes" }) + expect( + yield* bad({ + until: { type: "shell", run: "echo hi", matches: "[unclosed" }, + }), + ).toMatchObject({ _tag: "Schedule.InvalidError" }) + }), + ) + }) }) diff --git a/packages/core/test/tool-schedule.test.ts b/packages/core/test/tool-schedule.test.ts index 89a3d20331dd..d76825761c22 100644 --- a/packages/core/test/tool-schedule.test.ts +++ b/packages/core/test/tool-schedule.test.ts @@ -47,8 +47,10 @@ const memory = () => { id: Schedule.ID.make(`sch_${rows.size + 1}`), sessionID: input.sessionID, name: input.name, - when: input.when, + when: input.when ?? { type: "every", every: "30 seconds" }, tz: "America/New_York", + if: input.if, + until: input.until, action: input.action, status: "active", nextDueAt: DateTime.add(now, { minutes: 20 }), @@ -261,6 +263,32 @@ describe("ScheduleTools", () => { expect(rows.get(Schedule.ID.make("sch_1"))).toMatchObject({ status: "active", sessionID: otherSessionID }) }), ) + + it.effect("creates until schedules with optional when and formats check conditions", () => + Effect.gen(function* () { + const { registry, rows } = yield* setup(new ConfigExperimental.Info({ schedules: true })) + const created = yield* run(registry, "schedule_create", { + until: { run: "gh pr checks 59962" }, + prompt: "CI passed", + }) + expect(created).toMatchObject({ + id: "sch_1", + when: { type: "every", every: "30 seconds" }, + until: { type: "shell", run: "gh pr checks 59962" }, + }) + expect(rows.get(Schedule.ID.make("sch_1"))?.until).toEqual({ + type: "shell", + run: "gh pr checks 59962", + }) + expect( + yield* direct(registry, "schedule_create", { + when: { every: "2 minutes" }, + if: { run: "git status -s", changes: true }, + prompt: "working tree changed", + }), + ).toContain('if "git status -s", active') + }), + ) }) function document(experimental: ConfigExperimental.Info | undefined): Entry { diff --git a/packages/protocol/openapi.json b/packages/protocol/openapi.json index a8449d20a5dc..7b32cff5f78d 100644 --- a/packages/protocol/openapi.json +++ b/packages/protocol/openapi.json @@ -21186,6 +21186,57 @@ "required": ["type", "prompt"], "additionalProperties": false }, + "Schedule.Condition": { + "anyOf": [ + { + "$ref": "#/components/schemas/Schedule.Condition.Shell" + } + ] + }, + "Schedule.Condition.Shell": { + "type": "object", + "properties": { + "type": { + "type": "string", + "enum": ["shell"] + }, + "run": { + "type": "string" + }, + "is": { + "type": "string" + }, + "matches": { + "type": "string" + }, + "changes": { + "type": "boolean" + } + }, + "required": ["type", "run"], + "additionalProperties": false + }, + "Schedule.ConditionInput": { + "type": "object", + "properties": { + "run": { + "type": "string", + "minLength": 1, + "description": "Shell command to run in the Session's directory. Alone, exit 0 wakes and stdout is appended." + }, + "is": { + "type": "string" + }, + "matches": { + "type": "string" + }, + "changes": { + "type": "boolean" + } + }, + "required": ["run"], + "additionalProperties": false + }, "Schedule.CreatePayload": { "type": "object", "properties": { @@ -21198,16 +21249,22 @@ "when": { "$ref": "#/components/schemas/Schedule.WhenInput" }, + "if": { + "$ref": "#/components/schemas/Schedule.ConditionInput" + }, + "until": { + "$ref": "#/components/schemas/Schedule.ConditionInput" + }, "ends": { "$ref": "#/components/schemas/Schedule.Ends" }, "prompt": { "type": "string", "minLength": 1, - "description": "The message the Session receives each time the schedule fires." + "description": "The message the Session receives when the schedule wakes." } }, - "required": ["sessionID", "when", "prompt"], + "required": ["sessionID", "prompt"], "additionalProperties": false }, "Schedule.Duration": { @@ -21253,6 +21310,12 @@ "tz": { "type": "string" }, + "if": { + "$ref": "#/components/schemas/Schedule.Condition" + }, + "until": { + "$ref": "#/components/schemas/Schedule.Condition" + }, "action": { "$ref": "#/components/schemas/Schedule.Action" }, @@ -21281,6 +21344,9 @@ "lastOutcome": { "$ref": "#/components/schemas/Schedule.Outcome" }, + "lastOutput": { + "type": "string" + }, "time": { "type": "object", "properties": { @@ -21300,7 +21366,7 @@ }, "Schedule.Outcome": { "type": "string", - "enum": ["fired", "late", "skipped", "disabled", "failed"] + "enum": ["fired", "late", "checked", "expired", "skipped", "disabled", "failed"] }, "Schedule.Status": { "type": "string", @@ -21480,8 +21546,7 @@ "required": ["cron"], "additionalProperties": false } - ], - "description": "When it fires: { in: \"20 minutes\" }, { at: \"2026-10-27T09:00\" }, { every: \"1 hour\" } or { cron: \"0 9 * * 1-5\" }. Wall-clock times use tz." + ] }, "ScheduleNotFoundErrorEncoded": { "type": "object", diff --git a/packages/schema/src/schedule.ts b/packages/schema/src/schedule.ts index 2b5ea306e9ee..dfbd27b3d110 100644 --- a/packages/schema/src/schedule.ts +++ b/packages/schema/src/schedule.ts @@ -69,10 +69,49 @@ export const Action = Schema.Union([PromptAction]).pipe( ) export type Action = typeof Action.Type +/** + * Runs `run` in the Session's directory before firing, with no model call. With none of `is`, `matches` or `changes` + * set, exit 0 wakes and any non-zero exit is "not yet". With `is`, `matches` or `changes`, non-zero exit is a check + * error, and exit 0 wakes when trimmed stdout equals `is`, stdout matches `matches`, or stdout differs from the + * previous run (`changes`). + */ +export const ShellCondition = Schema.Struct({ + type: Schema.tag("shell"), + run: Schema.String, + is: Schema.String.pipe(optional), + matches: Schema.String.pipe(optional), + changes: Schema.Boolean.pipe(optional), +}).annotate({ identifier: "Schedule.Condition.Shell" }) +export type ShellCondition = typeof ShellCondition.Type + +/** Checked before firing; the schedule wakes the Session only when the condition passes. */ +export const Condition = Schema.Union([ShellCondition]).pipe( + Schema.toTaggedUnion("type"), + Schema.annotate({ identifier: "Schedule.Condition" }), +) +export type Condition = typeof Condition.Type + +export const CheckMode = Schema.Literals(["if", "until"]).annotate({ identifier: "Schedule.CheckMode" }) +export type CheckMode = typeof CheckMode.Type + +export const StoredCheck = Schema.Struct({ + mode: CheckMode, + condition: Condition, +}).annotate({ identifier: "Schedule.StoredCheck" }) +export type StoredCheck = typeof StoredCheck.Type + export const Status = Schema.Literals(["active", "paused", "done"]).annotate({ identifier: "Schedule.Status" }) export type Status = typeof Status.Type -export const Outcome = Schema.Literals(["fired", "late", "skipped", "disabled", "failed"]).annotate({ +export const Outcome = Schema.Literals([ + "fired", + "late", + "checked", + "expired", + "skipped", + "disabled", + "failed", +]).annotate({ identifier: "Schedule.Outcome", }) export type Outcome = typeof Outcome.Type @@ -83,6 +122,8 @@ export const Info = Schema.Struct({ name: Schema.String.pipe(optional), when: When, tz: Schema.String, + if: Condition.pipe(optional), + until: Condition.pipe(optional), action: Action, status: Status, nextDueAt: DateTimeUtcFromMillis.pipe(optional), @@ -91,6 +132,7 @@ export const Info = Schema.Struct({ runCount: NonNegativeInt, lastFiredAt: DateTimeUtcFromMillis.pipe(optional), lastOutcome: Outcome.pipe(optional), + lastOutput: Schema.String.pipe(optional), time: Schema.Struct({ created: DateTimeUtcFromMillis, updated: DateTimeUtcFromMillis }), }).annotate({ identifier: "Schedule.Info" }) export type Info = typeof Info.Type @@ -98,7 +140,9 @@ export type Info = typeof Info.Type export const CreateInput = Schema.Struct({ sessionID: SessionID, name: Schema.String.pipe(optional), - when: When, + when: When.pipe(optional), + if: Condition.pipe(optional), + until: Condition.pipe(optional), ends: Ends.pipe(optional), action: Action, }).annotate({ identifier: "Schedule.CreateInput" }) @@ -116,18 +160,42 @@ export const WhenInput = Schema.Union([ ]).annotate({ identifier: "Schedule.WhenInput" }) export type WhenInput = typeof WhenInput.Type -/** A schedule as callers describe it: when it fires, when it stops, and the prompt it wakes the Session with. */ +export const ShellConditionInput = Schema.Struct({ + run: Schema.String.check(Schema.isMinLength(1)).annotate({ + description: "Shell command to run in the Session's directory. Alone, exit 0 wakes and stdout is appended.", + }), + is: Schema.String.pipe(optional).annotate({ + description: "Wake when exit is 0 and trimmed stdout equals this value.", + }), + matches: Schema.String.pipe(optional).annotate({ + description: "Wake when exit is 0 and stdout matches this regular expression.", + }), + changes: Schema.Boolean.pipe(optional).annotate({ + description: "Wake when exit is 0 and stdout differs from the previous check.", + }), +}).annotate({ identifier: "Schedule.ConditionInput" }) +export type ConditionInput = typeof ShellConditionInput.Type +export const ConditionInput = ShellConditionInput + +/** A schedule as callers describe it: when it fires, optional condition, when it stops, and the prompt it sends. */ export const Request = Schema.Struct({ name: Schema.String.pipe(optional).annotate({ description: "A short label for the schedule." }), - when: WhenInput.annotate({ + when: WhenInput.pipe(optional).annotate({ description: - 'When it fires: { in: "20 minutes" }, { at: "2026-10-27T09:00" }, { every: "1 hour" } or { cron: "0 9 * * 1-5" }. Wall-clock times use tz.', + 'When it fires: { in: "20 minutes" }, { at: "2026-10-27T09:00" }, { every: "1 hour" } or { cron: "0 9 * * 1-5" }. Optional with `until` (defaults to { every: "30 seconds" }).', + }), + if: ConditionInput.pipe(optional).annotate({ + description: "Recurring gate: check on each slot with no model call, and wake on every slot where it passes.", + }), + until: ConditionInput.pipe(optional).annotate({ + description: + 'One-shot gate: poll with no model call, wake once when it passes, and stop. Defaults ends.after to "24 hours".', }), ends: Ends.pipe(optional).annotate({ description: 'Stops a repeating schedule at the first limit reached: { after: "8 hours" }, { at }, { times: 3 }.', }), prompt: Schema.String.check(Schema.isMinLength(1)).annotate({ - description: "The message the Session receives each time the schedule fires.", + description: "The message the Session receives when the schedule wakes.", }), }).annotate({ identifier: "Schedule.Request" }) export type Request = typeof Request.Type @@ -135,11 +203,21 @@ export type Request = typeof Request.Type export const toCreateInput = (sessionID: SessionID, request: Request): CreateInput => ({ sessionID, name: request.name, - when: toWhen(request.when), + when: request.when === undefined ? undefined : toWhen(request.when), + if: request.if === undefined ? undefined : toCondition(request.if), + until: request.until === undefined ? undefined : toCondition(request.until), ends: request.ends, action: { type: "prompt", prompt: request.prompt }, }) +const toCondition = (condition: ConditionInput): Condition => ({ + type: "shell", + run: condition.run, + is: condition.is, + matches: condition.matches, + changes: condition.changes, +}) + const toWhen = (when: WhenInput): When => { if (when.in !== undefined) return { type: "in", in: when.in, tz: when.tz } if (when.at !== undefined) return { type: "at", at: when.at, tz: when.tz } diff --git a/packages/server/src/handlers/schedule.ts b/packages/server/src/handlers/schedule.ts index 1d65d9ec8cf6..701ec133e97b 100644 --- a/packages/server/src/handlers/schedule.ts +++ b/packages/server/src/handlers/schedule.ts @@ -1,3 +1,4 @@ +import { Permission } from "@opencode/core/permission" import { Schedule } from "@opencode/core/schedule" import type { Session } from "@opencode/core/session" import { ForbiddenError, InvalidRequestError, ScheduleNotFoundError } from "@opencode/protocol/errors" @@ -29,12 +30,16 @@ export const makeSchedule = Effect.gen(function* () { return { create: (payload: CreatePayload) => + // Stateless HTTP callers have no guaranteed UI listener to answer an interactive ask, so settle asks unattended. schedules.create(Schedule.toCreateInput(payload.sessionID, payload)).pipe( + Effect.provideService(Permission.Unattended, []), Effect.map((data) => ({ data })), Effect.catchTags({ "Schedule.InvalidError": (error) => Effect.fail(new InvalidRequestError({ message: error.message, kind: "schedule" })), "Schedule.DisabledError": () => Effect.fail(disabled()), + "Permission.BlockedError": (error) => Effect.fail(new ForbiddenError({ message: error.message })), + "Permission.CorrectedError": (error) => Effect.fail(new ForbiddenError({ message: error.feedback })), "Session.NotFoundError": (error) => Effect.fail(missingSession(error)), }), ), diff --git a/packages/server/test/schedule.test.ts b/packages/server/test/schedule.test.ts index 8919b3d8608f..fa1b97db5ded 100644 --- a/packages/server/test/schedule.test.ts +++ b/packages/server/test/schedule.test.ts @@ -91,6 +91,35 @@ it.live("manages schedules through the experimental API where the session's conf await expect(api.experimental.schedule.list({ sessionID: off.id })).rejects.toMatchObject({ _tag: "ForbiddenError", }) + + // Shell until check that already passes returns done immediately with its output. + const matched = await api.experimental.schedule.create({ + sessionID: session.id, + until: { run: "printf ready" }, + prompt: "ready now", + }) + expect(matched).toMatchObject({ + status: "done", + lastOutcome: "fired", + lastOutput: "ready", + until: { type: "shell", run: "printf ready" }, + }) + + // Stateless HTTP create settles an "ask" command as 403 ForbiddenError instead of hanging. + const askDir = path.join(dir, "ask") + await fs.mkdir(askDir, { recursive: true }) + await fs.writeFile( + path.join(askDir, "opencode.json"), + JSON.stringify({ experimental: { schedules: true }, permission: { shell: "ask" } }), + ) + const askSession = await api.session.create({ location: { directory: askDir } }) + await expect( + api.experimental.schedule.create({ + sessionID: askSession.id, + until: { run: "printf blocked" }, + prompt: "never", + }), + ).rejects.toMatchObject({ _tag: "ForbiddenError" }) }) }).pipe(Effect.scoped), )