diff --git a/bun.lock b/bun.lock index ab2ce122b2aa..e8938ae2a43f 100644 --- a/bun.lock +++ b/bun.lock @@ -675,6 +675,8 @@ "@opencode-ai/script": "workspace:*", "@opencode-ai/temporal": "workspace:*", "@standard-schema/spec": "1.0.0", + "@temporalio/client": "^1.21.0", + "@temporalio/common": "^1.21.0", "@tsconfig/bun": "catalog:", "@types/babel__core": "7.20.5", "@types/bun": "catalog:", @@ -4679,7 +4681,7 @@ "mrmime": ["mrmime@2.0.1", "", {}, "sha512-Y3wQdFg2Va6etvQ5I82yUhGdsKrcYox6p7FfL1LbK2J4V01F9TGlepTIhnK24t7koZibmg82KGglhA1XK5IsLQ=="], - "ms": ["ms@2.1.3", "", {}, "sha512-6FlzubTLZG3J2a/NVCAleEhjzq5oxgHyaCU9yYXvcLsvoVaHJq/s5xXI6/XXP6tz7R9xAOtHnSO/tXtF3WRTlA=="], + "ms": ["ms@3.0.0-canary.1", "", {}, "sha512-kh8ARjh8rMN7Du2igDRO9QJnqCb2xYTJxyQYK7vJJS4TvLLmsbyhiKpSW+t+y26gyOyMd0riphX0GeWKU3ky5g=="], "msgpackr": ["msgpackr@2.0.2", "", { "optionalDependencies": { "msgpackr-extract": "^3.0.4" } }, "sha512-c5hYOXFbP79Slh6Dzd2wzk+jnV7mX1UxfMYtilnY1NmalXPqG8DGb5cYCMBrW4AsH3zekBBZd4QrKz9NhtvYLQ=="], @@ -6395,8 +6397,6 @@ "@temporalio/client/uuid": ["uuid@11.1.1", "", { "bin": { "uuid": "dist/esm/bin/uuid" } }, "sha512-vIYxrBCC/N/K+Js3qSN88go7kIfNPssr/hHCesKCQNAjmgvYS2oqr69kIufEG+O4+PfezOH4EbIeHCfFov8ZgQ=="], - "@temporalio/common/ms": ["ms@3.0.0-canary.1", "", {}, "sha512-kh8ARjh8rMN7Du2igDRO9QJnqCb2xYTJxyQYK7vJJS4TvLLmsbyhiKpSW+t+y26gyOyMd0riphX0GeWKU3ky5g=="], - "@testing-library/dom/aria-query": ["aria-query@5.3.0", "", { "dependencies": { "dequal": "^2.0.3" } }, "sha512-b0P0sZPKtyu8HkeRAfCq0IfURZK+SuwMjY1UXGBU27wpAiTwQAIlq56IbIO+ytk/JjS1fMR14ee5WBBfKi5J6A=="], "@testing-library/dom/dom-accessibility-api": ["dom-accessibility-api@0.5.16", "", {}, "sha512-X7BJ2yElsnOJ30pZF4uIIDfBEVgF4XEBxL9Bxhy6dnrm5hkzqmsWHGTiHqRiITNhMyFLyAiWndIJP7Z1NTteDg=="], @@ -6519,6 +6519,8 @@ "cross-spawn/which": ["which@2.0.2", "", { "dependencies": { "isexe": "^2.0.0" }, "bin": { "node-which": "./bin/node-which" } }, "sha512-BLI3Tl1TW3Pvl70l3yq3Y64i+awpwXqsGBYWkkqMtnbXgrMD+yj7rhW0kuEDxzJaYXGjEW5ogapKNMEKNMjibA=="], + "debug/ms": ["ms@2.1.3", "", {}, "sha512-6FlzubTLZG3J2a/NVCAleEhjzq5oxgHyaCU9yYXvcLsvoVaHJq/s5xXI6/XXP6tz7R9xAOtHnSO/tXtF3WRTlA=="], + "dir-compare/minimatch": ["minimatch@3.1.5", "", { "dependencies": { "brace-expansion": "^1.1.7" } }, "sha512-VgjWUsnnT6n+NUk6eZq77zeFdpW2LWDzP6zFGrCbHXiYNul5Dzqk2HHQ5uFH2DNW5Xbp8+jVzaeNt94ssEEl4w=="], "dir-compare/p-limit": ["p-limit@3.1.0", "", { "dependencies": { "yocto-queue": "^0.1.0" } }, "sha512-TYOanM3wGwNGsZN2cVTYPArw454xnXj5qmWF1bEoAc4+cU/ol7GVh7odevjp1FNHduHc3KZMcFduxU5Xc6uJRQ=="], @@ -6607,6 +6609,8 @@ "htmlparser2/entities": ["entities@4.5.0", "", {}, "sha512-V0hjH4dGPh9Ao5p0MoRY6BVqtwCjhz6vI5LT8AJ55H+4g9/4vbHx1I54fS0XuclLhDHArPQCiMjDxjaL8fPxhw=="], + "humanize-ms/ms": ["ms@2.1.3", "", {}, "sha512-6FlzubTLZG3J2a/NVCAleEhjzq5oxgHyaCU9yYXvcLsvoVaHJq/s5xXI6/XXP6tz7R9xAOtHnSO/tXtF3WRTlA=="], + "iconv-corefoundation/cli-truncate": ["cli-truncate@2.1.0", "", { "dependencies": { "slice-ansi": "^3.0.0", "string-width": "^4.2.0" } }, "sha512-n8fOixwDD6b/ObinzTrp1ZKFzbgvKZvuz/TvejnLn1aQfC6r52XEx85FmuC+3HI+JM7coBRXUvNqEU2PHVrHpg=="], "iconv-corefoundation/node-addon-api": ["node-addon-api@1.7.2", "", {}, "sha512-ibPK3iA+vaY1eEjESkQkM0BbCqFOaZMiXRTtdB0u7b4djtY6JnsjvPdUHVMg6xQt3B8fpTTWHI9A+ADjM9frzg=="], @@ -6619,6 +6623,8 @@ "js-beautify/nopt": ["nopt@7.2.1", "", { "dependencies": { "abbrev": "^2.0.0" }, "bin": { "nopt": "bin/nopt.js" } }, "sha512-taM24ViiimT/XntxbPyJQzCG+p4EKOpgD3mxFwW38mGjVUrfERQOeY4EDHjdnptttfHuHQXFx+lTP08Q+mLa/w=="], + "jsonwebtoken/ms": ["ms@2.1.3", "", {}, "sha512-6FlzubTLZG3J2a/NVCAleEhjzq5oxgHyaCU9yYXvcLsvoVaHJq/s5xXI6/XXP6tz7R9xAOtHnSO/tXtF3WRTlA=="], + "katex/commander": ["commander@8.3.0", "", {}, "sha512-OkTL9umf+He2DZkUq8f8J9of7yL6RJKI24dVITBmNfZBmri9zYZQrKkuXiKhyfPSu8tUhnVBB1iKXevvnlR4Ww=="], "lazystream/readable-stream": ["readable-stream@2.3.8", "", { "dependencies": { "core-util-is": "~1.0.0", "inherits": "~2.0.3", "isarray": "~1.0.0", "process-nextick-args": "~2.0.0", "safe-buffer": "~5.1.1", "string_decoder": "~1.1.1", "util-deprecate": "~1.0.1" } }, "sha512-8p0AUk4XODgIewSi0l8Epjs+EVnWiK7NoDIEGU0HhE7+ZyY8D1IMY7odu5lRrFXGg71L15KG8QrPmum45RTtdA=="], @@ -6737,6 +6743,8 @@ "send/mime": ["mime@1.6.0", "", { "bin": { "mime": "cli.js" } }, "sha512-x0Vn8spI+wuJ1O6S7gnbaQg8Pxh4NNHb7KSINmEWKiPE4RKOplvijn+NkmYmmRgP68mc70j2EbeTFRsrswaQeg=="], + "send/ms": ["ms@2.1.3", "", {}, "sha512-6FlzubTLZG3J2a/NVCAleEhjzq5oxgHyaCU9yYXvcLsvoVaHJq/s5xXI6/XXP6tz7R9xAOtHnSO/tXtF3WRTlA=="], + "serialize-error/type-fest": ["type-fest@0.13.1", "", {}, "sha512-34R7HTnG0XIJcBSn5XhDd7nNFPRcXYRZrBB2O2jdKqYODldSzBAqzsWoZYYvduky73toYS/ESqxPvkDf/F0XMg=="], "sharp/detect-libc": ["detect-libc@2.1.2", "", {}, "sha512-Btj2BOOO83o3WyH59e8MgXsxEQVcarkUOpEYrubB0urwnN10yQ364rsiByU11nZlqWYZm05i/of7io4mzihBtQ=="], @@ -7781,6 +7789,8 @@ "@jsx-email/cli/vite/esbuild/@esbuild/win32-x64": ["@esbuild/win32-x64@0.18.20", "", { "os": "win32", "cpu": "x64" }, "sha512-kTdfRcSiDfQca/y9QIkng02avJ+NCaQvrMejlsB3RRv5sE9rRoeBPISaZpKxHELzRxZyLvNts1P27W3wV+8geQ=="], + "@modelcontextprotocol/sdk/express/send/ms": ["ms@2.1.3", "", {}, "sha512-6FlzubTLZG3J2a/NVCAleEhjzq5oxgHyaCU9yYXvcLsvoVaHJq/s5xXI6/XXP6tz7R9xAOtHnSO/tXtF3WRTlA=="], + "@modelcontextprotocol/sdk/express/type-is/content-type": ["content-type@2.0.0", "", {}, "sha512-j/O/d7GcZCyNl7/hwZAb606rzqkyvaDctLmckbxLzHvFBzTJHuGEdodATcP3yIRoDrLHkIATJuvzbFlp/ki2cQ=="], "@modelcontextprotocol/sdk/express/type-is/media-typer": ["media-typer@1.1.0", "", {}, "sha512-aisnrDP4GNe06UcKFnV5bfMNPBUw4jsLGaWwWfnH3v02GnBuXX2MCVn5RbrWo0j3pczUilYblq7fQ7Nw2t5XKw=="], diff --git a/packages/opencode/package.json b/packages/opencode/package.json index da2fff200461..1abf2fd2428c 100644 --- a/packages/opencode/package.json +++ b/packages/opencode/package.json @@ -33,6 +33,8 @@ "@octokit/webhooks-types": "7.6.1", "@opencode-ai/core": "workspace:*", "@opencode-ai/temporal": "workspace:*", + "@temporalio/client": "^1.21.0", + "@temporalio/common": "^1.21.0", "@opencode-ai/http-recorder": "workspace:*", "@opencode-ai/script": "workspace:*", "@standard-schema/spec": "1.0.0", diff --git a/packages/opencode/src/cli/cmd/detached.ts b/packages/opencode/src/cli/cmd/detached.ts index 5a091bfd93f4..ad8c85f717cb 100644 --- a/packages/opencode/src/cli/cmd/detached.ts +++ b/packages/opencode/src/cli/cmd/detached.ts @@ -5,13 +5,15 @@ // interchangeable (any of them reads the shared store and signals the same workflows), so a client // needs an endpoint and a session id, never a particular host. That is the whole reason a session // can outlive the process that started it, and it is why `start`, `running` and `watch` never talk -// to Temporal. `doctor` does, and loads the driver only when it runs, so the rest of the `session` -// command never pays for it. +// to Temporal. `doctor` and `schedule` do, and load the driver only when they run, so the rest of +// the `session` command never pays for it. import type { Argv } from "yargs" import { cmd } from "./cmd" import { UI } from "../ui" import { ServerAuth } from "@/server/auth" +// Type-only: the SDK types a duration as a template literal, and this takes one from a person. +import type { Duration } from "@temporalio/common" const DEFAULT_URL = "http://127.0.0.1:4096" @@ -129,6 +131,78 @@ export const SessionDoctorCommand = cmd({ }, }) +// A turn nobody starts. `start` still needs something running to hand the prompt to; a schedule +// does not, which is the difference between a session you can walk away from and one that runs +// without you. The session is created once, here, over HTTP like everything else in this file; the +// firing itself reaches only Temporal, and a deployment with no serve process at all still runs it. +export const SessionScheduleCommand = cmd({ + command: "schedule ", + describe: "run a prompt on a schedule, with no client at firing time", + builder: (yargs: Argv) => + remoteOptions(yargs) + .positional("prompt", { type: "string", describe: "what the agent should do", demandOption: true }) + .option("every", { type: "string", describe: "interval, e.g. 1h" }) + .option("cron", { type: "string", describe: "cron expression, e.g. '0 9 * * *'" }) + .option("id", { type: "string", describe: "schedule id (default: generated)" }) + .option("session", { type: "string", describe: "an existing session to prompt (default: a new one)" }) + .option("dir", { type: "string", describe: "session working directory (default: this one)" }), + handler: async (args) => { + const r = remote(args) + try { + if (!args.every && !args.cron) throw new Error("schedule wants --every= or --cron=") + const sessionID = + args.session ?? + ( + await call(r, "/session", { + method: "POST", + body: JSON.stringify({ directory: args.dir ?? process.cwd() }), + }) + ).id + const { TemporalConfig } = await import("@opencode-ai/temporal/config") + const config = TemporalConfig.fromEnv() + const { Client, Connection, ScheduleOverlapPolicy } = await import("@temporalio/client") + const connection = await Connection.connect(TemporalConfig.connectionOptions(config)) + try { + const client = new Client({ connection, namespace: config.namespace }) + const scheduleId = args.id ?? `opencode-${sessionID}` + await client.schedule.create({ + scheduleId, + spec: { + ...(args.cron ? { cronExpressions: [args.cron] } : {}), + ...(args.every ? { intervals: [{ every: args.every as Duration }] } : {}), + }, + // A firing that lands while the last one is still working is skipped rather than queued. + // An agent task is not a metrics scrape: two of them on one project is a bad day. + policies: { overlap: ScheduleOverlapPolicy.SKIP }, + action: { + type: "startWorkflow", + workflowType: "scheduledPrompt", + taskQueue: config.taskQueue, + args: [ + { + sessionID, + text: args.prompt, + session: { idleTimeout: config.idleTimeout, stepped: config.stepped === true }, + }, + ], + }, + }) + if (args.json) { + emit(JSON.stringify({ schedule: scheduleId, session: sessionID })) + return + } + emit(scheduleId) + UI.println(` every firing prompts ${sessionID}; follow it with: opencode session watch ${sessionID}`) + } finally { + await connection.close() + } + } catch (error) { + UI.error(error instanceof Error ? error.message : String(error)) + process.exitCode = 1 + } + }, +}) + export const SessionRunningCommand = cmd({ command: "running", describe: "list the sessions this deployment is executing right now", diff --git a/packages/opencode/src/cli/cmd/session.ts b/packages/opencode/src/cli/cmd/session.ts index d0df40554ab7..94b746b63b72 100644 --- a/packages/opencode/src/cli/cmd/session.ts +++ b/packages/opencode/src/cli/cmd/session.ts @@ -1,7 +1,13 @@ import type { Argv } from "yargs" import { Effect } from "effect" import { cmd } from "./cmd" -import { SessionDoctorCommand, SessionRunningCommand, SessionStartCommand, SessionWatchCommand } from "./detached" +import { + SessionDoctorCommand, + SessionRunningCommand, + SessionScheduleCommand, + SessionStartCommand, + SessionWatchCommand, +} from "./detached" import { effectCmd, fail } from "../effect-cmd" import { Session } from "@/session/session" import { SessionID } from "../../session/schema" @@ -52,6 +58,7 @@ export const SessionCommand = cmd({ .command(SessionStartCommand) .command(SessionRunningCommand) .command(SessionWatchCommand) + .command(SessionScheduleCommand) .command(SessionDoctorCommand) .demandCommand(), async handler() {}, diff --git a/packages/temporal/README.md b/packages/temporal/README.md index e3f7565ce908..9afc34cdef7c 100644 --- a/packages/temporal/README.md +++ b/packages/temporal/README.md @@ -281,6 +281,9 @@ opencode session running --attach http://gateway:4096 # follow one from anywhere, and stop when the turn stops opencode session watch ses_abc123 --attach http://gateway:4096 + +# a turn nobody starts: the firing needs no client and no serve process +opencode session schedule "review yesterday's merges" --cron "0 9 * * *" --attach http://gateway:4096 ``` `--attach` takes any serve in the deployment, because they are interchangeable: each reads the same @@ -293,6 +296,11 @@ can read it from the replayable stream. A turn the user stopped or a provider er no ordinary ending, so `watch` also asks the running set on a slow poll, the one reading that can end a follow without being able to hang one. +`schedule` creates a Temporal schedule whose firing runs a workflow that admits the prompt itself +and starts the session's supervisor, so at firing time there is no client and no serve process. The +prompt is admitted as queued, so a firing that lands while the last turn is still working is drained +when that turn ends, and overlapping firings are skipped rather than stacked. + ### Picking a deployment rather than assembling one The settings are not independent, and getting them wrong fails as something else later: a store @@ -341,5 +349,5 @@ an outcome query or a person to decide what happened. `scripts/detached-session-check.sh` runs the client story against a Temporal dev server, one standalone worker, two serve processes and one shared store: a turn started on serve A survives A -being killed mid-tool, serve B replays it, and `start`, `running` and `watch` work from a cold -client. It needs an OpenAI key and does not run in CI. +being killed mid-tool, serve B replays it, and `start`, `running`, `watch` and `schedule` work from +a cold client. It needs an OpenAI key and does not run in CI. diff --git a/packages/temporal/scripts/detached-session-check.sh b/packages/temporal/scripts/detached-session-check.sh index 42a7512e15c0..a24b2cfd6745 100755 --- a/packages/temporal/scripts/detached-session-check.sh +++ b/packages/temporal/scripts/detached-session-check.sh @@ -166,6 +166,30 @@ grep -q "WATCHED" "$RUN/logs/watch.txt" && ok "session watch followed the turn" [ "$took" -lt 100 ] && ok "session watch stopped when the turn did (${took}s)" \ || bad "session watch stopped when the turn did" "${took}s, so it hung" +# --- 6. a turn nobody starts. The schedule is a Temporal object, so at firing time there is no +# client and no HTTP call: the workflow admits the prompt itself and starts the session's own +# supervisor. Prompted into a fresh session so what arrives can only have come from the firing. +sched=$(timeout 90 bun run "$OC" session schedule \ + "Use the bash tool to run exactly: echo SCHEDULED. Then report the output." \ + --every 10s --attach "$B" --dir "$RUN/proj" --json 2>/dev/null) +sid3=$(printf '%s' "$sched" | sed -n 's/.*"session":"\([^"]*\)".*/\1/p') +scheduleId=$(printf '%s' "$sched" | sed -n 's/.*"schedule":"\([^"]*\)".*/\1/p') +[ -n "$sid3" ] && ok "session schedule created one" || bad "session schedule created one" "$sched" + +# Long enough for a firing plus a turn, and nothing here prompts it. +answered="" +for _ in $(seq 1 24); do + sleep 5 + answered=$(curl -s -u "$AUTH" "$B/api/session/$sid3/message" 2>/dev/null || true) + case "$answered" in *SCHEDULED*) break ;; esac +done +case "$answered" in + *SCHEDULED*) ok "a firing ran a turn with no client involved" ;; + *) bad "a firing ran a turn with no client involved" "$(printf '%s' "$answered" | head -c 200)" ;; +esac +[ -n "$scheduleId" ] && temporal schedule delete --schedule-id "$scheduleId" \ + --address "127.0.0.1:$PORT_TEMPORAL" >/dev/null 2>&1 + echo [ "$fails" -eq 0 ] && echo "detached-session-check: OK" || echo "detached-session-check: $fails failed" exit $([ "$fails" -eq 0 ] && echo 0 || echo 1) diff --git a/packages/temporal/src/executor.ts b/packages/temporal/src/executor.ts index 052163b71d1e..acc7998de2e8 100644 --- a/packages/temporal/src/executor.ts +++ b/packages/temporal/src/executor.ts @@ -17,6 +17,7 @@ import { SessionExecution } from "@opencode-ai/core/session/execution" import { makeStepActivities, makeSteppedTurnActivities } from "./activities" import { makeDrains } from "./drain" import { makeSteppedDrains } from "./stepped-drain" +import { makeScheduleDrains } from "./schedule-drain" import { queueForWorktree, queueForWorker } from "./queue" import { Database } from "@opencode-ai/core/database/database" import { ProjectTable } from "@opencode-ai/core/project/sql" @@ -115,6 +116,9 @@ const layer = Layer.effect( // The stepped mode's three drains. Registered unconditionally: which mode a session runs is a // property of its workflow input, so a worker has to be able to serve either. const stepped = makeSteppedDrains({ inSession: drains.inSession, stepQueue: STEP_QUEUE }) + // What a schedule fires into: admitting a prompt is a row in the store, and a workflow cannot + // write one. Registered on every worker, because a firing lands wherever one is polling. + const schedules = makeScheduleDrains({ db, events }) // Worker connection (native) hosts the runTurnStep activity + the workflow. Skipped in // client-only role so serve can run without an embedded worker. @@ -137,7 +141,11 @@ const layer = Layer.effect( namespace: NAMESPACE, taskQueue: POLL_QUEUE, workflowsPath: fileURLToPath(new URL("./workflow.ts", import.meta.url)), - activities: { ...makeStepActivities(drains.stepDrain), ...makeSteppedTurnActivities(stepped) }, + activities: { + ...makeStepActivities(drains.stepDrain), + ...makeSteppedTurnActivities(stepped), + promptSession: schedules.promptDrain, + }, }), ) const runHandle = worker.run() @@ -159,7 +167,11 @@ const layer = Layer.effect( connection: nativeConn, namespace: NAMESPACE, taskQueue: STEP_QUEUE, - activities: { ...makeStepActivities(drains.stepDrain), ...makeSteppedTurnActivities(stepped) }, + activities: { + ...makeStepActivities(drains.stepDrain), + ...makeSteppedTurnActivities(stepped), + promptSession: schedules.promptDrain, + }, }), ) const pinnedHandle = pinnedWorker.run() diff --git a/packages/temporal/src/schedule-drain.ts b/packages/temporal/src/schedule-drain.ts new file mode 100644 index 000000000000..d68e5a0a122c --- /dev/null +++ b/packages/temporal/src/schedule-drain.ts @@ -0,0 +1,41 @@ +// Admitting a prompt to a session that already exists, without waking anything. This is what a +// start with no client is made of: a prompt is a durable row before it is work, and writing that +// row needs the store, which a workflow cannot reach. Waking the session is the workflow's own job. +// Separating the two is what lets a schedule fire into a deployment where nothing runs but workers. + +import { Effect } from "effect" +import type { Database } from "@opencode-ai/core/database/database" +import { EventV2 } from "@opencode-ai/core/event" +import { SessionInput } from "@opencode-ai/core/session/input" +import { SessionMessage } from "@opencode-ai/core/session/message" +import { SessionSchema } from "@opencode-ai/core/session/schema" +import { Prompt } from "@opencode-ai/schema/prompt" + +export interface PromptDrainInput { + readonly sessionID: string + /** Derived by the workflow from the firing, so a re-driven activity admits nothing twice. */ + readonly messageID: string + readonly text: string +} + +export const makeScheduleDrains = ({ + db, + events, +}: { + readonly db: Database.Interface["db"] + readonly events: EventV2.Interface +}) => ({ + promptDrain: async (input: PromptDrainInput) => + SessionInput.admit(db, events, { + id: SessionMessage.ID.make(input.messageID), + sessionID: SessionSchema.ID.make(input.sessionID), + prompt: Prompt.make({ text: input.text }), + delivery: "queue", + }).pipe( + Effect.asVoid, + // Admission is control-plane input: it must leave the active drain's ownership unchanged. + Effect.provideService(EventV2.EventOwner, undefined), + Effect.scoped, + Effect.runPromise, + ), +}) diff --git a/packages/temporal/src/workflow.ts b/packages/temporal/src/workflow.ts index 5bdbcd375f18..7cfcdc26f10e 100644 --- a/packages/temporal/src/workflow.ts +++ b/packages/temporal/src/workflow.ts @@ -20,10 +20,14 @@ import { allHandlersFinished, log, workflowInfo, + startChild, + getExternalWorkflowHandle, + ParentClosePolicy, } from "@temporalio/workflow" +import { WorkflowExecutionAlreadyStartedError } from "@temporalio/common" import type { StepActivities, SteppedTurnActivities } from "./activities" import { isHaltFailure, makeSteppedTurn } from "./stepped-turn" -import { SIGNALS, RESUME_UPDATE } from "./protocol" +import { SIGNALS, RESUME_UPDATE, WORKFLOW_ID_PREFIX } from "./protocol" import { makeSupervisor, type SupervisorRuntime } from "./supervisor" const activityOptions = { @@ -67,6 +71,16 @@ const pinnedTo = (taskQueue: string) => ({ }).sealStep, }) +// Admitting a prompt is a row in the store, so it is an activity; it is small and it must not hold +// a firing open if no worker is polling. +const { promptSession } = proxyActivities<{ + promptSession(input: { sessionID: string; messageID: string; text: string }): Promise +}>({ + startToCloseTimeout: "2 minutes", + scheduleToCloseTimeout: "30 minutes", + retry: { maximumAttempts: 10 }, +}) + export const wake = defineSignal(SIGNALS.wake) export const interrupt = defineSignal(SIGNALS.interrupt) export const resume = defineUpdate(RESUME_UPDATE) @@ -180,3 +194,38 @@ export async function sessionTurn(sessionID: string, options?: SessionTurnOption idleTimeout ? { idleTimeout } : undefined, ).sessionTurn(sessionID, startWithWake) } + +/** + * A turn nobody started. + * + * A schedule fires this, and it runs where no client and no serve process exist: the prompt is + * admitted by an activity, because it is a row in the store, and the session's own supervisor is + * started as an abandoned child (or signalled, when it is already running). Nothing here waits for + * the turn: this workflow's job is to hand the work over and finish, which is what makes a firing + * cheap and a missed one visible in the schedule rather than in a run that never ends. + * + * The message id comes from the firing's own workflow id, so a re-drive admits the same prompt + * rather than a second one. + */ +export async function scheduledPrompt(input: { + readonly sessionID: string + readonly text: string + readonly session?: SessionTurnOptions +}): Promise { + const messageID = `msg_sched_${workflowInfo().workflowId}` + await promptSession({ sessionID: input.sessionID, messageID, text: input.text }) + const options: SessionTurnOptions = { ...input.session, startWithWake: true } + try { + await startChild(sessionTurn, { + workflowId: `${WORKFLOW_ID_PREFIX}${input.sessionID}`, + args: [input.sessionID, options], + parentClosePolicy: ParentClosePolicy.ABANDON, + }) + } catch (error) { + // The session is already being driven, which is the ordinary case for a schedule that fires + // faster than a turn takes. The prompt is admitted either way; what it needs is a wake, because + // a supervisor waiting out its idle period is not watching the store. + if (!(error instanceof WorkflowExecutionAlreadyStartedError)) throw error + await getExternalWorkflowHandle(`${WORKFLOW_ID_PREFIX}${input.sessionID}`).signal(wake) + } +} diff --git a/packages/temporal/test/schedule-drain.test.ts b/packages/temporal/test/schedule-drain.test.ts new file mode 100644 index 000000000000..f4cbf9a2ff4f --- /dev/null +++ b/packages/temporal/test/schedule-drain.test.ts @@ -0,0 +1,64 @@ +import { expect } from "bun:test" +import { Effect } from "effect" +import { eq } from "drizzle-orm" +import { Database } from "@opencode-ai/core/database/database" +import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder" +import { LayerNode } from "@opencode-ai/core/effect/layer-node" +import { EventV2 } from "@opencode-ai/core/event" +import { EventSequenceTable } from "@opencode-ai/core/event/sql" +import { Project } from "@opencode-ai/core/project" +import { ProjectTable } from "@opencode-ai/core/project/sql" +import { AbsolutePath } from "@opencode-ai/core/schema" +import { SessionSchema } from "@opencode-ai/core/session/schema" +import { SessionInputTable, SessionTable } from "@opencode-ai/core/session/sql" +import { SessionProjector } from "@opencode-ai/core/session/projector" +import { testEffect } from "../../core/test/lib/effect" +import { makeScheduleDrains } from "../src/schedule-drain" + +const it = testEffect( + AppNodeBuilder.build(LayerNode.group([Database.node, EventV2.node, SessionProjector.node]), [ + [Database.node, Database.layerFromPath(":memory:")], + ]), +) + +it.effect("admits later schedule firings without taking the runner's owner", () => + Effect.gen(function* () { + const { db } = yield* Database.Service + const events = yield* EventV2.Service + const sessionID = SessionSchema.ID.make("ses_schedule") + yield* db + .insert(ProjectTable) + .values({ + id: Project.ID.global, + worktree: AbsolutePath.make("/project"), + sandboxes: [], + }) + .run() + yield* db + .insert(SessionTable) + .values({ + id: sessionID, + project_id: Project.ID.global, + slug: "schedule", + directory: "/project", + title: "schedule", + version: "test", + }) + .run() + const { promptDrain } = makeScheduleDrains({ db, events }) + const first = { sessionID, messageID: "msg_schedule_first", text: "check the project" } + yield* Effect.promise(() => promptDrain(first)) + yield* events.claim(sessionID, "run:1:1") + yield* Effect.promise(() => promptDrain({ ...first, messageID: "msg_schedule_second" })) + yield* Effect.promise(() => promptDrain(first)) + + const prompts = yield* db.select().from(SessionInputTable).all() + const owner = yield* db + .select() + .from(EventSequenceTable) + .where(eq(EventSequenceTable.aggregate_id, sessionID)) + .get() + expect(prompts.map((row) => String(row.id)).sort()).toEqual(["msg_schedule_first", "msg_schedule_second"]) + expect(owner?.owner_id).toBe("run:1:1") + }), +) diff --git a/packages/temporal/test/scheduled-prompt-workflow.test.ts b/packages/temporal/test/scheduled-prompt-workflow.test.ts new file mode 100644 index 000000000000..35c55f72c026 --- /dev/null +++ b/packages/temporal/test/scheduled-prompt-workflow.test.ts @@ -0,0 +1,37 @@ +import { expect, it } from "bun:test" +import { fileURLToPath } from "node:url" +import { TestWorkflowEnvironment } from "@temporalio/testing" +import { Worker } from "@temporalio/worker" + +it("keeps distinct firing IDs after a long schedule name", async () => { + const env = await TestWorkflowEnvironment.createLocal() + const received: string[] = [] + const prefix = "daily-review-".repeat(5) + try { + const worker = await Worker.create({ + connection: env.nativeConnection, + namespace: env.namespace, + taskQueue: "schedule-id-check", + workflowsPath: fileURLToPath(new URL("../src/workflow.ts", import.meta.url)), + activities: { + promptSession: async (input: { messageID: string }) => { + received.push(input.messageID) + }, + runTurnStep: async () => ({ ran: true, continue: false, step: 1, promotion: null }), + }, + }) + await worker.runUntil(async () => { + for (const day of ["01", "02"]) { + await env.client.workflow.execute("scheduledPrompt", { + workflowId: `${prefix}-workflow-2026-09-${day}T09:00:00Z`, + taskQueue: "schedule-id-check", + args: [{ sessionID: "ses_schedule_id", text: "review yesterday's changes" }], + }) + } + }) + expect(received).toHaveLength(2) + expect(new Set(received).size).toBe(2) + } finally { + await env.teardown() + } +}, 120_000)