Skip to content

fix(schedules): submit scheduled runs via prompt_async with SSE monitoring #335

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
chriswritescode-dev merged 2 commits into from
Aug 10, 2026
Merged

fix(schedules): submit scheduled runs via prompt_async with SSE monitoring #335

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
ed4d4b0
fix(schedules): submit scheduled runs via prompt_async with SSE monit…
chriswritescode-dev Aug 10, 2026
3eccbdf
fix(session): use session directory for SSE and queries when it diffe…
chriswritescode-dev Aug 10, 2026
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
1 change: 1 addition & 0 deletions .gitignore
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
node_modules/
.pnpm-store/
*.log
.env
.env.local
Expand Down
2 changes: 1 addition & 1 deletion backend/src/index.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -335,7 +335,7 @@ sseAggregator.setPasswordResolver(() => new SettingsService(db).getOpenCodeServe
sseAggregator.start()

sseAggregator.setScheduledSessionsResolver(
() => scheduleService.getActiveRunSessionIds(),
() => scheduleService.getActiveRunSessions(),
)

const openCodeRestartCoordinator = new OpenCodeRestartCoordinator(openCodeClient, sseAggregator)
Expand Down
230 changes: 118 additions & 112 deletions backend/src/services/schedules.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -44,7 +44,7 @@ import { resolveOpenCodeModel } from './opencode-models'
import type { OpenCodeClient } from './opencode/client'
import type { ScheduleWorktreeManager } from './schedule-worktree'
import type { Repo } from '../types/repo'
import { sseAggregator, type SSEEvent } from './sse-aggregator'
import { sseAggregator, type SSEEvent, type ScheduledSessionRef } from './sse-aggregator'
import { getErrorMessage } from '../utils/error-utils'
import { logger } from '../utils/logger'
import { buildAssistantRepo } from './assistant-mode'
Expand All @@ -63,13 +63,6 @@ interface SessionResponse {
id: string
}

interface PromptResponse {
parts?: Array<{
type?: string
text?: string
}>
}

interface SessionMessagePart {
type?: string
text?: string
Expand Down Expand Up @@ -101,22 +94,23 @@ interface SessionStatus {
next?: number
}

const RUN_POLL_INTERVAL_MS = 2_000
const RUN_POLL_TIMEOUT_MS = 5 * 60_000
const SESSION_STOPPED_ERROR = 'The session stopped without producing an assistant response. This usually means OpenCode restarted mid-run or the session was interrupted. Open the linked session to inspect any partial output and rerun if needed.'

interface SessionSignal {
errorText: string | null
disposed: boolean
}

interface SessionMonitor {
getErrorText(): string | null
isIdle(): boolean
markSubmitted(): void
nextSignal(): Promise<SessionSignal>
dispose(): void
}

function extractResponseText(response: PromptResponse): string {
return (response.parts ?? [])
.filter((part) => part.type === 'text' && typeof part.text === 'string')
.map((part) => part.text?.replace(/<think>[\s\S]*?<\/think>\s*/g, '').trim() ?? '')
.filter(Boolean)
.join('\n\n')
}
type AssistantOutcome =
| { kind: 'busy' }
| { kind: 'settled'; responseText: string | null; errorText: string | null }
| { kind: 'stopped'; responseText: string | null }

function buildSessionTitle(job: ScheduleJob): string {
return `Scheduled: ${job.name}`
Expand Down Expand Up @@ -253,18 +247,6 @@ function buildRunStartedLog(input: {
].join('\n')
}

function parsePromptResponse(responseText: string): PromptResponse | null {
if (!responseText.trim()) {
return null
}

try {
return JSON.parse(responseText) as PromptResponse
} catch {
return null
}
}

function extractAssistantMessageText(parts: SessionMessagePart[] | undefined): string {
return (parts ?? [])
.filter((part) => part.type === 'text' && typeof part.text === 'string')
Expand Down Expand Up @@ -326,8 +308,19 @@ function getSessionStatusType(event: SSEEvent): string | null {
}

function createSessionMonitor(directory: string, sessionId: string): SessionMonitor {
let errorText: string | null = null
let idle = false
const queued: SessionSignal[] = []
let waiting: ((signal: SessionSignal) => void) | null = null
let disposed = false

const push = (signal: SessionSignal): void => {
if (waiting) {
const resolve = waiting
waiting = null
resolve(signal)
return
}
queued.push(signal)
}

const unsubscribe = sseAggregator.onEvent((eventDirectory, event) => {
if (eventDirectory !== directory) {
Expand All @@ -339,24 +332,37 @@ function createSessionMonitor(directory: string, sessionId: string): SessionMoni
}

if (event.type === 'session.error') {
errorText = getSessionErrorText(event) ?? 'The session reported an unknown error.'
push({ errorText: getSessionErrorText(event) ?? 'The session reported an unknown error.', disposed: false })
return
}

if (event.type === 'session.idle') {
idle = true
return
}

if (event.type === 'session.status' && getSessionStatusType(event) === 'idle') {
idle = true
if (event.type === 'session.idle' || (event.type === 'session.status' && getSessionStatusType(event) === 'idle')) {
push({ errorText: null, disposed: false })
}
})

return {
getErrorText: () => errorText,
isIdle: () => idle,
dispose: unsubscribe,
markSubmitted: () => {
queued.length = 0
},
nextSignal: () => {
const next = queued.shift()
if (next) {
return Promise.resolve(next)
}
if (disposed) {
return Promise.resolve({ errorText: null, disposed: true })
}
return new Promise<SessionSignal>((resolve) => { waiting = resolve })
},
dispose: () => {
if (disposed) {
return
}
disposed = true
unsubscribe()
push({ errorText: null, disposed: true })
},
}
}

Expand All @@ -375,9 +381,27 @@ export class ScheduleService {
this.onJobChange = handler
}

getActiveRunSessionIds(): Set<string> {
const runs = listRunningScheduleRuns(this.db)
return new Set(runs.filter(r => r.sessionId).map(r => r.sessionId!))
getActiveRunSessions(): ScheduledSessionRef[] {
const refs: ScheduledSessionRef[] = []

for (const run of listRunningScheduleRuns(this.db)) {
if (!run.sessionId) continue

const directory = run.worktreePath ?? this.findRepoPath(run.repoId)
if (!directory) continue

refs.push({ sessionID: run.sessionId, directory })
}

return refs
}

private findRepoPath(repoId: number): string | null {
try {
return this.assertRepo(repoId).fullPath
} catch {
return null
}
}

listAllEnabledJobs(): ScheduleJob[] {
Expand Down Expand Up @@ -773,7 +797,7 @@ export class ScheduleService {
try {
const promptResponse = await this.openCodeClient.forward({
method: 'POST',
path: `/session/${input.sessionId}/message`,
path: `/session/${input.sessionId}/prompt_async`,
directory: input.directory,
body: JSON.stringify({
parts: [{ type: 'text', text: await buildPromptWithSkills(input.job.prompt, input.job.skillMetadata, input.directory, this.openCodeClient) }],
Expand All @@ -787,40 +811,7 @@ export class ScheduleService {
throw new ScheduleServiceError(errorText || 'Failed to run scheduled prompt', 502)
}

const promptBody = await promptResponse.text()
const promptResult = parsePromptResponse(promptBody)

if (promptResult) {
const currentRun = getScheduleRunById(this.db, input.repoId, input.job.id, input.runId)
if (!currentRun || currentRun.status !== 'running') {
return
}

const finishedAt = Date.now()
const responseText = extractResponseText(promptResult)
updateScheduleRun(this.db, input.repoId, input.job.id, input.runId, {
status: 'completed',
finishedAt,
sessionId: input.sessionId,
sessionTitle: input.sessionTitle,
responseText,
logText: buildRunLog({
job: input.job,
triggerSource: input.triggerSource,
sessionId: input.sessionId,
sessionTitle: input.sessionTitle,
responseText,
finishedAt,
}),
})

updateScheduleJobRunState(this.db, input.repoId, input.job.id, {
lastRunAt: finishedAt,
nextRunAt: input.triggerSource === 'manual' ? input.job.nextRunAt : computeNextRunAtForJob(input.job, finishedAt),
})

return
}
input.sessionMonitor.markSubmitted()

await this.monitorRunCompletion({
sessionMonitor: input.sessionMonitor,
Expand Down Expand Up @@ -905,9 +896,8 @@ export class ScheduleService {
}

const repo = this.assertRepo(input.repoId)
const currentMessages = await this.listSessionMessages(input.directory, input.sessionId)
const currentAssistantState = getAssistantMessageState(currentMessages)
if (currentAssistantState?.completed || currentAssistantState?.errorText) {
const currentAssistantState = await this.readSettledAssistantState(input.directory, input.sessionId)
if (currentAssistantState) {
await this.finalizeRecoveredRun(input.job, {
id: input.runId,
repoId: input.repoId,
Expand All @@ -923,7 +913,7 @@ export class ScheduleService {
return
}

const response = await this.waitForAssistantMessage(input.job, input.sessionId, input.sessionMonitor, input.directory)
const response = await this.waitForAssistantMessage(input.sessionId, input.sessionMonitor, input.directory)
const currentRun = getScheduleRunById(this.db, input.repoId, input.job.id, input.runId)
if (!currentRun || currentRun.status !== 'running') {
return
Expand Down Expand Up @@ -1131,46 +1121,62 @@ export class ScheduleService {
}
}

/**
* An assistant message completing is not the end of a run: a multi-step agent
* settles one message per tool call. Only a session that has gone idle has
* finished, and only then is the final message guaranteed to carry its parts.
*/
Comment on lines +1124 to +1128

@coderabbitai coderabbitai Bot Aug 10, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win

Remove the added block comment.

The coding guidelines forbid comments in TypeScript files. Encode the intent in the method name instead, for example readOutcomeOnlyWhenSessionIdle, or keep readAssistantOutcome and rely on the explicit AssistantOutcome union.

As per coding guidelines: "Do not add comments; code should be self-documenting."

🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@backend/src/services/schedules.ts` around lines 1124 - 1128, Remove the added
block comment near the assistant outcome handling. Preserve its intent through a
self-documenting method name such as readOutcomeOnlyWhenSessionIdle, or retain
readAssistantOutcome while relying on the explicit AssistantOutcome union.

Source: Coding guidelines

private async readAssistantOutcome(directory: string, sessionId: string): Promise<AssistantOutcome> {
const sessionStatus = (await this.getSessionStatuses(directory))[sessionId]
if (sessionStatus && sessionStatus.type !== 'idle') {
return { kind: 'busy' }
}

const assistantState = getAssistantMessageState(await this.listSessionMessages(directory, sessionId))
if (assistantState?.completed || assistantState?.errorText) {
return { kind: 'settled', responseText: assistantState.responseText, errorText: assistantState.errorText }
}

return { kind: 'stopped', responseText: assistantState?.responseText ?? null }
}

private async readSettledAssistantState(
directory: string,
sessionId: string,
): Promise<{ responseText: string | null; errorText: string | null } | null> {
const outcome = await this.readAssistantOutcome(directory, sessionId)
return outcome.kind === 'settled'
? { responseText: outcome.responseText, errorText: outcome.errorText }
: null
}

private async waitForAssistantMessage(
job: ScheduleJob,
sessionId: string,
sessionMonitor: SessionMonitor,
directory: string,
): Promise<{ responseText: string | null; errorText: string | null }> {
const startedAt = Date.now()

while (Date.now() - startedAt < RUN_POLL_TIMEOUT_MS) {
const messages = await this.listSessionMessages(directory, sessionId)
const assistantState = getAssistantMessageState(messages)
for (;;) {
const signal = await sessionMonitor.nextSignal()

if (assistantState && (assistantState.completed || assistantState.errorText)) {
if (signal.errorText || signal.disposed) {
const messages = await this.listSessionMessages(directory, sessionId)
return {
responseText: assistantState.responseText,
errorText: assistantState.errorText,
responseText: getAssistantMessageState(messages)?.responseText ?? null,
errorText: signal.errorText ?? SESSION_STOPPED_ERROR,
}
}

const sessionErrorText = sessionMonitor.getErrorText()
if (sessionErrorText) {
return {
responseText: null,
errorText: sessionErrorText,
}
}
const outcome = await this.readAssistantOutcome(directory, sessionId)

if (sessionMonitor.isIdle()) {
return {
responseText: null,
errorText: 'The session became idle without producing an assistant response. Open the linked session to inspect any pending questions, permissions, or provider issues.',
}
if (outcome.kind === 'busy') {
continue
}

await Bun.sleep(RUN_POLL_INTERVAL_MS)
}
if (outcome.kind === 'settled') {
return { responseText: outcome.responseText, errorText: outcome.errorText }
}

return {
responseText: null,
errorText: 'Timed out waiting for the assistant response. Open the linked session to inspect any pending questions, permissions, or provider issues.',
return { responseText: outcome.responseText, errorText: SESSION_STOPPED_ERROR }
}
}
Comment on lines 1153 to 1181

@coderabbitai coderabbitai Bot Aug 10, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🩺 Stability & Availability | 🟠 Major | ⚡ Quick win

waitForAssistantMessage can wait forever.

The previous implementation used fixed-interval polling with a timeout. The new loop blocks on sessionMonitor.nextSignal() and only resumes when a session.idle, session.status idle, or session.error event arrives for the directory, or when dispose() runs. dispose() only runs in the finally of monitorRunCompletion, which is reached after this loop returns, so it cannot unblock the loop.

If the upstream event for the session is never delivered, for example when the OpenCode process for that directory dies and the aggregator never emits a further status for the session, the run stays running forever. ScheduleService.activeRuns keeps the job id, so runJob then rejects every later trigger with "Schedule is already running".

Add a bounded fallback. One option is a maximum wait that re-reads the session state, another is a periodic re-check that calls readAssistantOutcome even without a signal.

🛡️ Sketch of a bounded wait 
-    for (;;) {
-      const signal = await sessionMonitor.nextSignal()
+    for (;;) {
+      const signal = await Promise.race([
+        sessionMonitor.nextSignal(),
+        Bun.sleep(SESSION_SIGNAL_RECHECK_MS).then((): SessionSignal | null => null),
+      ])
+
+      if (!signal) {
+        const polled = await this.readAssistantOutcome(directory, sessionId)
+        if (polled.kind === 'busy') continue
+        if (polled.kind === 'settled') {
+          return { responseText: polled.responseText, errorText: polled.errorText }
+        }
+        return { responseText: polled.responseText, errorText: SESSION_STOPPED_ERROR }
+      }

Confirm whether another component already bounds scheduled run duration.

#!/bin/bash
# Description: Look for any timeout, watchdog, or stale-run reaper covering schedule runs.
set -euo pipefail

rg -nP --type=ts -C4 '\b(setTimeout|AbortSignal\.timeout|TIMEOUT|_MS)\b' backend/src/services/schedules.ts backend/src/services/schedule-worktree.ts 2>/dev/null || true

fd -e ts . backend/src --exec rg -nP -C4 'stale|watchdog|reap|maxRunDuration|runTimeout' {} \;

rg -nP --type=ts -C4 '\bactiveRuns\b' backend/src
🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@backend/src/services/schedules.ts` around lines 1153 - 1181, Bound
waitForAssistantMessage so it cannot remain blocked indefinitely when
sessionMonitor.nextSignal() never emits another event. Add a periodic timeout or
re-check that invokes readAssistantOutcome and preserves the existing busy,
settled, and stopped result handling, ensuring the wait eventually returns an
error or terminal outcome and allows cleanup to remove the run from activeRuns.


Expand Down
Loading