Skip to content

feat: move scheduled jobs and run history to the server #2538

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
Dani Akash (DaniAkash) merged 4 commits into from
Sep 3, 2026
Merged

feat: move scheduled jobs and run history to the server #2538

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
959ff4a
feat(server): add local storage for scheduled job runs
DaniAkash Sep 3, 2026
ea29360
feat(server): carry the per-job run cap across with the runs
DaniAkash Sep 3, 2026
ae9ce3e
feat(app): read and write scheduled jobs and runs through the server
DaniAkash Sep 3, 2026
94191da
fix(app): record a finished run against the current job
DaniAkash Sep 3, 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
115 changes: 54 additions & 61 deletions packages/browseros-agent/apps/app/entrypoints/background/scheduledJobRuns.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,45 +1,53 @@
import { onScheduleMessage } from '@/lib/messaging/schedules/scheduleMessages'
import { createAlarmFromJob } from '@/lib/schedules/createAlarmFromJob'
import { getChatServerResponse } from '@/lib/schedules/getChatServerResponse'
import type {
ScheduledJob,
ScheduledJobRun,
} from '@/lib/schedules/scheduleTypes'
import {
scheduledJobRunStorage,
scheduledJobStorage,
} from '@/lib/schedules/scheduleStorage'
import type { ScheduledJobRun } from '@/lib/schedules/scheduleTypes'
listScheduledJobRunsOrNull,
listScheduledJobsOrNull,
putScheduledJob,
putScheduledJobRun,
} from '@/modules/schedules/schedules.api'

const MAX_RUNS_PER_JOB = 15
const STALE_TIMEOUT_MS = 10 * 60 * 1000 // 10 minutes
const TWENTY_FOUR_HOURS_MS = 24 * 60 * 60 * 1000

const runAbortControllers = new Map<string, AbortController>()

export const scheduledJobRuns = async () => {
// Every read below distinguishes an unreachable server from an empty list.
// Treating the two alike would look like "nothing is scheduled": alarms would
// not be rebuilt on startup and schedules would quietly stop firing, with no
// failed run to show for it. Skipping the pass instead leaves the next
// startup to retry.
const cleanupStaleJobRuns = async () => {
const current = (await scheduledJobRunStorage.getValue()) ?? []
const current = await listScheduledJobRunsOrNull()
if (current === null) return
const now = Date.now()

const updated = current.map((run) => {
if (run.status !== 'running') return run

const startedAt = new Date(run.startedAt).getTime()
if (now - startedAt > STALE_TIMEOUT_MS) {
return {
...run,
status: 'failed' as const,
completedAt: new Date().toISOString(),
result: 'Job timed out!',
}
}
return run
})
const stale = current.filter(
(run) =>
run.status === 'running' &&
now - new Date(run.startedAt).getTime() > STALE_TIMEOUT_MS,
)

await scheduledJobRunStorage.setValue(updated)
for (const run of stale) {
await putScheduledJobRun({
...run,
status: 'failed',
completedAt: new Date().toISOString(),
result: 'Job timed out!',
})
}
}

const syncAlarmState = async () => {
const jobs = (await scheduledJobStorage.getValue()).filter(
(each) => each.enabled,
)
const loaded = await listScheduledJobsOrNull()
if (loaded === null) return
const jobs = loaded.filter((each) => each.enabled)

for (let i = 0; i < jobs.length; i++) {
const job = jobs[i]
Expand All @@ -56,55 +64,39 @@ export const scheduledJobRuns = async () => {
jobId: string,
status: ScheduledJobRun['status'],
): Promise<ScheduledJobRun> => {
// Trimming to the per-job cap happens on the server now, so creating a run
// no longer has to rewrite the job's whole history to stay bounded.
const jobRun: ScheduledJobRun = {
id: crypto.randomUUID(),
jobId,
startedAt: new Date().toISOString(),
status,
}

const current = (await scheduledJobRunStorage.getValue()) ?? []
const otherJobRuns = current.filter((r) => r.jobId !== jobId)
const thisJobRuns = current
.filter((r) => r.jobId === jobId)
.sort(
(a, b) =>
new Date(b.startedAt).getTime() - new Date(a.startedAt).getTime(),
)
.slice(0, MAX_RUNS_PER_JOB - 1)

await scheduledJobRunStorage.setValue([
...otherJobRuns,
...thisJobRuns,
jobRun,
])
await putScheduledJobRun(jobRun)
return jobRun
}

// Takes the run rather than its id: the caller already holds it, and merging
// locally avoids re-reading a list to update one row.
const updateJobRun = async (
runId: string,
run: ScheduledJobRun,
updates: Partial<Omit<ScheduledJobRun, 'id' | 'jobId' | 'startedAt'>>,
) => {
const current = (await scheduledJobRunStorage.getValue()) ?? []
await scheduledJobRunStorage.setValue(
current.map((r) => (r.id === runId ? { ...r, ...updates } : r)),
)
await putScheduledJobRun({ ...run, ...updates })
}

const updateJobLastRunAt = async (jobId: string) => {
const current = (await scheduledJobStorage.getValue()) ?? []
await scheduledJobStorage.setValue(
current.map((j) =>
j.id === jobId ? { ...j, lastRunAt: new Date().toISOString() } : j,
),
)
const updateJobLastRunAt = async (job: ScheduledJob) => {
await putScheduledJob({ ...job, lastRunAt: new Date().toISOString() })
}
Comment thread
DaniAkash marked this conversation as resolved.
Outdated

const executeScheduledJob = async (jobId: string): Promise<void> => {
const job = (await scheduledJobStorage.getValue()).find(
(each) => each.id === jobId,
)
const jobs = await listScheduledJobsOrNull()
if (jobs === null) {
throw new Error('Cannot reach the BrowserOS server to load the job')
}

const job = jobs.find((each) => each.id === jobId)
if (!job) {
throw new Error(`Job not found: ${jobId}`)
}
Expand All @@ -120,7 +112,7 @@ export const scheduledJobRuns = async () => {
providerId: job.providerId,
})

await updateJobRun(jobRun.id, {
await updateJobRun(jobRun, {
status: 'completed',
completedAt: new Date().toISOString(),
result: response.text,
Expand All @@ -135,15 +127,15 @@ export const scheduledJobRuns = async () => {
: e instanceof Error
? e.message
: String(e)
await updateJobRun(jobRun.id, {
await updateJobRun(jobRun, {
status: 'failed',
completedAt: new Date().toISOString(),
result: errorMessage,
error: errorMessage,
})
} finally {
runAbortControllers.delete(jobRun.id)
await updateJobLastRunAt(jobId)
await updateJobLastRunAt(job)
}
}

Expand All @@ -155,10 +147,11 @@ export const scheduledJobRuns = async () => {
runningMissedJobs = true

try {
const jobs = (await scheduledJobStorage.getValue()).filter(
(j) => j.enabled,
)
const runs = (await scheduledJobRunStorage.getValue()) ?? []
const loadedJobs = await listScheduledJobsOrNull()
const runs = await listScheduledJobRunsOrNull()
if (loadedJobs === null || runs === null) return

const jobs = loadedJobs.filter((j) => j.enabled)
const now = Date.now()
const cutoff = now - TWENTY_FOUR_HOURS_MS

Expand Down
149 changes: 6 additions & 143 deletions packages/browseros-agent/apps/app/lib/schedules/scheduleStorage.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,10 +1,12 @@
import { storage } from '@wxt-dev/storage'
import { useEffect, useState } from 'react'
import { sendScheduleMessage } from '@/lib/messaging/schedules/scheduleMessages'
import { createAlarmFromJob } from './createAlarmFromJob'
import type { ScheduledJob, ScheduledJobRun } from './scheduleTypes'

const getAlarmName = (jobId: string) => `scheduled-job-${jobId}`
/**
* Legacy extension storage for scheduled jobs and their runs.
*
* The server owns both now. These items remain only as the source the one-time
* import reads, and nothing writes them any more.
*/

export const scheduledJobStorage = storage.defineItem<ScheduledJob[]>(
'local:scheduledJobs',
Expand All @@ -19,142 +21,3 @@ export const scheduledJobRunStorage = storage.defineItem<ScheduledJobRun[]>(
fallback: [],
},
)

export const pendingDeletionStorage = storage.defineItem<string[]>(
'local:scheduledJobsPendingDeletion',
{
fallback: [],
},
)

export function useScheduledJobs() {
const [jobs, setJobs] = useState<ScheduledJob[]>([])

useEffect(() => {
scheduledJobStorage.getValue().then(setJobs)
const unwatch = scheduledJobStorage.watch((newValue) => {
setJobs(newValue ?? [])
})
return unwatch
}, [])

const addJob = async (
job: Omit<ScheduledJob, 'id' | 'createdAt' | 'updatedAt'>,
) => {
const now = new Date().toISOString()
const newJob: ScheduledJob = {
id: crypto.randomUUID(),
createdAt: now,
updatedAt: now,
...job,
}
const current = (await scheduledJobStorage.getValue()) ?? []
await scheduledJobStorage.setValue([...current, newJob])

if (newJob.enabled) {
await createAlarmFromJob(newJob)
}
}

const removeJob = async (id: string) => {
await chrome.alarms.clear(getAlarmName(id))

const pending = (await pendingDeletionStorage.getValue()) ?? []
if (!pending.includes(id)) {
await pendingDeletionStorage.setValue([...pending, id])
}

const currentJobs = (await scheduledJobStorage.getValue()) ?? []
await scheduledJobStorage.setValue(currentJobs.filter((j) => j.id !== id))

const currentRuns = (await scheduledJobRunStorage.getValue()) ?? []
await scheduledJobRunStorage.setValue(
currentRuns.filter((r) => r.jobId !== id),
)
}

const toggleJob = async (id: string, enabled: boolean) => {
const current = (await scheduledJobStorage.getValue()) ?? []
const job = current.find((j) => j.id === id)
if (!job) return

const updatedAt = new Date().toISOString()
await scheduledJobStorage.setValue(
current.map((j) => (j.id === id ? { ...j, enabled, updatedAt } : j)),
)

if (enabled) {
await createAlarmFromJob({ ...job, enabled })
} else {
await chrome.alarms.clear(getAlarmName(id))
}
}

const editJob = async (
id: string,
updates: Omit<ScheduledJob, 'id' | 'createdAt' | 'updatedAt'>,
) => {
const current = (await scheduledJobStorage.getValue()) ?? []
const existingJob = current.find((j) => j.id === id)
if (!existingJob) return

const updatedJob: ScheduledJob = {
id,
createdAt: existingJob.createdAt,
updatedAt: new Date().toISOString(),
...updates,
}
await scheduledJobStorage.setValue(
current.map((j) => (j.id === id ? updatedJob : j)),
)

await chrome.alarms.clear(getAlarmName(id))
if (updatedJob.enabled) {
await createAlarmFromJob(updatedJob)
}
}

const runJob = async (id: string) => {
return sendScheduleMessage('runScheduledJob', { jobId: id })
}

return { jobs, addJob, removeJob, editJob, toggleJob, runJob }
}

export function useScheduledJobRuns() {
const [jobRuns, setJobRuns] = useState<ScheduledJobRun[]>([])

useEffect(() => {
scheduledJobRunStorage.getValue().then(setJobRuns)
const unwatch = scheduledJobRunStorage.watch((newValue) => {
setJobRuns(newValue ?? [])
})
return unwatch
}, [])

const addJobRun = async (jobRun: ScheduledJobRun) => {
const current = (await scheduledJobRunStorage.getValue()) ?? []
await scheduledJobRunStorage.setValue([...current, jobRun])
}

const removeJobRun = async (id: string) => {
const current = (await scheduledJobRunStorage.getValue()) ?? []
await scheduledJobRunStorage.setValue(current.filter((r) => r.id !== id))
}

const editJobRun = async (
id: string,
updates: Partial<Omit<ScheduledJobRun, 'id'>>,
) => {
const current = (await scheduledJobRunStorage.getValue()) ?? []
await scheduledJobRunStorage.setValue(
current.map((r) => (r.id === id ? { ...r, ...updates } : r)),
)
}

const cancelJobRun = async (runId: string) => {
return sendScheduleMessage('cancelScheduledJobRun', { runId })
}

return { jobRuns, addJobRun, removeJobRun, editJobRun, cancelJobRun }
}
Loading
Loading