Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 13 additions & 3 deletions bun.lock

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

2 changes: 2 additions & 0 deletions packages/opencode/package.json
Original file line number Diff line number Diff line change
Expand Up @@ -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",
Expand Down
78 changes: 76 additions & 2 deletions packages/opencode/src/cli/cmd/detached.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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"

Expand Down Expand Up @@ -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 <prompt>",
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=<duration> or --cron=<expression>")
const sessionID =
args.session ??
(
await call<SessionInfo>(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",
Expand Down
9 changes: 8 additions & 1 deletion packages/opencode/src/cli/cmd/session.ts
Original file line number Diff line number Diff line change
@@ -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"
Expand Down Expand Up @@ -52,6 +58,7 @@ export const SessionCommand = cmd({
.command(SessionStartCommand)
.command(SessionRunningCommand)
.command(SessionWatchCommand)
.command(SessionScheduleCommand)
.command(SessionDoctorCommand)
.demandCommand(),
async handler() {},
Expand Down
12 changes: 10 additions & 2 deletions packages/temporal/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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
Expand Down Expand Up @@ -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.
24 changes: 24 additions & 0 deletions packages/temporal/scripts/detached-session-check.sh
Original file line number Diff line number Diff line change
Expand Up @@ -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)
16 changes: 14 additions & 2 deletions packages/temporal/src/executor.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down Expand Up @@ -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.
Expand All @@ -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()
Expand All @@ -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()
Expand Down
41 changes: 41 additions & 0 deletions packages/temporal/src/schedule-drain.ts
Original file line number Diff line number Diff line change
@@ -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,
),
})
Loading
Loading