mirror of
https://github.com/anomalyco/opencode.git
synced 2026-11-08 21:38:46 +00:00
feat(core): gate scheduled prompts with shell checks (#60137)
This commit is contained in:
Kit Langton
GitHub
parent
756f832902
commit
c784d17b83
20 changed files with 1482 additions and 90 deletions
4
packages/client/src/effect/api/api.ts
generated
4
packages/client/src/effect/api/api.ts
generated
|
|
@ -2374,7 +2374,9 @@ export type ExperimentalPersistentPtyConnectTokenOperation<E = never> = (
|
|||
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
|
||||
}
|
||||
|
|
|
|||
2
packages/client/src/effect/generated/client.ts
generated
2
packages/client/src/effect/generated/client.ts
generated
|
|
@ -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"],
|
||||
},
|
||||
|
|
|
|||
2
packages/client/src/promise/generated/client.ts
generated
2
packages/client/src/promise/generated/client.ts
generated
|
|
@ -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"],
|
||||
},
|
||||
|
|
|
|||
52
packages/client/src/promise/generated/types.ts
generated
52
packages/client/src/promise/generated/types.ts
generated
|
|
@ -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
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
2
packages/core/src/database/migration.gen.ts
generated
2
packages/core/src/database/migration.gen.ts
generated
|
|
@ -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[]
|
||||
|
|
|
|||
|
|
@ -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
|
||||
4
packages/core/src/database/schema.gen.ts
generated
4
packages/core/src/database/schema.gen.ts
generated
|
|
@ -122,6 +122,7 @@ const schema: Omit<DatabaseMigration.Migration, "id"> = {
|
|||
\`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<DatabaseMigration.Migration, "id"> = {
|
|||
\`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
|
||||
|
|
|
|||
|
|
@ -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<Permission.Ruleset | undefined>("@opencode/Permission/Unattended", {
|
||||
defaultValue: () => undefined,
|
||||
})
|
||||
|
||||
export class DeclinedError extends Schema.TaggedError<DeclinedError>()("Permission.DeclinedError", {}) {}
|
||||
|
||||
export class CorrectedError extends Schema.TaggedError<CorrectedError>()("Permission.CorrectedError", {
|
||||
feedback: Schema.String,
|
||||
}) {}
|
||||
}) {
|
||||
override get message() {
|
||||
return this.feedback
|
||||
}
|
||||
}
|
||||
|
||||
export class BlockedError extends Schema.TaggedError<BlockedError>()("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 {
|
||||
|
|
|
|||
|
|
@ -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.
|
||||
|
||||
|
|
|
|||
|
|
@ -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<NotFoundError>()("Schedule.NotFoundError", {
|
||||
scheduleID: ID,
|
||||
|
|
@ -51,7 +57,9 @@ export class DisabledError extends Schema.TaggedError<DisabledError>()("Schedule
|
|||
}
|
||||
|
||||
export interface Interface {
|
||||
readonly create: (input: CreateInput) => Effect.Effect<Info, InvalidError | DisabledError | Session.NotFoundError>
|
||||
readonly create: (
|
||||
input: CreateInput,
|
||||
) => Effect.Effect<Info, InvalidError | DisabledError | Session.NotFoundError | Permission.Error>
|
||||
readonly list: (input?: { readonly sessionID?: SessionSchema.ID }) => Effect.Effect<Info[]>
|
||||
readonly get: (id: ID) => Effect.Effect<Info, NotFoundError>
|
||||
readonly pause: (id: ID) => Effect.Effect<Info, NotFoundError>
|
||||
|
|
@ -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<Row, "ends_at">, 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<StoredCheck | undefined, InvalidError> => {
|
||||
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<string, unknown>,
|
||||
) {
|
||||
// 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,
|
||||
|
|
|
|||
265
packages/core/src/schedule/check.ts
Normal file
265
packages/core/src/schedule/check.ts
Normal file
|
|
@ -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<void, string> {
|
||||
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<string>; readonly commands: ReadonlyArray<string> },
|
||||
metadata?: Record<string, unknown>,
|
||||
) {
|
||||
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
|
||||
}
|
||||
|
|
@ -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<Schedule.Action>().notNull(),
|
||||
check: text({ mode: "json" }).$type<Schedule.StoredCheck>(),
|
||||
when: text({ mode: "json" }).$type<Schedule.When>().notNull(),
|
||||
tz: text().notNull(),
|
||||
status: text().$type<Schedule.Status>().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<SessionMessage.ID>(),
|
||||
last_outcome: text().$type<Schedule.Outcome>(),
|
||||
/** 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<Schema.Json>(),
|
||||
...Timestamps,
|
||||
},
|
||||
(table) => [
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<void> = 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<LocationServices, FileSystem.DirectoryNotFoundError> => {
|
||||
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<LocationServices, FileSystem.DirectoryNotFoundError>
|
||||
}
|
||||
// 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<LocationServices, FileSystem.DirectoryNotFoundError>,
|
||||
Layer.succeed(Plugin.Service, {
|
||||
awaitActivation: Effect.suspend(() => activationGate),
|
||||
} as unknown as Plugin.Interface),
|
||||
).pipe(Layer.fresh) as unknown as Layer.Layer<LocationServices, FileSystem.DirectoryNotFoundError>
|
||||
},
|
||||
)
|
||||
}),
|
||||
),
|
||||
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<Permission.Request>()
|
||||
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<Schedule.CreateInput, "sessionID" | "action">) =>
|
||||
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" })
|
||||
}),
|
||||
)
|
||||
})
|
||||
})
|
||||
|
|
|
|||
|
|
@ -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 {
|
||||
|
|
|
|||
75
packages/protocol/openapi.json
generated
75
packages/protocol/openapi.json
generated
|
|
@ -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",
|
||||
|
|
|
|||
|
|
@ -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 }
|
||||
|
|
|
|||
|
|
@ -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)),
|
||||
}),
|
||||
),
|
||||
|
|
|
|||
|
|
@ -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),
|
||||
)
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue