feat(android): add safe cron job management (#102997)

* feat(android): add safe cron job management

Co-authored-by: snowzlmbot <293528334+snowzlmbot@users.noreply.github.com>

* fix(android): harden cron editor state

* fix(android): preserve cron state across lifecycle

* fix(android): satisfy cron release gates

* fix(android): retain cron drafts outside saved state

* fix(android): scope cron auto-delete to one-shot jobs

* fix(android): use Compose activity owner

* docs(changelog): note Android cron management

* fix(cron): harden Android job management

* chore(i18n): refresh Android cron inventory

* test(cron): cover enriched read views

* chore(changelog): defer Android cron note to release

---------

Co-authored-by: snowzlmbot <293528334+snowzlmbot@users.noreply.github.com>
This commit is contained in:
Peter Steinberger
2026-07-10 04:12:18 +01:00
committed by GitHub
parent e8fcc93cd3
commit 1696366f88
36 changed files with 4470 additions and 284 deletions
File diff suppressed because it is too large Load Diff
+1
View File
@@ -20,6 +20,7 @@ OpenClaw Android is the officially released Google Play app. It connects to an O
- [x] Screen tab full functionality
- [x] Skill Workshop settings can filter proposals, inspect proposal content, and apply/reject/quarantine drafts through Gateway RPCs
- [x] Per-app language selection for translated resources follows Android system settings and persistence
- [x] Cron job settings support details, run history, run now, edits, enable/disable, and deletion with admin-scoped Gateway access
## Open in Android Studio
@@ -8,6 +8,8 @@ import kotlinx.serialization.json.buildJsonObject
internal object AndroidScreenshotFixture {
const val mainSessionKey = "agent:main:node-screenshot"
const val primarySessionTitle = "Android release planning"
const val cronJobId = "android-release-digest"
const val cronJobName = "Android release digest"
val agents =
listOf(
@@ -93,9 +95,97 @@ internal object AndroidScreenshotFixture {
"chat.history" -> chatHistory()
"sessions.list" -> sessionList()
"chat.metadata" -> chatMetadata()
"cron.list" -> cronList()
"cron.get" -> cronJob().toString()
"cron.runs" -> cronRuns()
else -> error("Screenshot fixture does not implement gateway method $method with params $paramsJson")
}
private fun cronList(): String =
buildJsonObject {
put(
"jobs",
buildJsonArray {
add(cronJob())
},
)
}.toString()
private fun cronJob() =
buildJsonObject {
put("id", JsonPrimitive(cronJobId))
put("name", JsonPrimitive(cronJobName))
put("enabled", JsonPrimitive(true))
put("createdAtMs", JsonPrimitive(1_783_468_800_000))
put("updatedAtMs", JsonPrimitive(1_783_555_200_000))
put("configRevision", JsonPrimitive("sha256:screenshot-fixture"))
put(
"schedule",
buildJsonObject {
put("kind", JsonPrimitive("every"))
put("everyMs", JsonPrimitive(86_400_000))
put("anchorMs", JsonPrimitive(1_783_468_800_000))
},
)
put("sessionTarget", JsonPrimitive("isolated"))
put("wakeMode", JsonPrimitive("now"))
put(
"payload",
buildJsonObject {
put("kind", JsonPrimitive("agentTurn"))
put("message", JsonPrimitive("Summarize Android release readiness."))
put("model", JsonPrimitive("openai/gpt-5.2"))
},
)
put(
"state",
buildJsonObject {
put("nextRunAtMs", JsonPrimitive(1_783_641_600_000))
put("lastRunAtMs", JsonPrimitive(1_783_555_200_000))
put("lastStatus", JsonPrimitive("ok"))
put("lastDurationMs", JsonPrimitive(1_842))
put("consecutiveErrors", JsonPrimitive(0))
put("consecutiveSkipped", JsonPrimitive(0))
put("lastDeliveryStatus", JsonPrimitive("delivered"))
},
)
}
private fun cronRuns(): String =
buildJsonObject {
put(
"entries",
buildJsonArray {
add(
buildJsonObject {
put("ts", JsonPrimitive(1_783_555_200_000))
put("jobId", JsonPrimitive(cronJobId))
put("runId", JsonPrimitive("android-release-digest-run-2"))
put("action", JsonPrimitive("finished"))
put("status", JsonPrimitive("ok"))
put("summary", JsonPrimitive("Release checklist ready"))
put("durationMs", JsonPrimitive(1_842))
put("deliveryStatus", JsonPrimitive("delivered"))
put("model", JsonPrimitive("openai/gpt-5.2"))
},
)
add(
buildJsonObject {
put("ts", JsonPrimitive(1_783_468_800_000))
put("jobId", JsonPrimitive(cronJobId))
put("runId", JsonPrimitive("android-release-digest-run-1"))
put("action", JsonPrimitive("finished"))
put("status", JsonPrimitive("error"))
put("error", JsonPrimitive("Play publish blocked"))
put("durationMs", JsonPrimitive(927))
put("deliveryStatus", JsonPrimitive("not-requested"))
put("model", JsonPrimitive("openai/gpt-5.2"))
},
)
},
)
}.toString()
private fun chatHistory(): String =
buildJsonObject {
put("sessionId", JsonPrimitive("screenshot-session"))
@@ -15,22 +15,37 @@ data class GatewayCronJobDetail(
val description: String,
val enabled: Boolean,
val deleteAfterRun: Boolean,
val scheduleKind: String,
val scheduleLabel: String,
val scheduleDetail: String,
val scheduleAt: String?,
val scheduleEveryMs: Long?,
val scheduleAnchorMs: Long?,
val scheduleCronExpr: String?,
val scheduleTimezone: String?,
val scheduleStaggerMs: Long?,
val scheduleCommand: String?,
val scheduleCwd: String?,
val sessionTarget: String,
val wakeMode: String,
val payloadKind: String,
val payloadText: String?,
val payloadLabel: String,
val payloadModel: String?,
val payloadThinking: String?,
val payloadCommandArgv: List<String>?,
val payloadCommandCwd: String?,
val deliveryLabel: String,
val failureAlertLabel: String,
val createdAtMs: Long,
val updatedAtMs: Long,
val configRevision: String?,
val nextRunAtMs: Long?,
val runningAtMs: Long?,
val lastRunAtMs: Long?,
val lastRunStatus: String?,
val lastError: String?,
val lastDiagnosticSummary: String?,
val lastDurationMs: Long?,
val consecutiveErrors: Long?,
val consecutiveSkipped: Long?,
@@ -75,6 +90,18 @@ internal class CronJobDetailRequestGuard {
}
}
fun beginIfCurrent(
rawId: String,
onBegin: (CronJobDetailRequest) -> Unit,
): CronJobDetailRequest? {
val id = rawId.trim().takeIf { it.isNotEmpty() } ?: return null
return synchronized(lock) {
if (selectedId != id) return@synchronized null
generation += 1
CronJobDetailRequest(id = id, generation = generation).also(onBegin)
}
}
fun cancel(onCancel: () -> Unit = {}) {
synchronized(lock) {
generation += 1
@@ -83,6 +110,20 @@ internal class CronJobDetailRequestGuard {
}
}
fun cancelIfCurrent(
rawId: String,
onCancel: () -> Unit,
): Boolean {
val id = rawId.trim().takeIf { it.isNotEmpty() } ?: return false
return synchronized(lock) {
if (selectedId != id) return@synchronized false
generation += 1
selectedId = null
onCancel()
true
}
}
fun publishIfCurrent(
request: CronJobDetailRequest,
publish: () -> Unit,
@@ -110,6 +151,9 @@ internal fun parseGatewayCronJobDetail(job: JsonObject?): GatewayCronJobDetail?
val sessionTarget = value.string("sessionTarget") ?: return null
val wakeMode = value.string("wakeMode") ?: return null
val payloadKind = payload.string("kind") ?: return null
val scheduleKind = schedule.string("kind") ?: return null
if (scheduleKind !in setOf("at", "every", "cron", "on-exit")) return null
if (payloadKind !in setOf("systemEvent", "agentTurn", "command")) return null
val state = value["state"].asObjectOrNull() ?: return null
return GatewayCronJobDetail(
@@ -118,22 +162,39 @@ internal fun parseGatewayCronJobDetail(job: JsonObject?): GatewayCronJobDetail?
description = value.string("description").orEmpty(),
enabled = value.boolean("enabled"),
deleteAfterRun = value.boolean("deleteAfterRun"),
scheduleKind = scheduleKind,
scheduleLabel = cronScheduleLabel(schedule),
scheduleDetail = cronScheduleDetail(schedule),
scheduleAt = schedule.string("at"),
scheduleEveryMs = schedule.long("everyMs"),
scheduleAnchorMs = schedule.long("anchorMs"),
scheduleCronExpr = schedule.string("expr"),
scheduleTimezone = schedule.string("tz"),
scheduleStaggerMs = schedule.long("staggerMs"),
scheduleCommand = schedule.string("command"),
scheduleCwd = schedule.string("cwd"),
sessionTarget = sessionTarget,
wakeMode = wakeMode,
payloadKind = payloadKind,
payloadText = cronPayloadText(payload),
payloadLabel = cronPayloadLabel(payload),
payloadModel = payload.string("model"),
payloadThinking = payload.string("thinking"),
payloadCommandArgv =
(payload["argv"] as? JsonArray)
?.mapNotNull { it.asStringOrNull() },
payloadCommandCwd = payload.string("cwd"),
deliveryLabel = cronDeliveryLabel(value["delivery"].asObjectOrNull()),
failureAlertLabel = cronFailureAlertLabel(value["failureAlert"]),
createdAtMs = createdAtMs,
updatedAtMs = updatedAtMs,
configRevision = value.string("configRevision"),
nextRunAtMs = state.long("nextRunAtMs"),
runningAtMs = state.long("runningAtMs"),
lastRunAtMs = state.long("lastRunAtMs"),
lastRunStatus = cronJobLastRunStatus(state),
lastError = state.string("lastError"),
lastDiagnosticSummary = state.string("lastDiagnosticSummary"),
lastDurationMs = state.long("lastDurationMs"),
consecutiveErrors = state.long("consecutiveErrors"),
consecutiveSkipped = state.long("consecutiveSkipped"),
@@ -0,0 +1,611 @@
package ai.openclaw.app
import ai.openclaw.app.gateway.GatewaySession
import ai.openclaw.app.node.asObjectOrNull
import ai.openclaw.app.node.asStringOrNull
import kotlinx.serialization.json.Json
import kotlinx.serialization.json.JsonArray
import kotlinx.serialization.json.JsonNull
import kotlinx.serialization.json.JsonObject
import kotlinx.serialization.json.JsonPrimitive
import kotlinx.serialization.json.booleanOrNull
import kotlinx.serialization.json.buildJsonObject
data class GatewayCronRunSummary(
val ts: Long,
val runId: String?,
val status: String?,
val summary: String?,
val error: String?,
val durationMs: Long?,
val deliveryStatus: String?,
val sessionKey: String?,
val model: String?,
)
sealed interface GatewayCronRunHistoryState {
data object Idle : GatewayCronRunHistoryState
data class Loading(
val id: String,
) : GatewayCronRunHistoryState
data class Loaded(
val id: String,
val runs: List<GatewayCronRunSummary>,
) : GatewayCronRunHistoryState
data class Error(
val id: String,
val message: String,
) : GatewayCronRunHistoryState
}
enum class GatewayCronAction {
Run,
Enable,
Disable,
Save,
Delete,
}
enum class GatewayCronNoticeKind {
Success,
Warning,
Error,
}
sealed interface GatewayCronActionState {
data object Idle : GatewayCronActionState
data class Running(
val id: String,
val action: GatewayCronAction,
) : GatewayCronActionState
data class Notice(
val id: String,
val message: String,
val kind: GatewayCronNoticeKind,
val deleted: Boolean = false,
) : GatewayCronActionState
}
/** Owns one queued manual run id per job so a stale tracker cannot clear a newer run. */
internal class PendingCronRunRegistry {
private val lock = Any()
private val runIdsByJob = linkedMapOf<String, String>()
fun contains(rawJobId: String): Boolean {
val jobId = rawJobId.trim().takeIf { it.isNotEmpty() } ?: return false
return synchronized(lock) { runIdsByJob.containsKey(jobId) }
}
fun begin(
rawJobId: String,
rawRunId: String,
publish: (Set<String>) -> Unit,
): Boolean {
val jobId = rawJobId.trim().takeIf { it.isNotEmpty() } ?: return false
val runId = rawRunId.trim().takeIf { it.isNotEmpty() } ?: return false
return synchronized(lock) {
if (runIdsByJob.containsKey(jobId)) return@synchronized false
runIdsByJob[jobId] = runId
publish(runIdsByJob.keys.toSet())
true
}
}
fun finish(
rawJobId: String,
rawRunId: String,
publish: (Set<String>) -> Unit,
): Boolean {
val jobId = rawJobId.trim().takeIf { it.isNotEmpty() } ?: return false
val runId = rawRunId.trim().takeIf { it.isNotEmpty() } ?: return false
return synchronized(lock) {
if (runIdsByJob[jobId] != runId) return@synchronized false
runIdsByJob.remove(jobId)
publish(runIdsByJob.keys.toSet())
true
}
}
fun clear(publish: (Set<String>) -> Unit) {
synchronized(lock) {
runIdsByJob.clear()
publish(emptySet())
}
}
}
sealed interface GatewayCronScheduleEdit {
data class At(
val at: String,
) : GatewayCronScheduleEdit
data class Every(
val everyMs: String,
val anchorMs: String,
) : GatewayCronScheduleEdit
data class Cron(
val expression: String,
val timezone: String,
val staggerMs: String,
) : GatewayCronScheduleEdit
data class OnExit(
val command: String,
val cwd: String,
) : GatewayCronScheduleEdit
}
sealed interface GatewayCronPayloadEdit {
data class SystemEvent(
val text: String,
) : GatewayCronPayloadEdit
data class AgentTurn(
val message: String,
val model: String,
val thinking: String,
) : GatewayCronPayloadEdit
data class Command(
val argvJson: String,
val cwd: String,
) : GatewayCronPayloadEdit
}
data class GatewayCronJobEdit(
val name: String,
val description: String,
val enabled: Boolean,
val deleteAfterRun: Boolean,
val schedule: GatewayCronScheduleEdit,
val sessionTarget: String,
val wakeMode: String,
val payload: GatewayCronPayloadEdit,
) {
fun withSchedule(value: GatewayCronScheduleEdit): GatewayCronJobEdit =
copy(
schedule = value,
deleteAfterRun = deleteAfterRun && value is GatewayCronScheduleEdit.At,
)
}
internal data class CronEditorDraftState(
val baseline: GatewayCronJobEdit,
val edit: GatewayCronJobEdit,
val savePending: Boolean = false,
val saveSucceeded: Boolean = false,
val hasIncomingConflict: Boolean = false,
) {
val isDirty: Boolean
get() = edit != baseline
val requiresResolution: Boolean
get() = isDirty || hasIncomingConflict
fun withEdit(value: GatewayCronJobEdit): CronEditorDraftState = copy(edit = value)
fun saveStarted(): CronEditorDraftState = copy(savePending = true, saveSucceeded = false)
fun saveAborted(): CronEditorDraftState = copy(savePending = false, saveSucceeded = false)
fun observeSaveNotice(kind: GatewayCronNoticeKind): CronEditorDraftState {
if (!savePending) return this
return if (kind == GatewayCronNoticeKind.Success) {
copy(saveSucceeded = true)
} else {
copy(savePending = false, saveSucceeded = false)
}
}
fun observeJob(job: GatewayCronJobDetail): CronEditorDraftState {
val incoming = job.toCronJobEdit()
if (incoming == edit) {
return CronEditorDraftState(
baseline = incoming,
edit = incoming,
)
}
if (incoming == baseline) {
return copy(hasIncomingConflict = false)
}
val canAdopt = !isDirty || saveSucceeded
if (!canAdopt) {
return copy(hasIncomingConflict = true)
}
return CronEditorDraftState(
baseline = incoming,
edit = incoming,
)
}
companion object {
fun from(job: GatewayCronJobDetail): CronEditorDraftState {
val edit = job.toCronJobEdit()
return CronEditorDraftState(
baseline = edit,
edit = edit,
)
}
}
}
internal fun CronEditorDraftState.reconcileRestoredAction(
isConnected: Boolean,
jobId: String,
actionState: GatewayCronActionState,
): CronEditorDraftState {
if (!savePending) return this
// Activity recreation retains the runtime action; process death does not.
// Preserve pending only when the restored runtime still owns this Save.
val retainedSaveState =
when (actionState) {
is GatewayCronActionState.Running ->
actionState.id == jobId && actionState.action == GatewayCronAction.Save
is GatewayCronActionState.Notice -> actionState.id == jobId
GatewayCronActionState.Idle -> false
}
return if (isConnected && retainedSaveState) this else saveAborted()
}
internal enum class GatewayCronRunSkipReason(
val message: String,
) {
NotDue("Cron job is not due yet."),
AlreadyRunning("Cron job is already running."),
RestartRecoveryPending("Gateway restart recovery is still in progress."),
InvalidSpec("Cron job has an invalid configuration."),
Stopped("Cron scheduler is stopped."),
}
internal sealed interface GatewayCronRunOutcome {
data class Started(
val runId: String?,
) : GatewayCronRunOutcome
data class Skipped(
val reason: GatewayCronRunSkipReason,
) : GatewayCronRunOutcome
data object Rejected : GatewayCronRunOutcome
}
internal fun cronRunShouldRefresh(outcome: GatewayCronRunOutcome): Boolean =
when (outcome) {
is GatewayCronRunOutcome.Started -> true
is GatewayCronRunOutcome.Skipped -> outcome.reason == GatewayCronRunSkipReason.InvalidSpec
GatewayCronRunOutcome.Rejected -> false
}
internal fun cronRunCompletionNotice(
jobId: String,
status: String?,
): GatewayCronActionState.Notice {
val (message, kind) =
when (status) {
"ok" -> "Cron run finished." to GatewayCronNoticeKind.Success
"skipped" -> "Cron run skipped." to GatewayCronNoticeKind.Warning
"error" -> "Cron run failed." to GatewayCronNoticeKind.Error
else -> "Cron run finished with an unknown status." to GatewayCronNoticeKind.Warning
}
return GatewayCronActionState.Notice(id = jobId, message = message, kind = kind)
}
internal fun isCronJobRevisionConflict(error: GatewaySession.ErrorShape): Boolean = error.details?.code == "CRON_JOB_CHANGED"
internal fun GatewayCronJobDetail.toCronJobEdit(): GatewayCronJobEdit =
GatewayCronJobEdit(
name = name,
description = description,
enabled = enabled,
// Gateway deletion only runs after a successful one-shot schedule.
deleteAfterRun = deleteAfterRun && scheduleKind == "at",
schedule =
when (scheduleKind) {
"at" -> GatewayCronScheduleEdit.At(at = scheduleAt.orEmpty())
"every" ->
GatewayCronScheduleEdit.Every(
everyMs = scheduleEveryMs?.toString().orEmpty(),
anchorMs = scheduleAnchorMs?.toString().orEmpty(),
)
"cron" ->
GatewayCronScheduleEdit.Cron(
expression = scheduleCronExpr.orEmpty(),
timezone = scheduleTimezone.orEmpty(),
staggerMs = scheduleStaggerMs?.toString().orEmpty(),
)
"on-exit" ->
GatewayCronScheduleEdit.OnExit(
command = scheduleCommand.orEmpty(),
cwd = scheduleCwd.orEmpty(),
)
else -> error("Unsupported cron schedule kind: $scheduleKind")
},
sessionTarget = sessionTarget,
wakeMode = wakeMode,
payload =
when (payloadKind) {
"systemEvent" -> GatewayCronPayloadEdit.SystemEvent(text = payloadText.orEmpty())
"agentTurn" ->
GatewayCronPayloadEdit.AgentTurn(
message = payloadText.orEmpty(),
model = payloadModel.orEmpty(),
thinking = payloadThinking.orEmpty(),
)
"command" ->
GatewayCronPayloadEdit.Command(
argvJson = JsonArray(payloadCommandArgv.orEmpty().map(::JsonPrimitive)).toString(),
cwd = payloadCommandCwd.orEmpty(),
)
else -> error("Unsupported cron payload kind: $payloadKind")
},
)
internal fun buildCronUpdateParams(
original: GatewayCronJobDetail,
edit: GatewayCronJobEdit,
): String {
val name = edit.name.trim()
require(name.isNotEmpty()) { "Cron job name is required." }
val description = edit.description.trim()
val sessionTarget = edit.sessionTarget.trim()
require(
sessionTarget == "main" ||
sessionTarget == "isolated" ||
sessionTarget == "current" ||
(sessionTarget.startsWith("session:") && sessionTarget.removePrefix("session:").isNotBlank()),
) { "Session target must be main, isolated, current, or session:<id>." }
val wakeMode = edit.wakeMode.trim()
require(wakeMode == "now" || wakeMode == "next-heartbeat") {
"Wake mode must be now or next-heartbeat."
}
val schedulePatch = buildCronSchedulePatch(original = original, edit = edit.schedule)
val payloadPatch = buildCronPayloadPatch(original = original, edit = edit.payload)
val patch =
buildJsonObject {
if (name != original.name) put("name", JsonPrimitive(name))
if (description != original.description) put("description", JsonPrimitive(description))
if (edit.enabled != original.enabled) put("enabled", JsonPrimitive(edit.enabled))
if (edit.deleteAfterRun != original.deleteAfterRun) {
put("deleteAfterRun", JsonPrimitive(edit.deleteAfterRun))
}
schedulePatch?.let { put("schedule", it) }
if (sessionTarget != original.sessionTarget) {
put("sessionTarget", JsonPrimitive(sessionTarget))
}
if (wakeMode != original.wakeMode) put("wakeMode", JsonPrimitive(wakeMode))
payloadPatch?.let { put("payload", it) }
}
require(patch.isNotEmpty()) { "No cron changes to save." }
val configRevision =
requireNotNull(original.configRevision) {
"Update the gateway before saving cron changes from Android."
}
return buildJsonObject {
put("id", JsonPrimitive(original.id))
put("expectedConfigRevision", JsonPrimitive(configRevision))
put("patch", patch)
}.toString()
}
internal fun parseGatewayCronRunOutcome(root: JsonObject?): GatewayCronRunOutcome? {
val value = root ?: return null
val ok = value.optionalBoolean("ok") ?: return null
if (!ok) return GatewayCronRunOutcome.Rejected
if (value.optionalBoolean("ran") == true) {
return GatewayCronRunOutcome.Started(runId = value.string("runId"))
}
if (value.optionalBoolean("enqueued") == true) {
val runId = value.string("runId") ?: return null
return GatewayCronRunOutcome.Started(runId = runId)
}
if (value.optionalBoolean("ran") != false) return null
val reason =
when (value.string("reason")) {
"not-due" -> GatewayCronRunSkipReason.NotDue
"already-running" -> GatewayCronRunSkipReason.AlreadyRunning
"restart-recovery-pending" -> GatewayCronRunSkipReason.RestartRecoveryPending
"invalid-spec" -> GatewayCronRunSkipReason.InvalidSpec
"stopped" -> GatewayCronRunSkipReason.Stopped
else -> return null
}
return GatewayCronRunOutcome.Skipped(reason)
}
internal fun parseGatewayCronRunHistory(entries: JsonArray?): List<GatewayCronRunSummary> =
entries
?.mapNotNull { item ->
val value = item.asObjectOrNull() ?: return@mapNotNull null
val ts = value.long("ts") ?: return@mapNotNull null
GatewayCronRunSummary(
ts = ts,
runId = value.string("runId"),
status = value.string("status"),
summary = value.string("summary"),
error = value.string("error"),
durationMs = value.long("durationMs"),
deliveryStatus = value.string("deliveryStatus"),
sessionKey = value.string("sessionKey"),
model = value.string("model"),
)
}.orEmpty()
private fun buildCronSchedulePatch(
original: GatewayCronJobDetail,
edit: GatewayCronScheduleEdit,
): JsonObject? =
when (edit) {
is GatewayCronScheduleEdit.At -> {
require(original.scheduleKind == "at") { "Changing schedule type is not supported here." }
val at = edit.at.trim()
require(at.isNotEmpty()) { "One-time cron jobs need an ISO time." }
if (at == original.scheduleAt) {
null
} else {
buildJsonObject {
put("kind", JsonPrimitive("at"))
put("at", JsonPrimitive(at))
}
}
}
is GatewayCronScheduleEdit.Every -> {
require(original.scheduleKind == "every") { "Changing schedule type is not supported here." }
val everyMs = edit.everyMs.trim().toLongOrNull()
require(everyMs != null && everyMs > 0L) { "Interval must be a positive number of milliseconds." }
val anchorMs = parseOptionalNonNegativeLong(edit.anchorMs, "Anchor")
if (everyMs == original.scheduleEveryMs && anchorMs == original.scheduleAnchorMs) {
null
} else {
buildJsonObject {
put("kind", JsonPrimitive("every"))
put("everyMs", JsonPrimitive(everyMs))
anchorMs?.let { put("anchorMs", JsonPrimitive(it)) }
}
}
}
is GatewayCronScheduleEdit.Cron -> {
require(original.scheduleKind == "cron") { "Changing schedule type is not supported here." }
val expression = edit.expression.trim()
require(expression.isNotEmpty()) { "Cron expression is required." }
val timezone = edit.timezone.trim().ifEmpty { null }
val requestedStaggerMs = parseOptionalNonNegativeLong(edit.staggerMs, "Stagger")
val staggerMs =
requestedStaggerMs ?: if (original.scheduleStaggerMs != null) 0L else null
if (
expression == original.scheduleCronExpr &&
timezone == original.scheduleTimezone &&
staggerMs == original.scheduleStaggerMs
) {
null
} else {
buildJsonObject {
put("kind", JsonPrimitive("cron"))
put("expr", JsonPrimitive(expression))
timezone?.let { put("tz", JsonPrimitive(it)) }
staggerMs?.let { put("staggerMs", JsonPrimitive(it)) }
}
}
}
is GatewayCronScheduleEdit.OnExit -> {
require(original.scheduleKind == "on-exit") { "Changing schedule type is not supported here." }
val command = edit.command.trim()
require(command.isNotEmpty()) { "On-exit cron jobs need a command." }
val cwd = edit.cwd.trim().ifEmpty { null }
if (command == original.scheduleCommand && cwd == original.scheduleCwd) {
null
} else {
buildJsonObject {
put("kind", JsonPrimitive("on-exit"))
put("command", JsonPrimitive(command))
cwd?.let { put("cwd", JsonPrimitive(it)) }
}
}
}
}
private fun buildCronPayloadPatch(
original: GatewayCronJobDetail,
edit: GatewayCronPayloadEdit,
): JsonObject? =
when (edit) {
is GatewayCronPayloadEdit.SystemEvent -> {
require(original.payloadKind == "systemEvent") { "Changing payload type is not supported here." }
val text = edit.text.trim()
require(text.isNotEmpty()) { "System event text is required." }
if (text == original.payloadText) {
null
} else {
buildJsonObject {
put("kind", JsonPrimitive("systemEvent"))
put("text", JsonPrimitive(text))
}
}
}
is GatewayCronPayloadEdit.AgentTurn -> {
require(original.payloadKind == "agentTurn") { "Changing payload type is not supported here." }
val message = edit.message.trim()
require(message.isNotEmpty()) { "Agent message is required." }
val model = edit.model.trim().ifEmpty { null }
val thinking = edit.thinking.trim().ifEmpty { null }
if (
message == original.payloadText &&
model == original.payloadModel &&
thinking == original.payloadThinking
) {
null
} else {
buildJsonObject {
put("kind", JsonPrimitive("agentTurn"))
if (message != original.payloadText) put("message", JsonPrimitive(message))
if (model != original.payloadModel) put("model", model?.let(::JsonPrimitive) ?: JsonNull)
if (thinking != original.payloadThinking) {
put("thinking", thinking?.let(::JsonPrimitive) ?: JsonNull)
}
}
}
}
is GatewayCronPayloadEdit.Command -> {
require(original.payloadKind == "command") { "Changing payload type is not supported here." }
val argv = parseCommandArgv(edit.argvJson)
val cwd = edit.cwd.trim().ifEmpty { null }
if (cwd == null && original.payloadCommandCwd != null) {
error("The gateway does not support clearing a command working directory.")
}
if (argv == original.payloadCommandArgv && cwd == original.payloadCommandCwd) {
null
} else {
buildJsonObject {
put("kind", JsonPrimitive("command"))
if (argv != original.payloadCommandArgv) {
put("argv", JsonArray(argv.map(::JsonPrimitive)))
}
if (cwd != original.payloadCommandCwd) put("cwd", JsonPrimitive(requireNotNull(cwd)))
}
}
}
}
private fun parseCommandArgv(raw: String): List<String> {
val value =
runCatching { Json.parseToJsonElement(raw) }.getOrNull() as? JsonArray
?: error("Command argv must be a JSON array.")
val argv =
value.map { item ->
val primitive = item as? JsonPrimitive
primitive?.takeIf { it.isString }?.content?.takeIf { it.isNotEmpty() }
?: error("Command argv entries must be non-empty strings.")
}
require(argv.isNotEmpty()) { "Command argv must contain at least one entry." }
return argv
}
private fun parseOptionalNonNegativeLong(
raw: String,
label: String,
): Long? {
val value = raw.trim()
if (value.isEmpty()) return null
val parsed = value.toLongOrNull()
require(parsed != null && parsed >= 0L) { "$label must be a non-negative number of milliseconds." }
return parsed
}
private fun JsonObject.string(key: String): String? =
this[key]
.asStringOrNull()
?.trim()
?.takeIf { it.isNotEmpty() }
private fun JsonObject.long(key: String): Long? =
(this[key] as? JsonPrimitive)
?.content
?.trim()
?.toLongOrNull()
private fun JsonObject.optionalBoolean(key: String): Boolean? = (this[key] as? JsonPrimitive)?.booleanOrNull
@@ -51,6 +51,27 @@ internal fun shouldStartRuntimeOnForeground(
onboardingCompleted: Boolean,
): Boolean = foreground && onboardingCompleted
internal class CronEditorDraftMemory {
private var retained: Pair<String, CronEditorDraftState>? = null
fun get(jobId: String): CronEditorDraftState? = retained?.takeIf { it.first == jobId }?.second
fun set(
jobId: String,
state: CronEditorDraftState?,
) {
if (state == null) {
clear(jobId)
} else {
retained = jobId to state
}
}
fun clear(jobId: String) {
if (retained?.first == jobId) retained = null
}
}
/**
* UI-facing bridge that exposes NodeRuntime and preference state as Compose-friendly StateFlows.
*/
@@ -64,6 +85,10 @@ class MainViewModel(
private val gatewayConfigOperationSeq = AtomicLong()
private val gatewayConfigOperationMutex = Mutex()
// One bounded heap-only slot follows the ViewModel across Activity recreation.
// Detail disposal clears it; process death drops it with the ViewModel.
internal val cronEditorDraftMemory = CronEditorDraftMemory()
@Volatile private var permissionRequester: PermissionRequester? = null
@Volatile private var foreground = false
@@ -196,6 +221,9 @@ class MainViewModel(
val cronRefreshing: StateFlow<Boolean> = runtimeState(initial = false) { it.cronRefreshing }
val cronErrorText: StateFlow<String?> = runtimeState(initial = null) { it.cronErrorText }
val cronJobDetailState: StateFlow<GatewayCronJobDetailState> = runtimeState(initial = GatewayCronJobDetailState.Idle) { it.cronJobDetailState }
val cronRunHistoryState: StateFlow<GatewayCronRunHistoryState> = runtimeState(initial = GatewayCronRunHistoryState.Idle) { it.cronRunHistoryState }
val cronActionState: StateFlow<GatewayCronActionState> = runtimeState(initial = GatewayCronActionState.Idle) { it.cronActionState }
val pendingCronRunJobIds: StateFlow<Set<String>> = runtimeState(initial = emptySet()) { it.pendingCronRunJobIds }
val usageSummary: StateFlow<GatewayUsageSummary> = runtimeState(initial = GatewayUsageSummary(updatedAtMs = null, providers = emptyList())) { it.usageSummary }
val usageRefreshing: StateFlow<Boolean> = runtimeState(initial = false) { it.usageRefreshing }
val usageErrorText: StateFlow<String?> = runtimeState(initial = null) { it.usageErrorText }
@@ -736,10 +764,40 @@ class MainViewModel(
ensureRuntime().loadCronJobDetail(id)
}
fun refreshCronRunHistory(id: String) {
ensureRuntime().refreshCronRunHistory(id)
}
fun clearCronJobDetail() {
ensureRuntime().clearCronJobDetail()
}
fun dismissCronActionNotice(id: String) {
ensureRuntime().dismissCronActionNotice(id)
}
fun runCronJob(id: String) {
ensureRuntime().runCronJob(id)
}
fun setCronJobEnabled(
id: String,
enabled: Boolean,
) {
ensureRuntime().setCronJobEnabled(id = id, enabled = enabled)
}
fun updateCronJob(
original: GatewayCronJobDetail,
edit: GatewayCronJobEdit,
) {
ensureRuntime().updateCronJob(original = original, edit = edit)
}
fun deleteCronJob(id: String) {
ensureRuntime().deleteCronJob(id)
}
fun refreshUsage() {
ensureRuntime().refreshUsage()
}
@@ -24,6 +24,7 @@ import ai.openclaw.app.gateway.GatewayDiscovery
import ai.openclaw.app.gateway.GatewayEndpoint
import ai.openclaw.app.gateway.GatewayRegistryEntry
import ai.openclaw.app.gateway.GatewayRegistryEntryKind
import ai.openclaw.app.gateway.GatewayRequestRejected
import ai.openclaw.app.gateway.GatewaySession
import ai.openclaw.app.gateway.GatewayTlsProbeFailure
import ai.openclaw.app.gateway.GatewayTlsProbeResult
@@ -116,6 +117,7 @@ import java.util.concurrent.atomic.AtomicReference
private const val MAX_PENDING_NOTIFICATION_EVENTS = 128
private const val NODE_APPROVAL_COMMAND_FRESH_MS = 30_000L
private const val CRON_RUN_TRACKING_POLL_MS = 2_000L
private const val OperatorAdminScope = "operator.admin"
private enum class SkillWorkshopGatewayAction(
@@ -374,6 +376,13 @@ class NodeRuntime private constructor(
val generation: Long,
)
private data class CronActionResult(
val message: String,
val kind: GatewayCronNoticeKind,
val refresh: Boolean,
val deleted: Boolean = false,
)
constructor(
context: Context,
prefs: SecurePrefs = SecurePrefs(context.applicationContext),
@@ -715,7 +724,17 @@ class NodeRuntime private constructor(
val cronErrorText: StateFlow<String?> = _cronErrorText.asStateFlow()
private val _cronJobDetailState = MutableStateFlow<GatewayCronJobDetailState>(GatewayCronJobDetailState.Idle)
val cronJobDetailState: StateFlow<GatewayCronJobDetailState> = _cronJobDetailState.asStateFlow()
private val _cronRunHistoryState = MutableStateFlow<GatewayCronRunHistoryState>(GatewayCronRunHistoryState.Idle)
val cronRunHistoryState: StateFlow<GatewayCronRunHistoryState> = _cronRunHistoryState.asStateFlow()
private val _cronActionState = MutableStateFlow<GatewayCronActionState>(GatewayCronActionState.Idle)
val cronActionState: StateFlow<GatewayCronActionState> = _cronActionState.asStateFlow()
private val _pendingCronRunJobIds = MutableStateFlow<Set<String>>(emptySet())
val pendingCronRunJobIds: StateFlow<Set<String>> = _pendingCronRunJobIds.asStateFlow()
private val cronJobDetailRequestGuard = CronJobDetailRequestGuard()
private val cronRunHistoryRequestGuard = CronJobDetailRequestGuard()
private val cronRefreshGuard = LatestGatewayRefreshGuard()
private val cronActionMutex = Mutex()
private val pendingCronRunRegistry = PendingCronRunRegistry()
private val _usageSummary = MutableStateFlow(GatewayUsageSummary(updatedAtMs = null, providers = emptyList()))
val usageSummary: StateFlow<GatewayUsageSummary> = _usageSummary.asStateFlow()
private val _usageRefreshing = MutableStateFlow(false)
@@ -756,7 +775,7 @@ class NodeRuntime private constructor(
val nodesDevicesRefreshing: StateFlow<Boolean> = _nodesDevicesRefreshing.asStateFlow()
private val _nodesDevicesErrorText = MutableStateFlow<String?>(null)
val nodesDevicesErrorText: StateFlow<String?> = _nodesDevicesErrorText.asStateFlow()
private val nodeApprovalRefreshGuard = GatewayNodeApprovalRefreshGuard()
private val nodeApprovalRefreshGuard = LatestGatewayRefreshGuard()
private val _execApprovals = MutableStateFlow<List<GatewayExecApprovalSummary>>(emptyList())
val execApprovals: StateFlow<List<GatewayExecApprovalSummary>> = _execApprovals.asStateFlow()
private val _execApprovalsRefreshing = MutableStateFlow(false)
@@ -848,7 +867,7 @@ class NodeRuntime private constructor(
}
},
onDisconnected = { message ->
clearOperatorGatewayState()
clearOperatorGatewayState(retirePendingCronRuns = false)
chat.applyMainSessionKey(resolveMainSessionKey())
chat.onDisconnected(message)
updateStatus {
@@ -869,7 +888,7 @@ class NodeRuntime private constructor(
customHeadersProvider = prefs::loadGatewayCustomHeaders,
)
private fun clearOperatorGatewayState() {
private fun clearOperatorGatewayState(retirePendingCronRuns: Boolean) {
invalidateNodeCapabilityApprovalState()
_serverName.value = null
_remoteAddress.value = null
@@ -885,11 +904,17 @@ class NodeRuntime private constructor(
_modelCatalogRefreshing.value = false
_modelCatalogErrorText.value = null
_talkSetupReadiness.value = GatewayTalkSetupReadiness.unverified()
cronRefreshGuard.invalidate()
_cronStatus.value = GatewayCronStatus(enabled = false, jobs = 0, nextWakeAtMs = null)
_cronJobs.value = emptyList()
_cronRefreshing.value = false
_cronErrorText.value = null
cronJobDetailRequestGuard.cancel { _cronJobDetailState.value = GatewayCronJobDetailState.Idle }
cronRunHistoryRequestGuard.cancel { _cronRunHistoryState.value = GatewayCronRunHistoryState.Idle }
_cronActionState.value = GatewayCronActionState.Idle
if (retirePendingCronRuns) {
pendingCronRunRegistry.clear { _pendingCronRunJobIds.value = it }
}
_usageSummary.value = GatewayUsageSummary(updatedAtMs = null, providers = emptyList())
_usageRefreshing.value = false
_usageErrorText.value = null
@@ -1408,17 +1433,174 @@ class NodeRuntime private constructor(
}
fun loadCronJobDetail(id: String) {
val request = cronJobDetailRequestGuard.begin(id) ?: return
_cronJobDetailState.value = GatewayCronJobDetailState.Loading(request.id)
scope.launch {
loadCronJobDetailFromGateway(request)
val detailRequest = cronJobDetailRequestGuard.begin(id) ?: return
val historyRequest = cronRunHistoryRequestGuard.begin(detailRequest.id) ?: return
_cronJobDetailState.value = GatewayCronJobDetailState.Loading(detailRequest.id)
_cronRunHistoryState.value = GatewayCronRunHistoryState.Loading(historyRequest.id)
if (mode == NodeRuntimeMode.ScreenshotFixture) {
applyScreenshotCronDetail(detailRequest = detailRequest, historyRequest = historyRequest)
return
}
scope.launch { loadCronJobDetailFromGateway(detailRequest) }
scope.launch { loadCronRunHistoryFromGateway(historyRequest) }
}
fun refreshCronRunHistory(id: String) {
val request = cronRunHistoryRequestGuard.begin(id) ?: return
_cronRunHistoryState.value = GatewayCronRunHistoryState.Loading(request.id)
if (mode == NodeRuntimeMode.ScreenshotFixture) {
publishScreenshotCronHistory(request)
return
}
scope.launch { loadCronRunHistoryFromGateway(request) }
}
fun clearCronJobDetail() {
cronJobDetailRequestGuard.cancel {
_cronJobDetailState.value = GatewayCronJobDetailState.Idle
}
cronRunHistoryRequestGuard.cancel {
_cronRunHistoryState.value = GatewayCronRunHistoryState.Idle
}
}
fun dismissCronActionNotice(id: String) {
val jobId = id.trim().takeIf { it.isNotEmpty() } ?: return
val notice = _cronActionState.value as? GatewayCronActionState.Notice
if (notice?.id == jobId) {
_cronActionState.value = GatewayCronActionState.Idle
}
}
fun runCronJob(id: String) {
val jobId = id.trim().takeIf { it.isNotEmpty() } ?: return
if (pendingCronRunRegistry.contains(jobId)) {
_cronActionState.value =
GatewayCronActionState.Notice(
id = jobId,
message = "This cron job already has a queued run.",
kind = GatewayCronNoticeKind.Warning,
)
return
}
launchCronAction(id = jobId, action = GatewayCronAction.Run) { gatewayScope, actionJobId ->
val response =
requestGatewayData(
gatewayScope,
"cron.run",
buildJsonObject {
put("id", JsonPrimitive(actionJobId))
put("mode", JsonPrimitive("force"))
}.toString(),
)
when (val outcome = parseGatewayCronRunOutcome(json.parseToJsonElement(response).asObjectOrNull())) {
is GatewayCronRunOutcome.Started -> {
outcome.runId?.let { runId ->
var trackingStarted = false
publishGatewayData(gatewayScope) {
trackingStarted =
pendingCronRunRegistry.begin(actionJobId, runId) {
_pendingCronRunJobIds.value = it
}
}
if (trackingStarted) {
trackQueuedCronRun(gatewayScope = gatewayScope, jobId = actionJobId, runId = runId)
}
}
CronActionResult(
message = if (outcome.runId == null) "Cron job started." else "Cron run queued.",
kind = GatewayCronNoticeKind.Success,
refresh = cronRunShouldRefresh(outcome),
)
}
is GatewayCronRunOutcome.Skipped ->
CronActionResult(
message = outcome.reason.message,
kind = GatewayCronNoticeKind.Warning,
refresh = cronRunShouldRefresh(outcome),
)
GatewayCronRunOutcome.Rejected ->
CronActionResult(
message = "Gateway rejected the cron run.",
kind = GatewayCronNoticeKind.Error,
refresh = false,
)
null -> error("Gateway returned an invalid cron run result.")
}
}
}
fun setCronJobEnabled(
id: String,
enabled: Boolean,
) {
launchCronAction(
id = id,
action = if (enabled) GatewayCronAction.Enable else GatewayCronAction.Disable,
) { gatewayScope, jobId ->
requestGatewayData(
gatewayScope,
"cron.update",
buildJsonObject {
put("id", JsonPrimitive(jobId))
put(
"patch",
buildJsonObject {
put("enabled", JsonPrimitive(enabled))
},
)
}.toString(),
)
CronActionResult(
message = if (enabled) "Cron job enabled." else "Cron job disabled.",
kind = GatewayCronNoticeKind.Success,
refresh = true,
)
}
}
fun updateCronJob(
original: GatewayCronJobDetail,
edit: GatewayCronJobEdit,
) {
launchCronAction(id = original.id, action = GatewayCronAction.Save) { gatewayScope, _ ->
try {
requestGatewayData(
gatewayScope,
"cron.update",
buildCronUpdateParams(original = original, edit = edit),
)
} catch (err: GatewayRequestRejected) {
if (!isCronJobRevisionConflict(err.gatewayError)) throw err
reloadCronJobIfSelected(original.id)
return@launchCronAction CronActionResult(
message = "This cron job changed on the gateway. Review the latest version before saving again.",
kind = GatewayCronNoticeKind.Warning,
refresh = false,
)
}
CronActionResult(
message = "Cron job updated.",
kind = GatewayCronNoticeKind.Success,
refresh = true,
)
}
}
fun deleteCronJob(id: String) {
launchCronAction(id = id, action = GatewayCronAction.Delete) { gatewayScope, jobId ->
requestGatewayData(
gatewayScope,
"cron.remove",
buildJsonObject { put("id", JsonPrimitive(jobId)) }.toString(),
)
CronActionResult(
message = "Cron job deleted.",
kind = GatewayCronNoticeKind.Success,
refresh = true,
deleted = true,
)
}
}
fun refreshUsage() {
@@ -1729,9 +1911,11 @@ class NodeRuntime private constructor(
_cronStatus.value =
GatewayCronStatus(
enabled = true,
jobs = 2,
jobs = 1,
nextWakeAtMs = 1_783_641_600_000,
)
_cronJobs.value = parseScreenshotCronJobs()
_operatorScopes.value = listOf(OperatorAdminScope)
_nodesDevicesSummary.value = AndroidScreenshotFixture.nodes
_channelsSummary.value = AndroidScreenshotFixture.channels
_nodeCapabilityApproval.value = GatewayNodeCapabilityApproval.Approved
@@ -1748,6 +1932,44 @@ class NodeRuntime private constructor(
chat.refreshSessions(limit = 20)
}
private fun parseScreenshotCronJobs(): List<GatewayCronJobSummary> {
// Screenshot mode parses gateway-shaped fixtures so UI navigation covers the live data contract.
val list =
json
.parseToJsonElement(AndroidScreenshotFixture.request("cron.list", null))
.asObjectOrNull()
return parseCronJobs(list?.get("jobs") as? JsonArray)
}
private fun applyScreenshotCronDetail(
detailRequest: CronJobDetailRequest,
historyRequest: CronJobDetailRequest,
) {
val detail =
json
.parseToJsonElement(AndroidScreenshotFixture.request("cron.get", cronJobGetParams(detailRequest.id)))
.asObjectOrNull()
?.let(::parseGatewayCronJobDetail)
?.takeIf { it.id == detailRequest.id }
cronJobDetailRequestGuard.publishIfCurrent(detailRequest) {
_cronJobDetailState.value =
detail?.let(GatewayCronJobDetailState::Loaded)
?: GatewayCronJobDetailState.Error(detailRequest.id, "Gateway returned an invalid cron job.")
}
publishScreenshotCronHistory(historyRequest)
}
private fun publishScreenshotCronHistory(request: CronJobDetailRequest) {
val history =
json
.parseToJsonElement(AndroidScreenshotFixture.request("cron.runs", cronJobGetParams(request.id)))
.asObjectOrNull()
val runs = parseGatewayCronRunHistory(history?.get("entries") as? JsonArray)
cronRunHistoryRequestGuard.publishIfCurrent(request) {
_cronRunHistoryState.value = GatewayCronRunHistoryState.Loaded(id = request.id, runs = runs)
}
}
init {
if (mode == NodeRuntimeMode.Live) {
if (prefs.voiceWakeMode.value != VoiceWakeMode.Off) {
@@ -3126,7 +3348,7 @@ class NodeRuntime private constructor(
connectAttemptSeq.incrementAndGet()
synchronized(gatewayDataScopeLock) {
gatewayDataGeneration += 1
clearOperatorGatewayState()
clearOperatorGatewayState(retirePendingCronRuns = true)
}
chat.onGatewayScopeChanging(retireRunState)
stopMessageSpeech()
@@ -3474,6 +3696,15 @@ class NodeRuntime private constructor(
}
}
private inline fun publishCronRefresh(
gatewayScope: GatewayDataScope,
refreshGeneration: Long,
crossinline publish: () -> Unit,
): Boolean =
publishGatewayData(gatewayScope) {
cronRefreshGuard.publishIfCurrent(refreshGeneration) { publish() }
}
private suspend fun refreshBrandingFromGateway() {
val gatewayScope = captureGatewayDataScope() ?: return
if (!gatewayConnectionDisplay.value.isConnected) return
@@ -3611,15 +3842,18 @@ class NodeRuntime private constructor(
}
private suspend fun refreshCronFromGateway() {
val refreshGeneration = cronRefreshGuard.begin()
val gatewayScope = captureGatewayDataScope() ?: return
publishGatewayData(gatewayScope) {
publishCronRefresh(gatewayScope, refreshGeneration) {
_cronRefreshing.value = true
_cronErrorText.value = null
}
if (!operatorConnected) {
_cronStatus.value = GatewayCronStatus(enabled = false, jobs = 0, nextWakeAtMs = null)
_cronJobs.value = emptyList()
_cronRefreshing.value = false
publishCronRefresh(gatewayScope, refreshGeneration) {
_cronStatus.value = GatewayCronStatus(enabled = false, jobs = 0, nextWakeAtMs = null)
_cronJobs.value = emptyList()
_cronRefreshing.value = false
}
return
}
try {
@@ -3635,14 +3869,18 @@ class NodeRuntime private constructor(
val listRes = requestGatewayData(gatewayScope, "cron.list", """{"includeDisabled":true,"limit":20,"sortBy":"nextRunAtMs","sortDir":"asc"}""")
val listRoot = json.parseToJsonElement(listRes).asObjectOrNull()
val jobs = parseCronJobs(listRoot?.get("jobs") as? JsonArray)
publishGatewayData(gatewayScope) {
publishCronRefresh(gatewayScope, refreshGeneration) {
_cronStatus.value = status
_cronJobs.value = jobs
}
} catch (_: Throwable) {
publishGatewayData(gatewayScope) { _cronErrorText.value = "Could not load cron jobs." }
publishCronRefresh(gatewayScope, refreshGeneration) {
_cronErrorText.value = "Could not load cron jobs."
}
} finally {
publishGatewayData(gatewayScope) { _cronRefreshing.value = false }
publishCronRefresh(gatewayScope, refreshGeneration) {
_cronRefreshing.value = false
}
}
}
@@ -3669,6 +3907,230 @@ class NodeRuntime private constructor(
}
}
private suspend fun loadCronRunHistoryFromGateway(request: CronJobDetailRequest) {
val gatewayScope = captureGatewayDataScope() ?: return
if (!operatorConnected) {
cronRunHistoryRequestGuard.publishIfCurrent(request) {
_cronRunHistoryState.value =
GatewayCronRunHistoryState.Error(
id = request.id,
message = "Connect the gateway to inspect cron run history.",
)
}
return
}
try {
val response =
requestGatewayData(
gatewayScope,
"cron.runs",
buildJsonObject {
put("id", JsonPrimitive(request.id))
put("limit", JsonPrimitive(20))
put("sortDir", JsonPrimitive("desc"))
}.toString(),
)
val root = json.parseToJsonElement(response).asObjectOrNull()
val runs = parseGatewayCronRunHistory(root?.get("entries") as? JsonArray)
publishGatewayData(gatewayScope) {
cronRunHistoryRequestGuard.publishIfCurrent(request) {
_cronRunHistoryState.value = GatewayCronRunHistoryState.Loaded(id = request.id, runs = runs)
}
}
} catch (err: CancellationException) {
throw err
} catch (_: Throwable) {
publishGatewayData(gatewayScope) {
cronRunHistoryRequestGuard.publishIfCurrent(request) {
_cronRunHistoryState.value =
GatewayCronRunHistoryState.Error(
id = request.id,
message = "Could not load cron run history.",
)
}
}
}
}
private fun launchCronAction(
id: String,
action: GatewayCronAction,
perform: suspend (GatewayDataScope, String) -> CronActionResult,
) {
val jobId = id.trim().takeIf { it.isNotEmpty() } ?: return
if (!operatorAdminScopeAvailable.value) {
_cronActionState.value =
GatewayCronActionState.Notice(
id = jobId,
message = "Cron changes require operator.admin access.",
kind = GatewayCronNoticeKind.Error,
)
return
}
if (!operatorConnected) {
_cronActionState.value =
GatewayCronActionState.Notice(
id = jobId,
message = "Connect the gateway to manage cron jobs.",
kind = GatewayCronNoticeKind.Error,
)
return
}
if (_cronActionState.value is GatewayCronActionState.Running) return
// One mutating RPC at a time keeps button taps and programmatic calls from racing.
if (!cronActionMutex.tryLock()) {
if (_cronActionState.value !is GatewayCronActionState.Running) {
_cronActionState.value =
GatewayCronActionState.Notice(
id = jobId,
message = "Another cron action is still finishing.",
kind = GatewayCronNoticeKind.Warning,
)
}
return
}
// Publish ownership before returning to Compose so Activity recreation can
// distinguish a retained Save from dead pending state after process death.
val actionScope = captureGatewayDataScope()
if (actionScope == null) {
cronActionMutex.unlock()
return
}
val started =
publishGatewayData(actionScope) {
_cronActionState.value = GatewayCronActionState.Running(id = jobId, action = action)
}
if (!started) {
cronActionMutex.unlock()
return
}
scope.launch {
var completionState: GatewayCronActionState.Notice? = null
try {
val result = perform(actionScope, jobId)
if (result.deleted) {
clearDeletedCronSelection(jobId)
}
if (result.refresh) {
refreshCronFromGateway()
if (!result.deleted) reloadCronJobIfSelected(jobId)
}
completionState =
GatewayCronActionState.Notice(
id = jobId,
message = result.message,
kind = result.kind,
deleted = result.deleted,
)
} catch (err: CancellationException) {
throw err
} catch (err: Throwable) {
val message = err.message?.trim()?.takeIf { it.isNotEmpty() } ?: "Cron action failed."
completionState =
GatewayCronActionState.Notice(
id = jobId,
message = message,
kind = GatewayCronNoticeKind.Error,
)
} finally {
cronActionMutex.unlock()
val notice = completionState
if (notice != null) {
publishGatewayData(actionScope) {
_cronActionState.value = notice
}
}
}
}
}
private fun reloadCronJobIfSelected(jobId: String) {
// Ownership checks and loading publication stay under each guard's lock;
// navigation that wins afterward invalidates these requests before publish.
val detailRequest =
cronJobDetailRequestGuard.beginIfCurrent(jobId) { request ->
_cronJobDetailState.value = GatewayCronJobDetailState.Loading(request.id)
}
val historyRequest =
cronRunHistoryRequestGuard.beginIfCurrent(jobId) { request ->
_cronRunHistoryState.value = GatewayCronRunHistoryState.Loading(request.id)
}
detailRequest?.let { scope.launch { loadCronJobDetailFromGateway(it) } }
historyRequest?.let { scope.launch { loadCronRunHistoryFromGateway(it) } }
}
private fun clearDeletedCronSelection(jobId: String) {
// A completed delete can race navigation to another job. Clear only state
// still owned by the deleted id so the newer detail/history survives.
cronJobDetailRequestGuard.cancelIfCurrent(jobId) {
_cronJobDetailState.value = GatewayCronJobDetailState.Idle
}
cronRunHistoryRequestGuard.cancelIfCurrent(jobId) {
_cronRunHistoryState.value = GatewayCronRunHistoryState.Idle
}
}
private fun trackQueuedCronRun(
gatewayScope: GatewayDataScope,
jobId: String,
runId: String,
) {
// cron.run acknowledges before lane admission. Track its exact run-log id
// so only this job stays deduped until terminal evidence or scope retirement.
scope.launch {
var completedRun: GatewayCronRunSummary? = null
while (isGatewayDataScopeCurrent(gatewayScope) && completedRun == null) {
completedRun =
try {
val response =
requestGatewayData(
gatewayScope,
"cron.runs",
buildJsonObject {
put("id", JsonPrimitive(jobId))
put("runId", JsonPrimitive(runId))
put("limit", JsonPrimitive(1))
put("sortDir", JsonPrimitive("desc"))
}.toString(),
)
val root = json.parseToJsonElement(response).asObjectOrNull()
parseGatewayCronRunHistory(root?.get("entries") as? JsonArray)
.firstOrNull { it.runId == runId }
} catch (err: CancellationException) {
throw err
} catch (_: Throwable) {
if (!isGatewayDataScopeCurrent(gatewayScope)) return@launch
null
}
if (completedRun == null) delay(CRON_RUN_TRACKING_POLL_MS)
}
if (!isGatewayDataScopeCurrent(gatewayScope)) return@launch
val terminalRun = completedRun ?: return@launch
var pendingCleared = false
val scopeCurrent =
publishGatewayData(gatewayScope) {
pendingCleared =
pendingCronRunRegistry.finish(jobId, runId) {
_pendingCronRunJobIds.value = it
}
}
if (!scopeCurrent || !pendingCleared) return@launch
refreshCronFromGateway()
reloadCronJobIfSelected(jobId)
publishGatewayData(gatewayScope) {
val currentAction = _cronActionState.value
val canPublish =
currentAction == GatewayCronActionState.Idle ||
(currentAction is GatewayCronActionState.Notice && currentAction.id == jobId)
if (canPublish) {
_cronActionState.value = cronRunCompletionNotice(jobId, terminalRun.status)
}
}
}
}
private suspend fun refreshUsageFromGateway() {
val gatewayScope = captureGatewayDataScope() ?: return
publishGatewayData(gatewayScope) {
@@ -5339,8 +5801,8 @@ internal fun GatewayNodeCapabilityApproval.withoutExactRequestId(): GatewayNodeC
internal fun GatewayNodesDevicesSummary.withoutExactApprovalRequestIds(): GatewayNodesDevicesSummary = copy(nodes = nodes.map { node -> node.copy(pendingRequestId = null) })
/** Prevents older node.list responses from overwriting newer approval state. */
internal class GatewayNodeApprovalRefreshGuard {
/** Prevents an older gateway response from publishing after a newer refresh begins. */
internal class LatestGatewayRefreshGuard {
private val lock = Any()
private var generation = 0L
@@ -5350,6 +5812,10 @@ internal class GatewayNodeApprovalRefreshGuard {
generation
}
fun invalidate() {
begin()
}
fun publishIfCurrent(
refreshGeneration: Long,
publish: () -> Unit,
@@ -0,0 +1,642 @@
package ai.openclaw.app.ui
import ai.openclaw.app.CronEditorDraftState
import ai.openclaw.app.GatewayCronActionState
import ai.openclaw.app.GatewayCronJobDetail
import ai.openclaw.app.GatewayCronJobEdit
import ai.openclaw.app.GatewayCronNoticeKind
import ai.openclaw.app.GatewayCronPayloadEdit
import ai.openclaw.app.GatewayCronRunHistoryState
import ai.openclaw.app.GatewayCronRunSummary
import ai.openclaw.app.GatewayCronScheduleEdit
import ai.openclaw.app.ui.design.ClawDetailRow
import ai.openclaw.app.ui.design.ClawIconBadge
import ai.openclaw.app.ui.design.ClawListPanel
import ai.openclaw.app.ui.design.ClawPanel
import ai.openclaw.app.ui.design.ClawPrimaryButton
import ai.openclaw.app.ui.design.ClawSecondaryButton
import ai.openclaw.app.ui.design.ClawSegmentedControl
import ai.openclaw.app.ui.design.ClawStatus
import ai.openclaw.app.ui.design.ClawStatusPill
import ai.openclaw.app.ui.design.ClawTextField
import ai.openclaw.app.ui.design.ClawTheme
import androidx.compose.foundation.layout.Arrangement
import androidx.compose.foundation.layout.Column
import androidx.compose.foundation.layout.Row
import androidx.compose.foundation.layout.fillMaxWidth
import androidx.compose.foundation.layout.heightIn
import androidx.compose.foundation.layout.size
import androidx.compose.material.icons.Icons
import androidx.compose.material.icons.filled.Delete
import androidx.compose.material.icons.filled.Edit
import androidx.compose.material.icons.filled.History
import androidx.compose.material.icons.filled.Lock
import androidx.compose.material.icons.filled.Pause
import androidx.compose.material.icons.filled.PlayArrow
import androidx.compose.material.icons.filled.Refresh
import androidx.compose.material.icons.filled.Save
import androidx.compose.material.icons.filled.Schedule
import androidx.compose.material3.AlertDialog
import androidx.compose.material3.Icon
import androidx.compose.material3.Switch
import androidx.compose.material3.Text
import androidx.compose.material3.TextButton
import androidx.compose.runtime.Composable
import androidx.compose.runtime.getValue
import androidx.compose.runtime.mutableStateOf
import androidx.compose.runtime.remember
import androidx.compose.runtime.setValue
import androidx.compose.ui.Alignment
import androidx.compose.ui.Modifier
import androidx.compose.ui.semantics.contentDescription
import androidx.compose.ui.semantics.semantics
import androidx.compose.ui.text.style.TextOverflow
import androidx.compose.ui.unit.dp
import java.text.DateFormat
import java.util.Date
@Composable
internal fun CronJobManagementPanel(
job: GatewayCronJobDetail,
editorDraft: CronEditorDraftState,
onEditorDraftChange: (CronEditorDraftState) -> Unit,
historyState: GatewayCronRunHistoryState,
actionState: GatewayCronActionState,
runPending: Boolean,
operatorAdminScopeAvailable: Boolean,
onRun: () -> Unit,
onToggleEnabled: () -> Unit,
onSave: (GatewayCronJobEdit) -> Unit,
onRefreshHistory: () -> Unit,
onDelete: () -> Unit,
) {
val busy = actionState is GatewayCronActionState.Running
val notice = (actionState as? GatewayCronActionState.Notice)?.takeIf { it.id == job.id }
var showDeleteConfirmation by remember(job.id) { mutableStateOf(false) }
if (showDeleteConfirmation) {
AlertDialog(
onDismissRequest = { showDeleteConfirmation = false },
confirmButton = {
TextButton(
onClick = {
showDeleteConfirmation = false
onDelete()
},
) {
Text("Delete")
}
},
dismissButton = {
TextButton(onClick = { showDeleteConfirmation = false }) {
Text("Cancel")
}
},
title = { Text("Delete cron job?") },
text = { Text("This permanently removes the scheduled job from the gateway.") },
)
}
notice?.let { value ->
ClawPanel {
Text(
text = value.message,
style = ClawTheme.type.body,
color =
when (value.kind) {
GatewayCronNoticeKind.Success -> ClawTheme.colors.success
GatewayCronNoticeKind.Warning -> ClawTheme.colors.warning
GatewayCronNoticeKind.Error -> ClawTheme.colors.danger
},
)
}
}
if (!operatorAdminScopeAvailable) CronAdminAccessPanel()
if (editorDraft.requiresResolution) {
ClawPanel {
Text(
text =
if (editorDraft.hasIncomingConflict) {
"This job changed while you were editing. Revert to the latest gateway version before saving."
} else {
"Save or revert your edits before running, enabling, disabling, deleting, or refreshing this job."
},
style = ClawTheme.type.body,
color = ClawTheme.colors.warning,
)
}
}
CronActionPanel(
job = job,
enabled = operatorAdminScopeAvailable && !busy && !editorDraft.requiresResolution,
busy = busy,
runPending = runPending,
onRun = onRun,
onToggleEnabled = onToggleEnabled,
onDelete = { showDeleteConfirmation = true },
)
CronEditorPanel(
job = job,
draft = editorDraft,
onDraftChange = onEditorDraftChange,
enabled =
operatorAdminScopeAvailable &&
!busy &&
!editorDraft.savePending &&
!editorDraft.saveSucceeded,
canRevert = !busy && !editorDraft.savePending && !editorDraft.saveSucceeded,
busy = busy,
onSave = onSave,
)
CronRunHistoryPanel(
jobId = job.id,
state = historyState,
onRefresh = onRefreshHistory,
)
}
@Composable
private fun CronAdminAccessPanel() {
ClawPanel {
Column(verticalArrangement = Arrangement.spacedBy(7.dp)) {
Row(
verticalAlignment = Alignment.CenterVertically,
horizontalArrangement = Arrangement.spacedBy(8.dp),
) {
Icon(
imageVector = Icons.Default.Lock,
contentDescription = null,
modifier = Modifier.size(17.dp),
tint = ClawTheme.colors.text,
)
Text(text = "Admin access required", style = ClawTheme.type.section, color = ClawTheme.colors.text)
}
Text(
text =
"Cron changes require operator.admin. Setup codes intentionally do not grant it. " +
"Reconnect with the gateway's shared token or password to request admin access. " +
"If this device still lacks it, approve the pending scope upgrade from an existing admin client.",
style = ClawTheme.type.body,
color = ClawTheme.colors.textMuted,
)
}
}
}
@Composable
private fun CronActionPanel(
job: GatewayCronJobDetail,
enabled: Boolean,
busy: Boolean,
runPending: Boolean,
onRun: () -> Unit,
onToggleEnabled: () -> Unit,
onDelete: () -> Unit,
) {
ClawPanel {
Column(verticalArrangement = Arrangement.spacedBy(8.dp)) {
Row(
modifier = Modifier.fillMaxWidth(),
horizontalArrangement = Arrangement.spacedBy(8.dp),
) {
ClawPrimaryButton(
text =
when {
busy -> "Working"
runPending -> "Run Pending"
else -> "Run Now"
},
onClick = onRun,
modifier = Modifier.weight(1f),
enabled = enabled && !runPending,
icon = Icons.Default.PlayArrow,
)
ClawSecondaryButton(
text = if (job.enabled) "Disable" else "Enable",
onClick = onToggleEnabled,
modifier = Modifier.weight(1f),
enabled = enabled,
icon = if (job.enabled) Icons.Default.Pause else Icons.Default.PlayArrow,
)
}
ClawSecondaryButton(
text = "Delete Job",
onClick = onDelete,
modifier = Modifier.fillMaxWidth(),
enabled = enabled,
icon = Icons.Default.Delete,
)
}
}
}
@Composable
private fun CronEditorPanel(
job: GatewayCronJobDetail,
draft: CronEditorDraftState,
onDraftChange: (CronEditorDraftState) -> Unit,
enabled: Boolean,
canRevert: Boolean,
busy: Boolean,
onSave: (GatewayCronJobEdit) -> Unit,
) {
val edit = draft.edit
ClawPanel {
Column(verticalArrangement = Arrangement.spacedBy(9.dp)) {
Row(
verticalAlignment = Alignment.CenterVertically,
horizontalArrangement = Arrangement.spacedBy(8.dp),
) {
Icon(
imageVector = Icons.Default.Edit,
contentDescription = null,
modifier = Modifier.size(17.dp),
tint = ClawTheme.colors.text,
)
Text(text = "Edit Job", style = ClawTheme.type.section, color = ClawTheme.colors.text)
}
CronSwitchRow(
title = "Enabled",
subtitle = "Allow the scheduler to run this job.",
checked = edit.enabled,
onCheckedChange = { onDraftChange(draft.withEdit(edit.copy(enabled = it))) },
enabled = enabled,
)
if (edit.schedule is GatewayCronScheduleEdit.At) {
CronSwitchRow(
title = "Delete after run",
subtitle = "Remove this job after a successful one-shot run.",
checked = edit.deleteAfterRun,
onCheckedChange = { onDraftChange(draft.withEdit(edit.copy(deleteAfterRun = it))) },
enabled = enabled,
)
}
ClawTextField(
value = edit.name,
onValueChange = { onDraftChange(draft.withEdit(edit.copy(name = it))) },
placeholder = "Job name",
label = "Name",
enabled = enabled,
)
ClawTextField(
value = edit.description,
onValueChange = { onDraftChange(draft.withEdit(edit.copy(description = it))) },
placeholder = "Optional description",
label = "Description",
enabled = enabled,
minLines = 2,
)
CronScheduleEditor(
schedule = edit.schedule,
enabled = enabled,
onChange = { onDraftChange(draft.withEdit(edit.withSchedule(it))) },
)
ClawTextField(
value = edit.sessionTarget,
onValueChange = { onDraftChange(draft.withEdit(edit.copy(sessionTarget = it))) },
placeholder = "main, isolated, current, or session:<id>",
label = "Session target",
enabled = enabled,
)
ClawSegmentedControl(
options = listOf("next-heartbeat", "now"),
selected = edit.wakeMode,
onSelect = { onDraftChange(draft.withEdit(edit.copy(wakeMode = it))) },
modifier = Modifier.fillMaxWidth(),
enabledOptions = if (enabled) setOf("next-heartbeat", "now") else emptySet(),
)
CronPayloadEditor(
payload = edit.payload,
originalCommandCwd = job.payloadCommandCwd,
enabled = enabled,
onChange = { onDraftChange(draft.withEdit(edit.copy(payload = it))) },
)
ClawPrimaryButton(
text = if (busy) "Working" else "Save Changes",
onClick = {
onDraftChange(draft.saveStarted())
onSave(edit)
},
modifier = Modifier.fillMaxWidth(),
enabled =
enabled &&
draft.isDirty &&
!draft.hasIncomingConflict &&
!draft.savePending &&
!draft.saveSucceeded,
icon = Icons.Default.Save,
)
if (draft.requiresResolution) {
ClawSecondaryButton(
text = "Revert Changes",
onClick = { onDraftChange(CronEditorDraftState.from(job)) },
modifier = Modifier.fillMaxWidth(),
enabled = canRevert,
)
}
}
}
}
@Composable
private fun CronSwitchRow(
title: String,
subtitle: String,
checked: Boolean,
onCheckedChange: (Boolean) -> Unit,
enabled: Boolean,
) {
Row(
modifier = Modifier.fillMaxWidth().heightIn(min = 50.dp),
verticalAlignment = Alignment.CenterVertically,
horizontalArrangement = Arrangement.spacedBy(9.dp),
) {
Column(modifier = Modifier.weight(1f), verticalArrangement = Arrangement.spacedBy(1.dp)) {
Text(text = title, style = ClawTheme.type.body, color = ClawTheme.colors.text)
Text(
text = subtitle,
style = ClawTheme.type.caption,
color = ClawTheme.colors.textMuted,
maxLines = 2,
overflow = TextOverflow.Ellipsis,
)
}
Switch(
checked = checked,
onCheckedChange = onCheckedChange,
enabled = enabled,
modifier = Modifier.semantics { contentDescription = title },
)
}
}
@Composable
private fun CronScheduleEditor(
schedule: GatewayCronScheduleEdit,
enabled: Boolean,
onChange: (GatewayCronScheduleEdit) -> Unit,
) {
Text(
text = "Schedule · ${cronScheduleKindLabel(schedule)}",
style = ClawTheme.type.caption,
color = ClawTheme.colors.textMuted,
)
when (schedule) {
is GatewayCronScheduleEdit.At ->
ClawTextField(
value = schedule.at,
onValueChange = { onChange(schedule.copy(at = it)) },
placeholder = "ISO time, e.g. 2026-07-09T09:30:00Z",
label = "Run at",
enabled = enabled,
)
is GatewayCronScheduleEdit.Every -> {
ClawTextField(
value = schedule.everyMs,
onValueChange = { onChange(schedule.copy(everyMs = it.filter(Char::isDigit))) },
placeholder = "Milliseconds",
label = "Interval",
enabled = enabled,
)
ClawTextField(
value = schedule.anchorMs,
onValueChange = { onChange(schedule.copy(anchorMs = it.filter(Char::isDigit))) },
placeholder = "Epoch milliseconds (optional)",
label = "Anchor",
enabled = enabled,
)
}
is GatewayCronScheduleEdit.Cron -> {
ClawTextField(
value = schedule.expression,
onValueChange = { onChange(schedule.copy(expression = it)) },
placeholder = "Cron expression, e.g. 0 9 * * *",
label = "Expression",
enabled = enabled,
)
Row(modifier = Modifier.fillMaxWidth(), horizontalArrangement = Arrangement.spacedBy(8.dp)) {
ClawTextField(
value = schedule.timezone,
onValueChange = { onChange(schedule.copy(timezone = it)) },
placeholder = "e.g. America/New_York",
label = "Timezone",
enabled = enabled,
modifier = Modifier.weight(1f),
)
ClawTextField(
value = schedule.staggerMs,
onValueChange = { onChange(schedule.copy(staggerMs = it.filter(Char::isDigit))) },
placeholder = "0 = exact",
label = "Stagger ms",
enabled = enabled,
modifier = Modifier.weight(1f),
)
}
}
is GatewayCronScheduleEdit.OnExit -> {
ClawTextField(
value = schedule.command,
onValueChange = { onChange(schedule.copy(command = it)) },
placeholder = "Command to watch",
label = "Command",
enabled = enabled,
)
ClawTextField(
value = schedule.cwd,
onValueChange = { onChange(schedule.copy(cwd = it)) },
placeholder = "Optional path",
label = "Working directory",
enabled = enabled,
)
}
}
}
@Composable
private fun CronPayloadEditor(
payload: GatewayCronPayloadEdit,
originalCommandCwd: String?,
enabled: Boolean,
onChange: (GatewayCronPayloadEdit) -> Unit,
) {
Text(
text = "Payload · ${cronPayloadKindLabel(payload)}",
style = ClawTheme.type.caption,
color = ClawTheme.colors.textMuted,
)
when (payload) {
is GatewayCronPayloadEdit.SystemEvent ->
ClawTextField(
value = payload.text,
onValueChange = { onChange(payload.copy(text = it)) },
placeholder = "System event text",
label = "Event text",
enabled = enabled,
minLines = 3,
)
is GatewayCronPayloadEdit.AgentTurn -> {
ClawTextField(
value = payload.message,
onValueChange = { onChange(payload.copy(message = it)) },
placeholder = "Agent message",
label = "Message",
enabled = enabled,
minLines = 3,
)
Row(modifier = Modifier.fillMaxWidth(), horizontalArrangement = Arrangement.spacedBy(8.dp)) {
ClawTextField(
value = payload.model,
onValueChange = { onChange(payload.copy(model = it)) },
placeholder = "Optional override",
label = "Model",
enabled = enabled,
modifier = Modifier.weight(1f),
)
ClawTextField(
value = payload.thinking,
onValueChange = { onChange(payload.copy(thinking = it)) },
placeholder = "Optional override",
label = "Thinking",
enabled = enabled,
modifier = Modifier.weight(1f),
)
}
}
is GatewayCronPayloadEdit.Command -> {
val commandCwdCanBeCleared = originalCommandCwd == null
ClawTextField(
value = payload.argvJson,
onValueChange = { onChange(payload.copy(argvJson = it)) },
placeholder = "Command argv JSON array",
label = "Arguments",
enabled = enabled,
minLines = 2,
)
ClawTextField(
value = payload.cwd,
onValueChange = { value ->
if (commandCwdCanBeCleared || value.isNotBlank()) {
onChange(payload.copy(cwd = value))
}
},
placeholder = "Optional path",
label =
if (commandCwdCanBeCleared) {
"Command working directory"
} else {
"Command working directory · cannot clear"
},
enabled = enabled,
)
if (!commandCwdCanBeCleared) {
Text(
text = "The gateway can change this path but cannot clear an existing path.",
style = ClawTheme.type.caption,
color = ClawTheme.colors.textMuted,
)
}
}
}
}
@Composable
private fun CronRunHistoryPanel(
jobId: String,
state: GatewayCronRunHistoryState,
onRefresh: () -> Unit,
) {
val loading = (state as? GatewayCronRunHistoryState.Loading)?.id == jobId
val runs = (state as? GatewayCronRunHistoryState.Loaded)?.takeIf { it.id == jobId }?.runs.orEmpty()
val error = (state as? GatewayCronRunHistoryState.Error)?.takeIf { it.id == jobId }?.message
Row(
modifier = Modifier.fillMaxWidth(),
verticalAlignment = Alignment.CenterVertically,
horizontalArrangement = Arrangement.spacedBy(8.dp),
) {
Icon(
imageVector = Icons.Default.History,
contentDescription = null,
modifier = Modifier.size(17.dp),
tint = ClawTheme.colors.text,
)
Text(
text = "Recent Runs",
style = ClawTheme.type.section,
color = ClawTheme.colors.text,
modifier = Modifier.weight(1f),
)
ClawSecondaryButton(
text = if (loading) "Loading" else "Reload",
onClick = onRefresh,
enabled = !loading,
icon = Icons.Default.Refresh,
)
}
when {
error != null ->
ClawPanel {
Text(text = error, style = ClawTheme.type.body, color = ClawTheme.colors.warning)
}
runs.isEmpty() ->
ClawPanel {
Text(
text = if (loading) "Loading recent runs…" else "No recent runs yet.",
style = ClawTheme.type.body,
color = ClawTheme.colors.textMuted,
)
}
else -> ClawListPanel(items = runs) { run -> CronRunHistoryRow(run) }
}
}
@Composable
private fun CronRunHistoryRow(run: GatewayCronRunSummary) {
val status = cronRunStatus(run.status)
ClawDetailRow(
title = DateFormat.getDateTimeInstance(DateFormat.SHORT, DateFormat.SHORT).format(Date(run.ts)),
subtitle = cronRunSubtitle(run),
leading = { ClawIconBadge(icon = Icons.Default.Schedule) },
trailing = { ClawStatusPill(text = cronRunStatusText(run.status), status = status) },
)
}
private fun cronScheduleKindLabel(schedule: GatewayCronScheduleEdit): String =
when (schedule) {
is GatewayCronScheduleEdit.At -> "One time"
is GatewayCronScheduleEdit.Every -> "Interval"
is GatewayCronScheduleEdit.Cron -> "Cron"
is GatewayCronScheduleEdit.OnExit -> "On command exit"
}
private fun cronPayloadKindLabel(payload: GatewayCronPayloadEdit): String =
when (payload) {
is GatewayCronPayloadEdit.SystemEvent -> "System event"
is GatewayCronPayloadEdit.AgentTurn -> "Agent turn"
is GatewayCronPayloadEdit.Command -> "Command"
}
private fun cronRunSubtitle(run: GatewayCronRunSummary): String =
listOfNotNull(
run.durationMs?.let { "${it}ms" },
run.deliveryStatus,
run.model,
run.error ?: run.summary,
).joinToString(" · ").ifBlank { "No details" }
private fun cronRunStatusText(status: String?): String =
when (status?.lowercase()) {
"ok" -> "OK"
"error" -> "Issue"
"skipped" -> "Skipped"
else -> "Unknown"
}
private fun cronRunStatus(status: String?): ClawStatus =
when (status?.lowercase()) {
"ok" -> ClawStatus.Success
"error" -> ClawStatus.Danger
"skipped" -> ClawStatus.Warning
else -> ClawStatus.Neutral
}
@@ -4,12 +4,16 @@ import ai.openclaw.app.AndroidLicenseNotice
import ai.openclaw.app.AppLanguage
import ai.openclaw.app.AppearanceThemeMode
import ai.openclaw.app.BuildConfig
import ai.openclaw.app.CronEditorDraftState
import ai.openclaw.app.GatewayAgentSummary
import ai.openclaw.app.GatewayConnectionDisplay
import ai.openclaw.app.GatewayConnectionProblem
import ai.openclaw.app.GatewayCronActionState
import ai.openclaw.app.GatewayCronJobDetail
import ai.openclaw.app.GatewayCronJobDetailState
import ai.openclaw.app.GatewayCronJobEdit
import ai.openclaw.app.GatewayCronJobSummary
import ai.openclaw.app.GatewayCronRunHistoryState
import ai.openclaw.app.GatewayExecApprovalSummary
import ai.openclaw.app.GatewayTalkSetupReadiness
import ai.openclaw.app.GatewayTalkSetupState
@@ -31,6 +35,7 @@ import ai.openclaw.app.loadAndroidLicenseNotices
import ai.openclaw.app.locationModeAfterBackgroundSettings
import ai.openclaw.app.node.DeviceNotificationListenerService
import ai.openclaw.app.photoReadPermissionsForRequest
import ai.openclaw.app.reconcileRestoredAction
import ai.openclaw.app.setAppLanguage
import ai.openclaw.app.ui.design.ClawDetailRow
import ai.openclaw.app.ui.design.ClawIconBadge
@@ -66,6 +71,7 @@ import android.os.Looper
import android.provider.Settings
import android.widget.Toast
import androidx.activity.compose.BackHandler
import androidx.activity.compose.LocalActivity
import androidx.activity.compose.rememberLauncherForActivityResult
import androidx.activity.result.contract.ActivityResultContracts
import androidx.compose.foundation.BorderStroke
@@ -299,7 +305,7 @@ private fun CronJobsSettingsScreen(
)
ClawSecondaryButton(text = if (cronRefreshing) "Refreshing" else "Refresh", onClick = viewModel::refreshCronJobs, enabled = isConnected && !cronRefreshing, modifier = Modifier.fillMaxWidth())
ClawPanel {
Text(text = "Android shows scheduled work status. Create and edit schedules from the desktop app.", style = ClawTheme.type.body, color = ClawTheme.colors.textMuted)
Text(text = "Open a job to inspect its configuration and run history. Admin-scoped connections can also run, edit, enable, disable, or delete it.", style = ClawTheme.type.body, color = ClawTheme.colors.textMuted)
}
cronErrorText?.let { errorText ->
ClawPanel {
@@ -315,7 +321,7 @@ private fun CronJobsSettingsScreen(
ClawPanel {
Column(verticalArrangement = Arrangement.spacedBy(3.dp)) {
Text(text = "No scheduled jobs.", style = ClawTheme.type.section, color = ClawTheme.colors.text)
Text(text = "Create recurring OpenClaw work from the desktop app.", style = ClawTheme.type.body, color = ClawTheme.colors.textMuted)
Text(text = "Scheduled work created on the gateway will appear here.", style = ClawTheme.type.body, color = ClawTheme.colors.textMuted)
}
}
else -> CronJobsPanel(jobs = cronJobs, onJobClick = { selectedJobId = it.id })
@@ -330,13 +336,29 @@ private fun CronJobDetailSettingsScreen(
jobName: String?,
onBack: () -> Unit,
) {
BackHandler(onBack = onBack)
fun leaveDetail() {
viewModel.cronEditorDraftMemory.clear(jobId)
viewModel.dismissCronActionNotice(jobId)
onBack()
}
BackHandler(onBack = ::leaveDetail)
val detailState by viewModel.cronJobDetailState.collectAsState()
val historyState by viewModel.cronRunHistoryState.collectAsState()
val actionState by viewModel.cronActionState.collectAsState()
val pendingCronRunJobIds by viewModel.pendingCronRunJobIds.collectAsState()
val operatorAdminScopeAvailable by viewModel.operatorAdminScopeAvailable.collectAsState()
val isConnected by viewModel.isConnected.collectAsState()
val activity = LocalActivity.current
DisposableEffect(viewModel, jobId) {
onDispose { viewModel.clearCronJobDetail() }
DisposableEffect(activity, viewModel, jobId) {
onDispose {
viewModel.clearCronJobDetail()
if (cronDetailDisposalClearsTransientState(activity?.isChangingConfigurations == true)) {
viewModel.cronEditorDraftMemory.clear(jobId)
viewModel.dismissCronActionNotice(jobId)
}
}
}
LaunchedEffect(isConnected, jobId) {
@@ -346,18 +368,75 @@ private fun CronJobDetailSettingsScreen(
}
val current = (detailState as? GatewayCronJobDetailState.Loaded)?.job?.takeIf { it.id == jobId }
var editorDraft by remember(viewModel, jobId) {
mutableStateOf(viewModel.cronEditorDraftMemory.get(jobId))
}
var restoredDraftNeedsActionCheck by remember(viewModel, jobId) {
mutableStateOf(editorDraft?.savePending == true)
}
fun updateEditorDraft(value: CronEditorDraftState?) {
editorDraft = value
viewModel.cronEditorDraftMemory.set(jobId, value)
}
LaunchedEffect(isConnected, actionState, restoredDraftNeedsActionCheck) {
if (restoredDraftNeedsActionCheck) {
updateEditorDraft(
editorDraft?.reconcileRestoredAction(
isConnected = isConnected,
jobId = jobId,
actionState = actionState,
),
)
restoredDraftNeedsActionCheck = false
}
}
LaunchedEffect(isConnected) {
if (!isConnected) updateEditorDraft(editorDraft?.saveAborted())
}
LaunchedEffect(current) {
current?.let { job ->
updateEditorDraft(editorDraft?.observeJob(job) ?: CronEditorDraftState.from(job))
}
}
LaunchedEffect(actionState, current) {
val notice = actionState as? GatewayCronActionState.Notice
if (notice?.id == jobId) {
val observed = editorDraft?.observeSaveNotice(notice.kind)
updateEditorDraft(
current?.let { job ->
observed?.observeJob(job) ?: CronEditorDraftState.from(job)
} ?: observed,
)
}
}
val loading = (detailState as? GatewayCronJobDetailState.Loading)?.id == jobId
val errorText = (detailState as? GatewayCronJobDetailState.Error)?.takeIf { it.id == jobId }?.message
val deleted =
(actionState as? GatewayCronActionState.Notice)
?.takeIf { it.id == jobId }
?.deleted == true
LaunchedEffect(deleted) {
if (deleted) leaveDetail()
}
SettingsDetailFrame(
title = current?.name ?: jobName ?: "Cron Job",
subtitle = "Inspect scheduled gateway work.",
icon = Icons.Default.Bolt,
onBack = onBack,
onBack = ::leaveDetail,
) {
ClawSecondaryButton(
text = if (loading) "Refreshing" else "Refresh",
onClick = { viewModel.loadCronJobDetail(jobId) },
enabled = isConnected && !loading,
enabled =
cronDetailRefreshEnabled(
isConnected = isConnected,
loading = loading,
hasCurrentJob = current != null,
draftRequiresResolution = editorDraft?.requiresResolution == true,
saveSucceeded = editorDraft?.saveSucceeded == true,
),
modifier = Modifier.fillMaxWidth(),
)
@@ -374,11 +453,40 @@ private fun CronJobDetailSettingsScreen(
ClawPanel {
Text(text = if (loading) "Loading cron job…" else "Cron job not loaded.", style = ClawTheme.type.body, color = ClawTheme.colors.textMuted)
}
else -> CronJobDetailPanel(current)
else ->
CronJobDetailPanel(
job = current,
editorDraft = editorDraft ?: CronEditorDraftState.from(current),
onEditorDraftChange = ::updateEditorDraft,
historyState = historyState,
actionState = actionState,
runPending = jobId in pendingCronRunJobIds,
operatorAdminScopeAvailable = operatorAdminScopeAvailable,
onRun = { viewModel.runCronJob(current.id) },
onToggleEnabled = {
viewModel.setCronJobEnabled(id = current.id, enabled = !current.enabled)
},
onSave = { edit -> viewModel.updateCronJob(original = current, edit = edit) },
onRefreshHistory = { viewModel.refreshCronRunHistory(current.id) },
onDelete = { viewModel.deleteCronJob(current.id) },
)
}
}
}
internal fun cronDetailRefreshEnabled(
isConnected: Boolean,
loading: Boolean,
hasCurrentJob: Boolean,
draftRequiresResolution: Boolean,
saveSucceeded: Boolean,
): Boolean =
isConnected &&
!loading &&
(!hasCurrentJob || !draftRequiresResolution || saveSucceeded)
internal fun cronDetailDisposalClearsTransientState(isChangingConfigurations: Boolean): Boolean = !isChangingConfigurations
@Composable
private fun AgentsSettingsScreen(
viewModel: MainViewModel,
@@ -1905,7 +2013,7 @@ private fun CronJobListRow(
ClawDetailRow(
title = job.name,
subtitle = cronJobSubtitle(job),
modifier = Modifier.clickable(onClick = onClick),
modifier = Modifier.clickable(onClickLabel = "Open cron job detail", onClick = onClick),
leading = { ClawIconBadge(icon = Icons.Default.Bolt) },
trailing = {
Row(verticalAlignment = Alignment.CenterVertically, horizontalArrangement = Arrangement.spacedBy(6.dp)) {
@@ -1919,7 +2027,32 @@ private fun CronJobListRow(
@Composable
private fun CronJobDetailPanel(
job: GatewayCronJobDetail,
editorDraft: CronEditorDraftState,
onEditorDraftChange: (CronEditorDraftState) -> Unit,
historyState: GatewayCronRunHistoryState,
actionState: GatewayCronActionState,
runPending: Boolean,
operatorAdminScopeAvailable: Boolean,
onRun: () -> Unit,
onToggleEnabled: () -> Unit,
onSave: (GatewayCronJobEdit) -> Unit,
onRefreshHistory: () -> Unit,
onDelete: () -> Unit,
) {
CronJobManagementPanel(
job = job,
editorDraft = editorDraft,
onEditorDraftChange = onEditorDraftChange,
historyState = historyState,
actionState = actionState,
runPending = runPending,
operatorAdminScopeAvailable = operatorAdminScopeAvailable,
onRun = onRun,
onToggleEnabled = onToggleEnabled,
onSave = onSave,
onRefreshHistory = onRefreshHistory,
onDelete = onDelete,
)
SettingsMetricPanel(
rows =
listOf(
@@ -45,6 +45,8 @@ import androidx.compose.ui.draw.clip
import androidx.compose.ui.graphics.Color
import androidx.compose.ui.graphics.SolidColor
import androidx.compose.ui.graphics.vector.ImageVector
import androidx.compose.ui.semantics.contentDescription
import androidx.compose.ui.semantics.semantics
import androidx.compose.ui.text.style.TextOverflow
import androidx.compose.ui.unit.dp
import androidx.compose.ui.unit.sp
@@ -506,26 +508,39 @@ internal fun ClawTextField(
placeholder: String,
modifier: Modifier = Modifier,
minLines: Int = 1,
label: String? = null,
enabled: Boolean = true,
) {
val fieldModifier =
if (label == null) modifier else modifier.semantics { contentDescription = label }
BasicTextField(
value = value,
onValueChange = onValueChange,
enabled = enabled,
modifier =
modifier
fieldModifier
.fillMaxWidth()
.clip(RoundedCornerShape(ClawTheme.radii.control))
.background(ClawTheme.colors.surfaceRaised)
.border(1.dp, ClawTheme.colors.border, RoundedCornerShape(ClawTheme.radii.control))
.padding(horizontal = 11.dp, vertical = 8.dp),
textStyle = ClawTheme.type.body.copy(color = ClawTheme.colors.text),
textStyle =
ClawTheme.type.body.copy(
color = if (enabled) ClawTheme.colors.text else ClawTheme.colors.textSubtle,
),
cursorBrush = SolidColor(ClawTheme.colors.primary),
minLines = minLines,
decorationBox = { innerTextField ->
Box(modifier = Modifier.fillMaxWidth()) {
if (value.isEmpty()) {
Text(text = placeholder, style = ClawTheme.type.body, color = ClawTheme.colors.textSubtle)
Column(verticalArrangement = Arrangement.spacedBy(2.dp)) {
label?.let {
Text(text = it, style = ClawTheme.type.caption, color = ClawTheme.colors.textMuted)
}
Box(modifier = Modifier.fillMaxWidth()) {
if (value.isEmpty()) {
Text(text = placeholder, style = ClawTheme.type.body, color = ClawTheme.colors.textSubtle)
}
innerTextField()
}
innerTextField()
}
},
)
@@ -23,6 +23,22 @@ class AndroidScreenshotFixtureTest {
json
.parseToJsonElement(AndroidScreenshotFixture.request("chat.metadata", null))
.jsonObject
val cronJobs =
json
.parseToJsonElement(AndroidScreenshotFixture.request("cron.list", null))
.jsonObject["jobs"]
?.jsonArray
.orEmpty()
val cronDetail =
json
.parseToJsonElement(AndroidScreenshotFixture.request("cron.get", null))
.jsonObject
val cronRunEntries =
json
.parseToJsonElement(AndroidScreenshotFixture.request("cron.runs", null))
.jsonObject["entries"]
?.jsonArray
val parsedCronRuns = parseGatewayCronRunHistory(cronRunEntries)
assertEquals(3, sessions.size)
assertEquals(
@@ -35,6 +51,20 @@ class AndroidScreenshotFixtureTest {
)
assertEquals(1, metadata["models"]?.jsonArray?.size)
assertEquals(1, metadata["commands"]?.jsonArray?.size)
assertEquals(
AndroidScreenshotFixture.cronJobName,
cronJobs
.single()
.jsonObject["name"]
?.jsonPrimitive
?.content,
)
assertEquals(AndroidScreenshotFixture.cronJobId, cronDetail["id"]?.jsonPrimitive?.content)
assertEquals(2, parsedCronRuns.size)
assertEquals("android-release-digest-run-2", parsedCronRuns.first().runId)
assertEquals("Release checklist ready", parsedCronRuns.first().summary)
assertEquals("android-release-digest-run-1", parsedCronRuns.last().runId)
assertEquals("Play publish blocked", parsedCronRuns.last().error)
}
@Test
@@ -18,10 +18,17 @@ class CronJobDetailTest {
requireNotNull(detail)
assertEquals("job-1", detail.id)
assertEquals("Daily report", detail.name)
assertEquals("sha256:fixture", detail.configRevision)
assertEquals("cron", detail.scheduleKind)
assertEquals("0 9 * * *", detail.scheduleLabel)
assertEquals("0 9 * * * · Europe/Vienna · Stagger Every 5m", detail.scheduleDetail)
assertEquals("0 9 * * *", detail.scheduleCronExpr)
assertEquals("Europe/Vienna", detail.scheduleTimezone)
assertEquals(300000L, detail.scheduleStaggerMs)
assertEquals("Agent turn · openai/gpt-5.5 · Thinking high", detail.payloadLabel)
assertEquals("Summarize the day", detail.payloadText)
assertEquals("openai/gpt-5.5", detail.payloadModel)
assertEquals("high", detail.payloadThinking)
assertEquals("Announce · telegram · chat-42 · Account primary", detail.deliveryLabel)
assertEquals("After 3 · Announce · telegram · ops · Cooldown Every 1h", detail.failureAlertLabel)
assertEquals(2L, detail.consecutiveErrors)
@@ -40,6 +47,7 @@ class CronJobDetailTest {
requireNotNull(detail)
assertEquals("printf done", detail.payloadText)
assertEquals(listOf("printf", "done"), detail.payloadCommandArgv)
assertFalse(detail.payloadText.orEmpty().contains("secret-value"))
}
@@ -75,6 +83,24 @@ class CronJobDetailTest {
assertNull(guard.begin(" "))
}
@Test
fun requestGuardConditionsReloadAndCancellationOnCurrentSelection() {
val guard = CronJobDetailRequestGuard()
requireNotNull(guard.begin("job-a"))
requireNotNull(guard.begin("job-b"))
var loadingId = "none"
var cancelled = false
assertNull(guard.beginIfCurrent("job-a") { loadingId = it.id })
val reload = guard.beginIfCurrent("job-b") { loadingId = it.id }
assertEquals("job-b", reload?.id)
assertEquals("job-b", loadingId)
assertFalse(guard.cancelIfCurrent("job-a") { cancelled = true })
assertFalse(cancelled)
assertTrue(guard.cancelIfCurrent("job-b") { cancelled = true })
assertTrue(cancelled)
}
private fun parseJob(
payload: String =
"""{"kind":"agentTurn","message":"Summarize the day","model":"openai/gpt-5.5","thinking":"high"}""",
@@ -90,6 +116,7 @@ class CronJobDetailTest {
"deleteAfterRun": false,
"createdAtMs": 1000,
"updatedAtMs": 2000,
"configRevision": "sha256:fixture",
"schedule": {"kind":"cron","expr":"0 9 * * *","tz":"Europe/Vienna","staggerMs":300000},
"sessionTarget": "isolated",
"wakeMode": "now",
@@ -0,0 +1,472 @@
package ai.openclaw.app
import ai.openclaw.app.gateway.GatewayConnectErrorDetails
import ai.openclaw.app.gateway.GatewaySession
import kotlinx.serialization.json.Json
import kotlinx.serialization.json.JsonNull
import kotlinx.serialization.json.JsonObject
import kotlinx.serialization.json.jsonArray
import kotlinx.serialization.json.jsonObject
import kotlinx.serialization.json.jsonPrimitive
import org.junit.Assert.assertEquals
import org.junit.Assert.assertFalse
import org.junit.Assert.assertTrue
import org.junit.Test
class CronJobManagementTest {
@Test
fun parsesEveryClosedCronRunOutcome() {
val started = parseGatewayCronRunOutcome(objectJson("""{"ok":true,"ran":true}"""))
val queued =
parseGatewayCronRunOutcome(
objectJson("""{"ok":true,"enqueued":true,"runId":"run-1"}"""),
)
assertEquals(GatewayCronRunOutcome.Started(runId = null), started)
assertEquals(GatewayCronRunOutcome.Started(runId = "run-1"), queued)
mapOf(
"not-due" to GatewayCronRunSkipReason.NotDue,
"already-running" to GatewayCronRunSkipReason.AlreadyRunning,
"restart-recovery-pending" to GatewayCronRunSkipReason.RestartRecoveryPending,
"invalid-spec" to GatewayCronRunSkipReason.InvalidSpec,
"stopped" to GatewayCronRunSkipReason.Stopped,
).forEach { (raw, reason) ->
assertEquals(
GatewayCronRunOutcome.Skipped(reason),
parseGatewayCronRunOutcome(
objectJson("""{"ok":true,"ran":false,"reason":"$raw"}"""),
),
)
}
assertEquals(
GatewayCronRunOutcome.Rejected,
parseGatewayCronRunOutcome(objectJson("""{"ok":false}""")),
)
assertEquals(null, parseGatewayCronRunOutcome(objectJson("""{"ok":true,"ran":false,"reason":"future"}""")))
assertEquals(null, parseGatewayCronRunOutcome(objectJson("""{"ok":true,"enqueued":true}""")))
}
@Test
fun updatePatchIsMinimalAndClearsAgentOverridesWithNull() {
val original = requireNotNull(parseGatewayCronJobDetail(jobJson()))
val initial = original.toCronJobEdit()
val payload = initial.payload as GatewayCronPayloadEdit.AgentTurn
val edit = initial.copy(payload = payload.copy(model = "", thinking = ""))
val root = objectJson(buildCronUpdateParams(original = original, edit = edit))
val patch = root.getValue("patch").jsonObject
val payloadPatch = patch.getValue("payload").jsonObject
assertEquals("sha256:fixture", root.getValue("expectedConfigRevision").jsonPrimitive.content)
assertEquals(setOf("payload"), patch.keys)
assertEquals("agentTurn", payloadPatch.getValue("kind").jsonPrimitive.content)
assertEquals(JsonNull, payloadPatch["model"])
assertEquals(JsonNull, payloadPatch["thinking"])
assertFalse(payloadPatch.containsKey("message"))
}
@Test
fun intervalPatchPreservesAnchorAndOmitsUnchangedPayload() {
val original =
requireNotNull(
parseGatewayCronJobDetail(
jobJson(schedule = """{"kind":"every","everyMs":60000,"anchorMs":1000}"""),
),
)
val initial = original.toCronJobEdit()
val schedule = initial.schedule as GatewayCronScheduleEdit.Every
val edit = initial.copy(schedule = schedule.copy(everyMs = "120000"))
val patch =
objectJson(buildCronUpdateParams(original = original, edit = edit))
.getValue("patch")
.jsonObject
val schedulePatch = patch.getValue("schedule").jsonObject
assertEquals(setOf("schedule"), patch.keys)
assertEquals("120000", schedulePatch.getValue("everyMs").jsonPrimitive.content)
assertEquals("1000", schedulePatch.getValue("anchorMs").jsonPrimitive.content)
}
@Test
fun deleteAfterRunStaysAvailableOnlyForOneShotSchedules() {
val recurring =
requireNotNull(
parseGatewayCronJobDetail(jobJson(deleteAfterRun = true)),
).toCronJobEdit()
val oneShot =
requireNotNull(
parseGatewayCronJobDetail(
jobJson(
deleteAfterRun = true,
schedule = """{"kind":"at","at":"2026-07-10T09:00:00Z"}""",
),
),
).toCronJobEdit()
assertFalse(recurring.deleteAfterRun)
assertTrue(oneShot.deleteAfterRun)
assertFalse(
oneShot
.withSchedule(GatewayCronScheduleEdit.Every(everyMs = "60000", anchorMs = ""))
.deleteAfterRun,
)
}
@Test
fun commandArgvRejectsNonStringJsonPrimitives() {
val original =
requireNotNull(
parseGatewayCronJobDetail(
jobJson(payload = """{"kind":"command","argv":["echo"],"cwd":"/tmp"}"""),
),
)
val initial = original.toCronJobEdit()
val payload = initial.payload as GatewayCronPayloadEdit.Command
val edit = initial.copy(payload = payload.copy(argvJson = """["echo",1,true,null]"""))
val error = runCatching { buildCronUpdateParams(original = original, edit = edit) }.exceptionOrNull()
assertEquals("Command argv entries must be non-empty strings.", error?.message)
}
@Test
fun commandArgvPreservesWhitespaceOnlyEntriesAllowedByGateway() {
val original =
requireNotNull(
parseGatewayCronJobDetail(
jobJson(payload = """{"kind":"command","argv":["printf"," "],"cwd":"/tmp"}"""),
),
)
val edit = original.toCronJobEdit().copy(name = "Renamed command")
val patch =
objectJson(buildCronUpdateParams(original = original, edit = edit))
.getValue("patch")
.jsonObject
assertEquals(setOf("name"), patch.keys)
assertEquals("Renamed command", patch.getValue("name").jsonPrimitive.content)
}
@Test
fun commandPayloadRejectsClearingAnExistingWorkingDirectory() {
val original =
requireNotNull(
parseGatewayCronJobDetail(
jobJson(payload = """{"kind":"command","argv":["echo"],"cwd":"/tmp"}"""),
),
)
val initial = original.toCronJobEdit()
val payload = initial.payload as GatewayCronPayloadEdit.Command
val error =
runCatching {
buildCronUpdateParams(
original = original,
edit = initial.copy(payload = payload.copy(cwd = "")),
)
}.exceptionOrNull()
assertEquals("The gateway does not support clearing a command working directory.", error?.message)
}
@Test
fun historyParserRequiresTimestampAndKeepsUsefulFields() {
val entries =
Json
.parseToJsonElement(
"""
[
{"ts":1000,"runId":"run-1","status":"ok","summary":"done","durationMs":42},
{"runId":"missing-ts"}
]
""".trimIndent(),
).jsonArray
val runs = parseGatewayCronRunHistory(entries)
assertEquals(1, runs.size)
assertEquals("run-1", runs.single().runId)
assertEquals(42L, runs.single().durationMs)
}
@Test
fun invalidSpecSkipRefreshesPersistedDiagnosticsWithoutRefreshingOtherSkips() {
assertTrue(cronRunShouldRefresh(GatewayCronRunOutcome.Started(runId = "run-1")))
assertTrue(
cronRunShouldRefresh(
GatewayCronRunOutcome.Skipped(GatewayCronRunSkipReason.InvalidSpec),
),
)
assertFalse(
cronRunShouldRefresh(
GatewayCronRunOutcome.Skipped(GatewayCronRunSkipReason.AlreadyRunning),
),
)
assertFalse(cronRunShouldRefresh(GatewayCronRunOutcome.Rejected))
}
@Test
fun queuedRunCompletionNoticeMatchesTerminalHistoryStatus() {
listOf(
Triple("ok", "Cron run finished.", GatewayCronNoticeKind.Success),
Triple("skipped", "Cron run skipped.", GatewayCronNoticeKind.Warning),
Triple("error", "Cron run failed.", GatewayCronNoticeKind.Error),
Triple(null, "Cron run finished with an unknown status.", GatewayCronNoticeKind.Warning),
).forEach { (status, message, kind) ->
assertEquals(
GatewayCronActionState.Notice(id = "job", message = message, kind = kind),
cronRunCompletionNotice("job", status),
)
}
}
@Test
fun pendingRunRegistryDedupesOnlyTheSameJobAndIgnoresStaleTrackers() {
val registry = PendingCronRunRegistry()
val snapshots = mutableListOf<Set<String>>()
assertTrue(registry.begin("job-a", "run-a") { snapshots += it })
assertFalse(registry.begin("job-a", "run-a-duplicate") { snapshots += it })
assertTrue(registry.begin("job-b", "run-b") { snapshots += it })
assertTrue(registry.contains("job-a"))
assertTrue(registry.contains("job-b"))
assertFalse(registry.finish("job-a", "stale-run") { snapshots += it })
assertTrue(registry.finish("job-a", "run-a") { snapshots += it })
assertFalse(registry.contains("job-a"))
assertTrue(registry.contains("job-b"))
registry.clear { snapshots += it }
assertFalse(registry.contains("job-b"))
assertEquals(
listOf(setOf("job-a"), setOf("job-a", "job-b"), setOf("job-b"), emptySet()),
snapshots,
)
}
@Test
fun mapsCronRevisionConflicts() {
val conflict =
GatewaySession.ErrorShape(
code = "INVALID_REQUEST",
message = "changed",
details =
GatewayConnectErrorDetails(
code = "CRON_JOB_CHANGED",
canRetryWithDeviceToken = false,
recommendedNextStep = null,
),
)
val generic = conflict.copy(details = conflict.details?.copy(code = "OTHER"))
assertTrue(isCronJobRevisionConflict(conflict))
assertFalse(isCronJobRevisionConflict(generic))
}
@Test
fun updateFailsClosedWhenGatewayDoesNotProvideConfigRevision() {
val original = requireNotNull(parseGatewayCronJobDetail(jobJson(configRevision = null)))
val error =
runCatching {
buildCronUpdateParams(
original = original,
edit = original.toCronJobEdit().copy(name = "Renamed"),
)
}.exceptionOrNull()
assertEquals("Update the gateway before saving cron changes from Android.", error?.message)
}
@Test
fun detailAndHistoryGenerationsAdvanceIndependently() {
val detailGuard = CronJobDetailRequestGuard()
val historyGuard = CronJobDetailRequestGuard()
val detailA = requireNotNull(detailGuard.begin("job-a"))
val historyA = requireNotNull(historyGuard.begin("job-a"))
val historyB = requireNotNull(historyGuard.begin("job-b"))
var detailPublished = false
var historyPublished = "none"
assertTrue(detailGuard.publishIfCurrent(detailA) { detailPublished = true })
assertFalse(historyGuard.publishIfCurrent(historyA) { historyPublished = "a" })
assertTrue(historyGuard.publishIfCurrent(historyB) { historyPublished = "b" })
assertTrue(detailPublished)
assertEquals("b", historyPublished)
}
@Test
fun editorDraftPreservesDirtyFieldsAndMarksIncomingRevisionConflict() {
val original = requireNotNull(parseGatewayCronJobDetail(jobJson()))
var draft = CronEditorDraftState.from(original)
draft = draft.withEdit(draft.edit.copy(name = "Unsaved name"))
val unrelated =
requireNotNull(
parseGatewayCronJobDetail(
jobJson(name = "Gateway revision"),
),
)
draft = draft.observeJob(unrelated)
assertEquals("Unsaved name", draft.edit.name)
assertTrue(draft.isDirty)
assertTrue(draft.hasIncomingConflict)
assertTrue(draft.requiresResolution)
val returnedToBaseline = draft.withEdit(draft.baseline)
assertFalse(returnedToBaseline.isDirty)
assertTrue(returnedToBaseline.requiresResolution)
val reverted = CronEditorDraftState.from(unrelated)
assertEquals("Gateway revision", reverted.edit.name)
assertFalse(reverted.isDirty)
assertFalse(reverted.hasIncomingConflict)
}
@Test
fun editorDraftIgnoresRuntimeOnlyTimestampUpdates() {
val original = requireNotNull(parseGatewayCronJobDetail(jobJson()))
val draft =
CronEditorDraftState
.from(original)
.withEdit(original.toCronJobEdit().copy(name = "Unsaved name"))
val runtimeUpdate =
requireNotNull(
parseGatewayCronJobDetail(
jobJson(updatedAtMs = 3000),
),
)
val observed = draft.observeJob(runtimeUpdate)
assertEquals("Unsaved name", observed.edit.name)
assertTrue(observed.isDirty)
assertFalse(observed.hasIncomingConflict)
}
@Test
fun editorDraftAdoptsOnlyTheNewRevisionAfterSuccessfulSave() {
val original = requireNotNull(parseGatewayCronJobDetail(jobJson()))
var draft = CronEditorDraftState.from(original)
draft = draft.withEdit(draft.edit.copy(name = "Saved name"))
draft = draft.saveStarted().saveAborted()
assertFalse(draft.savePending)
assertEquals("Saved name", draft.edit.name)
draft = draft.saveStarted().observeSaveNotice(GatewayCronNoticeKind.Error)
assertEquals("Saved name", draft.edit.name)
draft = draft.saveStarted().observeSaveNotice(GatewayCronNoticeKind.Success)
assertEquals("Saved name", draft.observeJob(original).edit.name)
val saved =
requireNotNull(
parseGatewayCronJobDetail(
jobJson(name = "Saved name", updatedAtMs = 4000, configRevision = "sha256:saved"),
),
)
draft = draft.observeJob(saved)
assertEquals("Saved name", draft.baseline.name)
assertEquals(draft.baseline, draft.edit)
assertFalse(draft.savePending)
assertFalse(draft.saveSucceeded)
}
@Test
fun restoredPendingSaveTracksRetainedRuntimeAndRecoversAfterProcessDeath() {
val original = requireNotNull(parseGatewayCronJobDetail(jobJson()))
val pending =
CronEditorDraftState
.from(original)
.withEdit(original.toCronJobEdit().copy(name = "Saved name"))
.saveStarted()
val running = GatewayCronActionState.Running(id = original.id, action = GatewayCronAction.Save)
val success =
GatewayCronActionState.Notice(
id = original.id,
message = "Cron job updated.",
kind = GatewayCronNoticeKind.Success,
)
assertEquals(
pending,
pending.reconcileRestoredAction(isConnected = true, jobId = original.id, actionState = running),
)
assertEquals(
pending,
pending.reconcileRestoredAction(isConnected = true, jobId = original.id, actionState = success),
)
assertFalse(
pending
.reconcileRestoredAction(
isConnected = true,
jobId = original.id,
actionState = GatewayCronActionState.Idle,
).savePending,
)
assertFalse(
pending
.reconcileRestoredAction(
isConnected = false,
jobId = original.id,
actionState = running,
).savePending,
)
val applied =
requireNotNull(
parseGatewayCronJobDetail(
jobJson(name = "Saved name", updatedAtMs = 4000),
),
)
val recovered = pending.saveAborted().observeJob(applied)
assertEquals(recovered.baseline, recovered.edit)
assertFalse(recovered.requiresResolution)
}
@Test
fun latestRefreshGuardRejectsStaleAndInvalidatedResults() {
val guard = LatestGatewayRefreshGuard()
val stale = guard.begin()
val current = guard.begin()
var published = "none"
assertFalse(guard.publishIfCurrent(stale) { published = "stale" })
assertTrue(guard.publishIfCurrent(current) { published = "current" })
guard.invalidate()
assertFalse(guard.publishIfCurrent(current) { published = "invalidated" })
assertEquals("current", published)
}
private fun objectJson(raw: String) = Json.parseToJsonElement(raw).jsonObject
private fun jobJson(
name: String = "Daily report",
updatedAtMs: Long = 2000,
configRevision: String? = "sha256:fixture",
deleteAfterRun: Boolean = false,
schedule: String = """{"kind":"cron","expr":"0 9 * * *","tz":"UTC"}""",
payload: String =
"""{"kind":"agentTurn","message":"Summarize the day","model":"openai/gpt-5.5","thinking":"high"}""",
): JsonObject {
val configRevisionField =
configRevision?.let { """"configRevision":"$it",""" }.orEmpty()
return objectJson(
"""
{
"id":"job-1",
"name":"$name",
"description":"Daily digest",
"enabled":true,
"deleteAfterRun":$deleteAfterRun,
"createdAtMs":1000,
"updatedAtMs":$updatedAtMs,
$configRevisionField
"schedule":$schedule,
"sessionTarget":"isolated",
"wakeMode":"next-heartbeat",
"payload":$payload,
"state":{}
}
""".trimIndent(),
)
}
}
@@ -0,0 +1,220 @@
package ai.openclaw.app
import ai.openclaw.app.gateway.GatewayEndpoint
import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.sync.Mutex
import kotlinx.coroutines.withTimeout
import org.junit.Assert.assertEquals
import org.junit.Before
import org.junit.Test
import org.junit.runner.RunWith
import org.robolectric.RobolectricTestRunner
import org.robolectric.RuntimeEnvironment
import org.robolectric.annotation.Config
import java.lang.reflect.Field
import java.util.UUID
@RunWith(RobolectricTestRunner::class)
@Config(sdk = [34])
class CronRuntimeGuardTest {
@Before
fun clearPlainPrefs() {
RuntimeEnvironment
.getApplication()
.getSharedPreferences("openclaw.node", android.content.Context.MODE_PRIVATE)
.edit()
.clear()
.commit()
}
@Test
fun nonAdminConnectionRejectsMutationBeforeGatewayRequest() {
val runtime = createTestRuntime()
seedConnectedRuntime(runtime)
runtime.runCronJob("job-1")
assertEquals(
GatewayCronActionState.Notice(
id = "job-1",
message = "Cron changes require operator.admin access.",
kind = GatewayCronNoticeKind.Error,
),
runtime.cronActionState.value,
)
}
@Test
fun activeCronActionSerializesLaterMutationCalls() =
runBlocking {
val runtime = createTestRuntime()
seedConnectedRuntime(runtime)
readField<MutableStateFlow<List<String>>>(runtime, "_operatorScopes").value =
listOf("operator.admin")
withTimeout(2_000) {
while (!runtime.operatorAdminScopeAvailable.value) delay(10)
}
val actionMutex = readField<Mutex>(runtime, "cronActionMutex")
actionMutex.lock()
try {
runtime.runCronJob("job-1")
runtime.setCronJobEnabled(id = "job-1", enabled = false)
delay(50)
assertEquals(
GatewayCronActionState.Notice(
id = "job-1",
message = "Another cron action is still finishing.",
kind = GatewayCronNoticeKind.Warning,
),
runtime.cronActionState.value,
)
} finally {
actionMutex.unlock()
}
}
@Test
fun completedDeleteDoesNotClearNewerJobSelection() {
val runtime = createTestRuntime()
val detailState = readField<MutableStateFlow<GatewayCronJobDetailState>>(runtime, "_cronJobDetailState")
val historyState = readField<MutableStateFlow<GatewayCronRunHistoryState>>(runtime, "_cronRunHistoryState")
requireNotNull(readField<CronJobDetailRequestGuard>(runtime, "cronJobDetailRequestGuard").begin("job-b"))
requireNotNull(readField<CronJobDetailRequestGuard>(runtime, "cronRunHistoryRequestGuard").begin("job-b"))
detailState.value = GatewayCronJobDetailState.Loading("job-b")
historyState.value = GatewayCronRunHistoryState.Loading("job-b")
invokeStringMethod(runtime, "clearDeletedCronSelection", "job-a")
assertEquals(GatewayCronJobDetailState.Loading("job-b"), detailState.value)
assertEquals(GatewayCronRunHistoryState.Loading("job-b"), historyState.value)
invokeStringMethod(runtime, "clearDeletedCronSelection", "job-b")
assertEquals(GatewayCronJobDetailState.Idle, detailState.value)
assertEquals(GatewayCronRunHistoryState.Idle, historyState.value)
}
@Test
fun detailDisposalRetainsNoticeUntilExplicitJobDismissal() {
val runtime = createTestRuntime()
val actionState = readField<MutableStateFlow<GatewayCronActionState>>(runtime, "_cronActionState")
val notice =
GatewayCronActionState.Notice(
id = "job-a",
message = "Cron job updated.",
kind = GatewayCronNoticeKind.Success,
)
actionState.value = notice
runtime.clearCronJobDetail()
assertEquals(notice, actionState.value)
runtime.dismissCronActionNotice("job-b")
assertEquals(notice, actionState.value)
runtime.dismissCronActionNotice("job-a")
assertEquals(GatewayCronActionState.Idle, actionState.value)
}
@Test
fun pendingCronRunSurvivesReconnectButClearsWhenGatewayScopeRetires() {
val runtime = createTestRuntime()
val registry = readField<PendingCronRunRegistry>(runtime, "pendingCronRunRegistry")
val pending = readField<MutableStateFlow<Set<String>>>(runtime, "_pendingCronRunJobIds")
assertEquals(true, registry.begin("job-1", "run-1") { pending.value = it })
invokeBooleanMethod(runtime, "clearOperatorGatewayState", false)
assertEquals(setOf("job-1"), pending.value)
invokeBooleanMethod(runtime, "clearOperatorGatewayState", true)
assertEquals(emptySet<String>(), pending.value)
}
@Test
fun runningStateBlocksMutationAfterMutexRelease() =
runBlocking {
val runtime = createTestRuntime()
seedConnectedRuntime(runtime)
readField<MutableStateFlow<List<String>>>(runtime, "_operatorScopes").value =
listOf("operator.admin")
withTimeout(2_000) {
while (!runtime.operatorAdminScopeAvailable.value) delay(10)
}
val running = GatewayCronActionState.Running(id = "job-1", action = GatewayCronAction.Save)
readField<MutableStateFlow<GatewayCronActionState>>(runtime, "_cronActionState").value = running
runtime.runCronJob("job-1")
delay(50)
assertEquals(running, runtime.cronActionState.value)
}
private fun createTestRuntime(): NodeRuntime {
val app = RuntimeEnvironment.getApplication()
val securePrefs =
app.getSharedPreferences(
"openclaw.node.cron.guard.test.${UUID.randomUUID()}",
android.content.Context.MODE_PRIVATE,
)
return NodeRuntime(app, SecurePrefs(app, securePrefsOverride = securePrefs))
}
private fun seedConnectedRuntime(runtime: NodeRuntime) {
writeField(runtime, "connectedEndpoint", GatewayEndpoint.manual("127.0.0.1", 18789))
writeField(runtime, "operatorConnected", true)
}
private fun writeField(
target: Any,
name: String,
value: Any?,
) {
findField(target, name).set(target, value)
}
private fun <T> readField(
target: Any,
name: String,
): T {
@Suppress("UNCHECKED_CAST")
return findField(target, name).get(target) as T
}
private fun findField(
target: Any,
name: String,
): Field {
var type: Class<*>? = target.javaClass
while (type != null) {
try {
return type.getDeclaredField(name).apply { isAccessible = true }
} catch (_: NoSuchFieldException) {
type = type.superclass
}
}
error("Field $name not found on ${target.javaClass.name}")
}
private fun invokeStringMethod(
target: Any,
name: String,
value: String,
) {
target.javaClass
.getDeclaredMethod(name, String::class.java)
.apply { isAccessible = true }
.invoke(target, value)
}
private fun invokeBooleanMethod(
target: Any,
name: String,
value: Boolean,
) {
target.javaClass
.getDeclaredMethod(name, java.lang.Boolean.TYPE)
.apply { isAccessible = true }
.invoke(target, value)
}
}
@@ -180,7 +180,7 @@ class GatewayNodeApprovalStateTest {
@Test
fun ignoresStaleNodeApprovalRefreshResults() {
val guard = GatewayNodeApprovalRefreshGuard()
val guard = LatestGatewayRefreshGuard()
var approvalState = GatewayNodeApprovalState.Loading
val staleRefresh = guard.begin()
val currentRefresh = guard.begin()
@@ -1,6 +1,8 @@
package ai.openclaw.app
import org.junit.Assert.assertEquals
import org.junit.Assert.assertFalse
import org.junit.Assert.assertNull
import org.junit.Assert.assertTrue
import org.junit.Test
@@ -32,4 +34,41 @@ class MainViewModelTest {
),
)
}
@Test
fun cronEditorDraftMemoryIsBoundedAndClearsOnlyItsOwningJob() {
val memory = CronEditorDraftMemory()
val first = draft("First")
val second = draft("Second")
memory.set("job-a", first)
assertEquals(first, memory.get("job-a"))
assertNull(memory.get("job-b"))
memory.set("job-b", second)
assertNull(memory.get("job-a"))
memory.clear("job-a")
assertEquals(second, memory.get("job-b"))
memory.set("job-b", null)
assertNull(memory.get("job-b"))
}
private fun draft(name: String): CronEditorDraftState {
val edit =
GatewayCronJobEdit(
name = name,
description = "",
enabled = true,
deleteAfterRun = false,
schedule = GatewayCronScheduleEdit.At("2026-07-10T09:00:00Z"),
sessionTarget = "isolated",
wakeMode = "now",
payload = GatewayCronPayloadEdit.SystemEvent("Wake up"),
)
return CronEditorDraftState(
baseline = edit,
edit = edit.copy(name = "$name draft"),
)
}
}
@@ -92,6 +92,36 @@ class SettingsScreensTest {
assertEquals(null, gatewayNodeApprovalCommand(GatewayNodeCapabilityApproval.Approved))
}
@Test
fun cronDetailRefreshRecoversWhenDirtyDraftHasNoLoadedJob() {
assertEquals(
true,
cronDetailRefreshEnabled(
isConnected = true,
loading = false,
hasCurrentJob = false,
draftRequiresResolution = true,
saveSucceeded = false,
),
)
assertEquals(
false,
cronDetailRefreshEnabled(
isConnected = true,
loading = false,
hasCurrentJob = true,
draftRequiresResolution = true,
saveSucceeded = false,
),
)
}
@Test
fun cronDetailDisposalRetainsTransientStateOnlyForActivityRecreation() {
assertEquals(false, cronDetailDisposalClearsTransientState(isChangingConfigurations = true))
assertEquals(true, cronDetailDisposalClearsTransientState(isChangingConfigurations = false))
}
private fun authProblem(code: String): GatewayConnectionProblem =
GatewayConnectionProblem(
code = code,
@@ -0,0 +1,81 @@
package ai.openclaw.app.benchmark
import androidx.test.ext.junit.runners.AndroidJUnit4
import androidx.test.platform.app.InstrumentationRegistry
import androidx.test.uiautomator.By
import androidx.test.uiautomator.UiDevice
import androidx.test.uiautomator.UiObject2
import androidx.test.uiautomator.Until
import org.junit.Assert.assertFalse
import org.junit.Assert.assertNotNull
import org.junit.Assert.assertTrue
import org.junit.Before
import org.junit.Test
import org.junit.runner.RunWith
@RunWith(AndroidJUnit4::class)
class CronJobNavigationTest {
private lateinit var device: UiDevice
@Before
fun setUp() {
device = UiDevice.getInstance(InstrumentationRegistry.getInstrumentation())
device.executeShellCommand("am force-stop $packageName")
device.executeShellCommand(
"am start -W -n $packageName/.MainActivity " +
"--ez openclaw.screenshotMode true --es openclaw.screenshotScene settings",
)
assertNotNull(device.wait(Until.findObject(By.text("Settings")), waitTimeoutMs))
}
@Test
fun opensCronJobFixtureDetail() {
findTextAfterScrolling("Cron Jobs").click()
val cronJobLabel = findTextAfterScrolling("Android release digest")
val cronJobRow =
checkNotNull(
generateSequence(cronJobLabel) { it.parent }
.firstOrNull { it.isClickable },
) { "Cron fixture row must expose a click action" }
assertTrue("Cron fixture row must expose a click action", cronJobRow.isClickable)
assertFalse(device.hasObject(By.text("Inspect scheduled gateway work.")))
cronJobRow.click()
assertNotNull(device.wait(Until.findObject(By.text("Inspect scheduled gateway work.")), waitTimeoutMs))
assertNotNull(findTextAfterScrolling("Run Now"))
assertNotNull(findTextAfterScrolling("Recent Runs"))
assertNotNull(findTextAfterScrolling("Release checklist ready", exact = false))
assertNotNull(findTextAfterScrolling("OK"))
assertNotNull(findTextAfterScrolling("Play publish blocked", exact = false))
assertNotNull(findTextAfterScrolling("Issue"))
}
private fun findTextAfterScrolling(
text: String,
exact: Boolean = true,
): UiObject2 {
val selector = if (exact) By.text(text) else By.textContains(text)
repeat(maxScrolls + 1) { attempt ->
device.wait(Until.findObject(selector), shortWaitMs)?.let { return it }
if (attempt < maxScrolls) {
device.swipe(
device.displayWidth / 2,
(device.displayHeight * 0.8f).toInt(),
device.displayWidth / 2,
(device.displayHeight * 0.25f).toInt(),
24,
)
device.waitForIdle()
}
}
error("Could not find UI text: $text")
}
private companion object {
const val packageName = "ai.openclaw.app"
const val waitTimeoutMs = 10_000L
const val shortWaitMs = 1_000L
const val maxScrolls = 6
}
}
@@ -7184,6 +7184,7 @@ public struct CronJob: Codable, Sendable {
public let deleteafterrun: Bool?
public let createdatms: Int
public let updatedatms: Int
public let configrevision: String?
public let schedule: AnyCodable
public let trigger: [String: AnyCodable]?
public let sessiontarget: AnyCodable
@@ -7216,6 +7217,7 @@ public struct CronJob: Codable, Sendable {
deleteafterrun: Bool?,
createdatms: Int,
updatedatms: Int,
configrevision: String? = nil,
schedule: AnyCodable,
trigger: [String: AnyCodable]?,
sessiontarget: AnyCodable,
@@ -7247,6 +7249,7 @@ public struct CronJob: Codable, Sendable {
self.deleteafterrun = deleteafterrun
self.createdatms = createdatms
self.updatedatms = updatedatms
self.configrevision = configrevision
self.schedule = schedule
self.trigger = trigger
self.sessiontarget = sessiontarget
@@ -7280,6 +7283,7 @@ public struct CronJob: Codable, Sendable {
case deleteafterrun = "deleteAfterRun"
case createdatms = "createdAtMs"
case updatedatms = "updatedAtMs"
case configrevision = "configRevision"
case schedule
case trigger
case sessiontarget = "sessionTarget"
@@ -36,6 +36,28 @@ describe("cron protocol validators", () => {
expect(validateCronAddParams(minimalAddParams)).toBe(true);
});
it("rejects schedule integers that SQLite cannot round-trip safely", () => {
const unsafe = Number.MAX_SAFE_INTEGER + 1;
expect(
validateCronAddParams({
...minimalAddParams,
schedule: { kind: "every", everyMs: unsafe },
}),
).toBe(false);
expect(
validateCronUpdateParams({
id: "job-1",
patch: { schedule: { kind: "every", everyMs: 60_000, anchorMs: unsafe } },
}),
).toBe(false);
expect(
validateCronUpdateParams({
id: "job-1",
patch: { schedule: { kind: "cron", expr: "0 * * * *", staggerMs: unsafe } },
}),
).toBe(false);
});
it("accepts trigger add, patch, and clear shapes", () => {
expect(
validateCronAddParams({
@@ -151,6 +173,37 @@ describe("cron protocol validators", () => {
expect(validateCronUpdateParams({ jobId: "job-2", patch: { enabled: true } })).toBe(true);
});
it("accepts only non-empty cron config revisions", () => {
expect(
validateCronUpdateParams({
id: "job-1",
expectedConfigRevision: "sha256:current",
patch: { enabled: false },
}),
).toBe(true);
expect(
validateCronUpdateParams({
id: "job-1",
expectedConfigRevision: "",
patch: { enabled: false },
}),
).toBe(false);
expect(
validateCronUpdateParams({
id: "job-1",
expectedConfigRevision: 1,
patch: { enabled: false },
}),
).toBe(false);
expect(
validateCronUpdateParams({
id: "job-1",
expectedConfigRevision: "x".repeat(129),
patch: { enabled: false },
}),
).toBe(false);
});
it("accepts nullable model clears only on update payload patches", () => {
expect(
validateCronUpdateParams({
+8 -3
View File
@@ -68,6 +68,7 @@ function cronRunStatusSchema(options: Record<string, unknown> = {}) {
}
const CronRunStatusSchema = cronRunStatusSchema();
const CronConfigRevisionSchema = Type.String({ minLength: 1, maxLength: 128 });
const DeprecatedCronRunStatusSchema = cronRunStatusSchema({
deprecated: true,
description: "Deprecated alias for lastRunStatus.",
@@ -220,8 +221,8 @@ export const CronScheduleSchema = Type.Union([
Type.Object(
{
kind: Type.Literal("every"),
everyMs: Type.Integer({ minimum: 1 }),
anchorMs: Type.Optional(Type.Integer({ minimum: 0 })),
everyMs: Type.Integer({ minimum: 1, maximum: Number.MAX_SAFE_INTEGER }),
anchorMs: Type.Optional(Type.Integer({ minimum: 0, maximum: Number.MAX_SAFE_INTEGER })),
},
{ additionalProperties: false },
),
@@ -230,7 +231,7 @@ export const CronScheduleSchema = Type.Union([
kind: Type.Literal("cron"),
expr: NonEmptyString,
tz: Type.Optional(Type.String()),
staggerMs: Type.Optional(Type.Integer({ minimum: 0 })),
staggerMs: Type.Optional(Type.Integer({ minimum: 0, maximum: Number.MAX_SAFE_INTEGER })),
},
{ additionalProperties: false },
),
@@ -490,6 +491,8 @@ export const CronJobSchema = Type.Object(
deleteAfterRun: Type.Optional(Type.Boolean()),
createdAtMs: Type.Integer({ minimum: 0 }),
updatedAtMs: Type.Integer({ minimum: 0 }),
/** Opaque Gateway-computed token for the job definition, excluding scheduler state. */
configRevision: Type.Optional(CronConfigRevisionSchema),
schedule: CronScheduleSchema,
trigger: Type.Optional(CronTriggerSchema),
sessionTarget: CronSessionTargetSchema,
@@ -589,6 +592,8 @@ export const CronJobPatchSchema = Type.Object(
/** Updates a cron job by id or legacy jobId alias. */
export const CronUpdateParamsSchema = cronIdOrJobIdParams({
patch: CronJobPatchSchema,
/** Rejects the patch when the current definition does not match the caller's token. */
expectedConfigRevision: Type.Optional(CronConfigRevisionSchema),
});
/** Removes a cron job by id or legacy jobId alias. */
+1
View File
@@ -82,6 +82,7 @@ const DEFAULTED_OPTIONAL_INIT_PARAM_ENTRIES: readonly [string, readonly string[]
"declarationKey",
"displayName",
"owner",
"configRevision",
"nextRunAtMs",
"lastRunAtMs",
"lastRunStatus",
+215
View File
@@ -0,0 +1,215 @@
import { describe, expect, it } from "vitest";
import { resolveCronJobConfigRevision } from "./config-revision.js";
import { setupCronServiceSuite } from "./service.test-harness.js";
import { loadCronStore, saveCronStore } from "./store.js";
import type { CronJob } from "./types.js";
const { makeStorePath } = setupCronServiceSuite({ prefix: "cron-config-revision-" });
function makeJob(): CronJob {
return {
id: "job-1",
name: "daily report",
enabled: true,
createdAtMs: 1_000,
updatedAtMs: 2_000,
schedule: { kind: "cron", expr: "0 9 * * *", tz: "UTC" },
sessionTarget: "isolated",
wakeMode: "next-heartbeat",
payload: { kind: "agentTurn", message: "Summarize the day" },
delivery: { mode: "announce", channel: "telegram", to: "chat-1" },
state: {},
};
}
describe("resolveCronJobConfigRevision", () => {
it("ignores runtime timestamps and scheduler state", () => {
const original = makeJob();
const cyclicState: Record<string, unknown> = {};
cyclicState.self = cyclicState;
const runtimeChanged: CronJob = {
...original,
updatedAtMs: 9_000,
state: {
lastRunAtMs: 8_000,
lastRunStatus: "ok",
nextRunAtMs: 10_000,
triggerState: cyclicState,
},
};
expect(resolveCronJobConfigRevision(runtimeChanged)).toBe(
resolveCronJobConfigRevision(original),
);
});
it("changes for definition updates and same-id recreation", () => {
const original = makeJob();
expect(resolveCronJobConfigRevision({ ...original, description: "changed" })).not.toBe(
resolveCronJobConfigRevision(original),
);
expect(resolveCronJobConfigRevision({ ...original, createdAtMs: 2_000 })).not.toBe(
resolveCronJobConfigRevision(original),
);
});
it("is stable across nested key ordering", () => {
const original = makeJob();
const reordered: CronJob = {
...original,
payload: {
kind: "command",
argv: ["printenv"],
env: { B: "2", A: "1" },
},
};
const canonical: CronJob = {
...reordered,
payload: {
kind: "command",
argv: ["printenv"],
env: { A: "1", B: "2" },
},
};
expect(resolveCronJobConfigRevision(reordered)).toBe(resolveCronJobConfigRevision(canonical));
});
it("preserves order when case-insensitive command env keys collide on Windows", () => {
const firstWinsLast: CronJob = {
...makeJob(),
payload: {
kind: "command",
argv: ["printenv"],
env: { Path: "first", PATH: "second" },
},
};
const secondWinsLast: CronJob = {
...firstWinsLast,
payload: {
kind: "command",
argv: ["printenv"],
env: { PATH: "second", Path: "first" },
},
};
expect(resolveCronJobConfigRevision(firstWinsLast)).not.toBe(
resolveCronJobConfigRevision(secondWinsLast),
);
});
it("distinguishes inherited and explicitly cleared delivery fields", () => {
const inherited = makeJob();
const explicitlyCleared: CronJob = {
...inherited,
delivery: {
mode: "announce",
channel: "telegram",
to: "chat-1",
failureDestination: { channel: undefined },
},
};
expect(resolveCronJobConfigRevision(explicitlyCleared)).not.toBe(
resolveCronJobConfigRevision(inherited),
);
});
it("is stable across the SQLite store round-trip", async () => {
const { storePath } = await makeStorePath();
const job: CronJob = {
...makeJob(),
agentId: undefined,
description: undefined,
payload: {
kind: "agentTurn",
message: "Summarize the day",
toolsAllow: ["read"],
toolsAllowIsDefault: false,
},
delivery: {
mode: "announce",
failureDestination: {
channel: undefined,
accountId: undefined,
},
},
};
await saveCronStore(storePath, { version: 1, jobs: [job] });
const reloaded = (await loadCronStore(storePath)).jobs[0];
if (!reloaded) {
throw new Error("expected the persisted cron job to reload");
}
expect(resolveCronJobConfigRevision(reloaded)).toBe(resolveCronJobConfigRevision(job));
});
it("matches SQLite normalization across schedule, payload, trigger, and alert variants", async () => {
const { storePath } = await makeStorePath();
const jobs: CronJob[] = [
{
...makeJob(),
id: "command-empty-env",
schedule: { kind: "every", everyMs: Number.MAX_SAFE_INTEGER, anchorMs: 0 },
payload: { kind: "command", argv: ["true"], env: {}, input: "" },
failureAlert: false,
},
{
...makeJob(),
id: "default-tools-without-list",
schedule: { kind: "cron", expr: "0 9 * * *", tz: "" },
payload: {
kind: "agentTurn",
message: "Summarize the day",
toolsAllowIsDefault: true,
},
failureAlert: {},
trigger: { script: "json({ fire: true })", once: true },
},
{
...makeJob(),
id: "windows-env-key-order",
payload: {
kind: "command",
argv: ["printenv"],
env: { Path: "first", PATH: "second" },
},
},
{
...makeJob(),
id: "default-empty-tools",
schedule: { kind: "at", at: "2027-01-01T00:00:00.000Z" },
payload: {
kind: "agentTurn",
message: "Summarize the day",
toolsAllow: [],
toolsAllowIsDefault: true,
},
},
{
...makeJob(),
id: "on-exit-system-event",
schedule: { kind: "on-exit", command: "true", cwd: "/tmp" },
sessionTarget: "main",
payload: { kind: "systemEvent", text: "Process exited" },
delivery: undefined,
failureAlert: { after: 2, cooldownMs: 0, includeSkipped: false },
},
];
await saveCronStore(storePath, { version: 1, jobs });
const reloadedById = new Map((await loadCronStore(storePath)).jobs.map((job) => [job.id, job]));
for (const job of jobs) {
const reloaded = reloadedById.get(job.id);
if (!reloaded) {
throw new Error(`expected persisted cron job ${job.id} to reload`);
}
expect(resolveCronJobConfigRevision(reloaded), job.id).toBe(
resolveCronJobConfigRevision(job),
);
}
});
});
+39
View File
@@ -0,0 +1,39 @@
/** Opaque revision token for cron configuration, excluding scheduler-maintained state. */
import { stableStringify } from "../agents/stable-stringify.js";
import { sha256Base64Url } from "../infra/crypto-digest.js";
import { projectCronJobThroughStorageCodec } from "./store/row-codec.js";
import type { CronJob } from "./types.js";
function configRevisionDefinition(projected: CronJob) {
const { updatedAtMs: _updatedAtMs, state: _state, ...definition } = projected;
if (definition.payload.kind !== "command" || !definition.payload.env) {
return definition;
}
const foldedKeys = new Set<string>();
const hasWindowsCollision = Object.keys(definition.payload.env).some((key) => {
const folded = key.toLowerCase();
if (foldedKeys.has(folded)) {
return true;
}
foldedKeys.add(folded);
return false;
});
if (!hasWindowsCollision) {
return definition;
}
// Windows resolves case-insensitive duplicate env keys in insertion order.
// Preserve that order only when it changes command execution semantics.
const { env, ...payload } = definition.payload;
return { ...definition, payload: { ...payload, envEntries: Object.entries(env) } };
}
/** Hashes the job definition while preserving meaningful own-undefined config fields. */
export function resolveCronJobConfigRevision(job: CronJob): string {
// The storage projector canonicalizes every persisted config seam. Feed it
// neutral runtime fields so large or malformed trigger state cannot affect the token.
const projected = projectCronJobThroughStorageCodec({ ...job, updatedAtMs: 0, state: {} });
const fingerprint = stableStringify(configRevisionDefinition(projected));
return `sha256:${sha256Base64Url(fingerprint)}`;
}
+14
View File
@@ -73,6 +73,20 @@ function normalizeMainSystemEventCreateJob(params: {
}
describe("normalizeCronJobCreate", () => {
it("trims cron timezones and drops blank values", () => {
const trimmed = normalizeMainSystemEventCreateJob({
name: "trimmed-timezone",
schedule: { kind: "cron", expr: "0 * * * *", tz: " Europe/Vienna " },
});
const blank = normalizeMainSystemEventCreateJob({
name: "blank-timezone",
schedule: { kind: "cron", expr: "0 * * * *", tz: " " },
});
expect(trimmed.schedule).toMatchObject({ tz: "Europe/Vienna" });
expect(blank.schedule).not.toHaveProperty("tz");
});
it("normalizes trigger scripts and preserves patch clears", () => {
const normalized = normalizeCronJobCreate({
name: "watcher",
+6
View File
@@ -100,6 +100,7 @@ function coerceSchedule(schedule: UnknownRecord) {
? rawKind
: undefined;
const exprRaw = normalizeOptionalString(schedule.expr) ?? "";
const timezone = normalizeOptionalString(schedule.tz);
const commandRaw = normalizeOptionalString(schedule.command) ?? "";
const cwdRaw = normalizeOptionalString(schedule.cwd) ?? "";
const everyMs = coerceFiniteScheduleNumber(schedule.everyMs);
@@ -123,6 +124,11 @@ function coerceSchedule(schedule: UnknownRecord) {
} else if ("expr" in next) {
delete next.expr;
}
if (timezone) {
next.tz = timezone;
} else if ("tz" in next) {
delete next.tz;
}
if (everyMs !== undefined && everyMs >= 1) {
next.everyMs = Math.floor(everyMs);
+4 -3
View File
@@ -1,7 +1,8 @@
/** Coerces cron schedule number fields with strict finite-number parsing. */
/** Coerces cron schedule number fields with strict safe-range parsing. */
import { parseStrictFiniteNumber } from "@openclaw/normalization-core/number-coercion";
/** Coerces schedule numeric fields without accepting partial or non-finite numbers. */
/** Coerces schedule numeric fields without accepting partial, non-finite, or unsafe values. */
export function coerceFiniteScheduleNumber(value: unknown): number | undefined {
return parseStrictFiniteNumber(value);
const parsed = parseStrictFiniteNumber(value);
return parsed !== undefined && Math.abs(parsed) <= Number.MAX_SAFE_INTEGER ? parsed : undefined;
}
+2
View File
@@ -249,6 +249,8 @@ describe("coerceFiniteScheduleNumber", () => {
expect(coerceFiniteScheduleNumber("0x10")).toBeUndefined();
expect(coerceFiniteScheduleNumber(Number.NaN)).toBeUndefined();
expect(coerceFiniteScheduleNumber(Infinity)).toBeUndefined();
expect(coerceFiniteScheduleNumber(Number.MAX_SAFE_INTEGER + 1)).toBeUndefined();
expect(coerceFiniteScheduleNumber(String(Number.MAX_SAFE_INTEGER + 1))).toBeUndefined();
expect(coerceFiniteScheduleNumber(null)).toBeUndefined();
expect(coerceFiniteScheduleNumber(undefined)).toBeUndefined();
});
+97 -5
View File
@@ -16,7 +16,7 @@ import {
} from "../../process/command-queue.js";
import { CommandLane } from "../../process/lanes.js";
import { saveCronStore } from "../store.js";
import { enqueueRun, run, start } from "./ops.js";
import { enqueueRun, remove, run, start } from "./ops.js";
import type { CronEvent } from "./state.js";
import { createCronServiceState } from "./state.js";
import { onTimer } from "./timer.js";
@@ -31,6 +31,7 @@ function expectQueuedRunAck(result: unknown) {
expect(ack.ok).toBe(true);
expect(ack.enqueued).toBe(true);
expect(typeof ack.runId).toBe("string");
return ack.runId as string;
}
function requireMockCall(
@@ -92,7 +93,11 @@ describe("cron service ops regressions", () => {
}
});
it("skips forced manual runs while a timer-triggered run is in progress", async () => {
it("records queued forced runs that lose a timer race as skipped", async () => {
vi.useRealTimers();
clearCommandLane(CommandLane.Cron);
setCommandLaneConcurrency(CommandLane.Cron, 1);
const store = opsRegressionFixtures.makeStorePath();
const dueAt = Date.now() - 1;
const job = createIsolatedRegressionJob({
@@ -105,11 +110,20 @@ describe("cron service ops regressions", () => {
});
await saveCronStore(store.storePath, { version: 1, jobs: [job] });
const blockerStarted = createDeferred<void>();
const releaseBlocker = createDeferred<void>();
const blocker = enqueueCommandInLane(CommandLane.Cron, async () => {
blockerStarted.resolve();
return await releaseBlocker.promise;
});
await blockerStarted.promise;
let resolveRun:
| ((value: { status: "ok" | "error" | "skipped"; summary?: string; error?: string }) => void)
| undefined;
const started = createDeferred<void>();
const finished = createDeferred<void>();
const events: CronEvent[] = [];
const runIsolatedAgentJob = vi.fn(
async () =>
await new Promise<{ status: "ok" | "error" | "skipped"; summary?: string; error?: string }>(
@@ -127,6 +141,7 @@ describe("cron service ops regressions", () => {
requestHeartbeat: vi.fn(),
runIsolatedAgentJob,
onEvent: (evt: CronEvent) => {
events.push(evt);
if (evt.jobId !== job.id) {
return;
}
@@ -138,17 +153,31 @@ describe("cron service ops regressions", () => {
},
});
const ack = await enqueueRun(state, job.id, "force");
const runId = expectQueuedRunAck(ack);
const timerPromise = onTimer(state);
await started.promise;
expect(runIsolatedAgentJob).toHaveBeenCalledTimes(1);
const manualResult = await run(state, job.id, "force");
expect(manualResult).toEqual({ ok: true, ran: false, reason: "already-running" });
releaseBlocker.resolve();
await blocker;
await waitForActiveTasks(5_000);
expect(runIsolatedAgentJob).toHaveBeenCalledTimes(1);
expect(events).toContainEqual(
expect.objectContaining({
jobId: job.id,
action: "finished",
status: "skipped",
error: "queued manual run skipped before execution: already-running",
runId,
}),
);
resolveRun?.({ status: "ok", summary: "done" });
await finished.promise;
await timerPromise;
clearCommandLane(CommandLane.Cron);
});
it("does not double-run a job when cron.run overlaps a due timer tick", async () => {
@@ -497,6 +526,7 @@ describe("cron service ops regressions", () => {
await blockerStarted.promise;
const runIsolatedAgentJob = vi.fn(async () => ({ status: "ok" as const }));
const events: CronEvent[] = [];
const state = createCronServiceState({
cronEnabled: true,
storePath: store.storePath,
@@ -505,10 +535,11 @@ describe("cron service ops regressions", () => {
enqueueSystemEvent: vi.fn(),
requestHeartbeat: vi.fn(),
runIsolatedAgentJob,
onEvent: (evt) => events.push(evt),
});
const ack = await enqueueRun(state, job.id, "force");
expectQueuedRunAck(ack);
const runId = expectQueuedRunAck(ack);
state.stopped = true;
releaseBlocker.resolve();
@@ -519,6 +550,67 @@ describe("cron service ops regressions", () => {
expect(
state.store?.jobs.find((entry) => entry.id === job.id)?.state.runningAtMs,
).toBeUndefined();
expect(events).toContainEqual(
expect.objectContaining({
jobId: job.id,
action: "finished",
status: "skipped",
error: "queued manual run skipped before execution: stopped",
runId,
}),
);
clearCommandLane(CommandLane.Cron);
});
it("emits one terminal event when a queued job is removed during execution", async () => {
vi.useRealTimers();
clearCommandLane(CommandLane.Cron);
setCommandLaneConcurrency(CommandLane.Cron, 1);
const store = opsRegressionFixtures.makeStorePath();
const dueAt = Date.parse("2026-02-06T10:05:04.000Z");
const job = createDueIsolatedJob({
id: "queued-removed-manual",
nowMs: dueAt,
nextRunAtMs: dueAt,
});
await saveCronStore(store.storePath, { version: 1, jobs: [job] });
const started = createDeferred<void>();
const execution = createDeferred<{ status: "ok"; summary: string }>();
const events: CronEvent[] = [];
const state = createCronServiceState({
cronEnabled: true,
storePath: store.storePath,
log: noopLogger,
nowMs: () => dueAt,
enqueueSystemEvent: vi.fn(),
requestHeartbeat: vi.fn(),
runIsolatedAgentJob: vi.fn(async () => {
started.resolve();
return await execution.promise;
}),
onEvent: (evt) => events.push(evt),
});
const ack = await enqueueRun(state, job.id, "force");
const runId = expectQueuedRunAck(ack);
await started.promise;
await expect(remove(state, job.id)).resolves.toEqual({ ok: true, removed: true });
execution.resolve({ status: "ok", summary: "completed after removal" });
await waitForActiveTasks(5_000);
const terminalEvents = events.filter((evt) => evt.action === "finished" && evt.runId === runId);
expect(terminalEvents).toEqual([
expect.objectContaining({
jobId: job.id,
status: "ok",
summary: "completed after removal",
}),
]);
expect(state.store?.jobs.some((entry) => entry.id === job.id)).toBe(false);
clearCommandLane(CommandLane.Cron);
});
+158 -40
View File
@@ -53,6 +53,7 @@ import { locked } from "./locked.js";
import { normalizeOptionalAgentId } from "./normalize.js";
import type {
CronAddOptions,
CronEvent,
CronServiceState,
CronUpdatePrecondition,
CronWakeMode,
@@ -800,6 +801,7 @@ type PreparedManualRun =
jobId: string;
runId?: string;
taskRunId?: string;
terminalTracker?: ManualRunTerminalTracker;
activeJobMarker?: CronActiveJobMarker;
startedAt: number;
executionJob: CronJob;
@@ -809,8 +811,22 @@ type PreparedManualRun =
type ManualRunOptions = {
runId?: string;
payload?: CronPayload;
terminalTracker?: ManualRunTerminalTracker;
};
type ManualRunTerminalTracker = { emitted: boolean };
function emitManualRunFinished(
state: CronServiceState,
evt: CronEvent & { action: "finished" },
tracker?: ManualRunTerminalTracker,
): void {
emit(state, evt);
if (tracker) {
tracker.emitted = true;
}
}
type ManualRunDisposition =
| Extract<PreparedManualRun, { ran: false }>
| { ok: true; runnable: true };
@@ -831,6 +847,8 @@ async function skipInvalidPersistedManualRun(params: {
state: CronServiceState;
job: CronJob;
mode?: "due" | "force";
runId?: string;
terminalTracker?: ManualRunTerminalTracker;
error: unknown;
}) {
const endedAt = params.state.deps.nowMs();
@@ -852,19 +870,24 @@ async function skipInvalidPersistedManualRun(params: {
{ preserveSchedule: params.mode === "force" },
);
emit(params.state, {
jobId: params.job.id,
action: "finished",
status: "skipped",
error: errorText,
diagnostics,
runAtMs: endedAt,
durationMs: params.job.state.lastDurationMs,
nextRunAtMs: params.job.state.nextRunAtMs,
deliveryStatus: params.job.state.lastDeliveryStatus,
deliveryError: params.job.state.lastDeliveryError,
failureNotificationDelivery: failureNotificationDeliveryFromJobState(params.job),
});
emitManualRunFinished(
params.state,
{
jobId: params.job.id,
action: "finished",
status: "skipped",
error: errorText,
diagnostics,
runId: params.runId,
runAtMs: endedAt,
durationMs: params.job.state.lastDurationMs,
nextRunAtMs: params.job.state.nextRunAtMs,
deliveryStatus: params.job.state.lastDeliveryStatus,
deliveryError: params.job.state.lastDeliveryError,
failureNotificationDelivery: failureNotificationDeliveryFromJobState(params.job),
},
params.terminalTracker,
);
if (shouldDelete && params.state.store) {
params.state.store.jobs = params.state.store.jobs.filter((entry) => entry.id !== params.job.id);
@@ -965,6 +988,8 @@ async function inspectManualRunPreflight(
state: CronServiceState,
id: string,
mode?: "due" | "force",
runId?: string,
terminalTracker?: ManualRunTerminalTracker,
): Promise<ManualRunPreflightResult> {
return await locked(state, async () => {
warnIfDisabled(state, "run");
@@ -983,7 +1008,7 @@ async function inspectManualRunPreflight(
try {
assertSupportedJobSpec(job);
} catch (error) {
await skipInvalidPersistedManualRun({ state, job, mode, error });
await skipInvalidPersistedManualRun({ state, job, mode, runId, terminalTracker, error });
return { ok: true, ran: false, reason: "invalid-spec" as const };
}
if (typeof job.state.runningAtMs === "number") {
@@ -1021,7 +1046,13 @@ async function prepareManualRun(
mode?: "due" | "force",
opts?: ManualRunOptions,
): Promise<PreparedManualRun> {
const preflight = await inspectManualRunPreflight(state, id, mode);
const preflight = await inspectManualRunPreflight(
state,
id,
mode,
opts?.runId,
opts?.terminalTracker,
);
if (!preflight.ok) {
return preflight;
}
@@ -1080,6 +1111,7 @@ async function prepareManualRun(
jobId: job.id,
runId: opts?.runId ?? taskRunId,
taskRunId,
terminalTracker: opts?.terminalTracker,
activeJobMarker,
startedAt: preflight.now,
executionJob,
@@ -1109,12 +1141,49 @@ async function finishPreparedManualRun(
coreResult = { status: "error", error: normalizeCronRunErrorText(err) };
}
const endedAt = state.deps.nowMs();
const emitMissingQueuedTerminal = () => {
const tracker = prepared.terminalTracker;
if (!tracker || tracker.emitted) {
return;
}
const job = state.store?.jobs.find((entry) => entry.id === jobId);
const triggerSkipped = coreResult.status === "ok" && coreResult.triggerEval?.fired === false;
// enqueueRun acknowledges a concrete run id, so every accepted request
// needs one terminal event even if the job or service owner changes mid-run.
emitManualRunFinished(
state,
{
jobId,
action: "finished",
job,
status: triggerSkipped ? "skipped" : coreResult.status,
error: triggerSkipped
? "queued manual run skipped: trigger condition not met"
: coreResult.error,
summary: triggerSkipped ? undefined : coreResult.summary,
diagnostics: coreResult.diagnostics,
delivered: coreResult.delivered,
delivery: coreResult.delivery,
sessionId: coreResult.sessionId,
sessionKey: coreResult.sessionKey,
runId,
runAtMs: startedAt,
durationMs: Math.max(0, endedAt - startedAt),
nextRunAtMs: job?.state.nextRunAtMs,
model: coreResult.model,
provider: coreResult.provider,
usage: coreResult.usage,
},
tracker,
);
};
tryFinishManualTaskRun(state, {
taskRunId,
coreResult,
endedAt,
});
if (!isCronActiveJobMarkerCurrent(prepared.activeJobMarker)) {
emitMissingQueuedTerminal();
return;
}
@@ -1161,30 +1230,34 @@ async function finishPreparedManualRun(
triggerEval: coreResult.triggerEval,
});
emit(state, {
jobId: job.id,
action: "finished",
job,
status: coreResult.status,
error: coreResult.error,
summary: coreResult.summary,
diagnostics: coreResult.diagnostics,
delivered: job.state.lastDelivered,
deliveryStatus: job.state.lastDeliveryStatus,
deliveryError: job.state.lastDeliveryError,
failureNotificationDelivery: failureNotificationDeliveryFromJobState(job),
delivery: coreResult.delivery,
sessionId: coreResult.sessionId,
sessionKey: coreResult.sessionKey,
runId,
runAtMs: startedAt,
durationMs: job.state.lastDurationMs,
nextRunAtMs: job.state.nextRunAtMs,
...(coreResult.triggerEval?.fired ? { triggerFired: true } : {}),
model: coreResult.model,
provider: coreResult.provider,
usage: coreResult.usage,
});
emitManualRunFinished(
state,
{
jobId: job.id,
action: "finished",
job,
status: coreResult.status,
error: coreResult.error,
summary: coreResult.summary,
diagnostics: coreResult.diagnostics,
delivered: job.state.lastDelivered,
deliveryStatus: job.state.lastDeliveryStatus,
deliveryError: job.state.lastDeliveryError,
failureNotificationDelivery: failureNotificationDeliveryFromJobState(job),
delivery: coreResult.delivery,
sessionId: coreResult.sessionId,
sessionKey: coreResult.sessionKey,
runId,
runAtMs: startedAt,
durationMs: job.state.lastDurationMs,
nextRunAtMs: job.state.nextRunAtMs,
...(coreResult.triggerEval?.fired ? { triggerFired: true } : {}),
model: coreResult.model,
provider: coreResult.provider,
usage: coreResult.usage,
},
prepared.terminalTracker,
);
}
if (shouldDelete && state.store) {
@@ -1230,6 +1303,7 @@ async function finishPreparedManualRun(
if (finalized) {
armTimer(state);
}
emitMissingQueuedTerminal();
} finally {
clearManualCronJobActive(state, jobId, prepared.activeJobMarker);
}
@@ -1258,11 +1332,31 @@ export async function enqueueRun(state: CronServiceState, id: string, mode?: "du
}
const runId = `manual:${id}:${state.deps.nowMs()}:${nextManualRunId++}`;
const terminalTracker: ManualRunTerminalTracker = { emitted: false };
void enqueueCommandInLane(
CommandLane.Cron,
async () => {
const result = await run(state, id, mode, { runId });
const result = await run(state, id, mode, { runId, terminalTracker });
if (result.ok && "ran" in result && !result.ran) {
if (result.reason !== "invalid-spec") {
const finishedAt = state.deps.nowMs();
const job = state.store?.jobs.find((entry) => entry.id === id);
emitManualRunFinished(
state,
{
jobId: id,
action: "finished",
job,
status: "skipped",
error: `queued manual run skipped before execution: ${result.reason}`,
runId,
runAtMs: finishedAt,
durationMs: 0,
nextRunAtMs: job?.state.nextRunAtMs,
},
terminalTracker,
);
}
state.deps.log.info(
{ jobId: id, runId, reason: result.reason },
"cron: queued manual run skipped before execution",
@@ -1280,6 +1374,30 @@ export async function enqueueRun(state: CronServiceState, id: string, mode?: "du
},
},
).catch((err: unknown) => {
if (terminalTracker.emitted) {
state.deps.log.error(
{ jobId: id, runId, err: String(err) },
"cron: queued manual run failed after emitting its terminal event",
);
return;
}
const finishedAt = state.deps.nowMs();
const job = state.store?.jobs.find((entry) => entry.id === id);
emitManualRunFinished(
state,
{
jobId: id,
action: "finished",
job,
status: "error",
error: normalizeCronRunErrorText(err),
runId,
runAtMs: finishedAt,
durationMs: 0,
nextRunAtMs: job?.state.nextRunAtMs,
},
terminalTracker,
);
state.deps.log.error(
{ jobId: id, runId, err: String(err) },
"cron: queued manual run background execution failed",
+1
View File
@@ -34,6 +34,7 @@ describe("cron stagger helpers", () => {
expect(normalizeCronStaggerMs("abc")).toBeUndefined();
expect(normalizeCronStaggerMs("1e3")).toBeUndefined();
expect(normalizeCronStaggerMs("0x10")).toBeUndefined();
expect(normalizeCronStaggerMs(Number.MAX_SAFE_INTEGER + 1)).toBeUndefined();
});
it("resolves effective stagger for cron schedules", () => {
+2 -1
View File
@@ -44,7 +44,8 @@ export function normalizeCronStaggerMs(raw: unknown): number | undefined {
if (!Number.isFinite(numeric)) {
return undefined;
}
return Math.max(0, Math.floor(numeric));
const normalized = Math.max(0, Math.floor(numeric));
return Number.isSafeInteger(normalized) ? normalized : undefined;
}
/** Returns the default anti-thundering-herd stagger for top-of-hour recurring schedules. */
+14
View File
@@ -283,6 +283,20 @@ function rowToCronJob(row: CronJobRow): CronJob | null {
};
}
/** Projects a live job through the same normalization/codecs used by SQLite persistence. */
export function projectCronJobThroughStorageCodec(job: CronJob): CronJob {
const normalized = normalizeCronJobForSqlite(job);
if (!normalized) {
throw new Error(`cannot project invalid cron job ${job.id}`);
}
const row = bindCronJobRow("config-revision", normalized, 0) as CronJobRow;
const projected = rowToCronJob(row);
if (!projected) {
throw new Error(`cannot project cron job ${job.id} through storage codecs`);
}
return projected;
}
/** Loads cron rows in config order with deterministic fallbacks for old rows. */
export function loadCronRows(db: DatabaseSync, storeKey: string): CronJobRow[] {
return executeSqliteQuerySync(
+40 -1
View File
@@ -15,6 +15,7 @@ import {
validateWakeParams,
} from "../../../packages/gateway-protocol/src/index.js";
import type { OpenClawConfig } from "../../config/types.openclaw.js";
import { resolveCronJobConfigRevision } from "../../cron/config-revision.js";
import {
assertValidCronAnnounceDelivery,
assertValidCronCreateDelivery,
@@ -73,9 +74,19 @@ type CronListCallerScopeContext = {
};
};
class CronJobConfigRevisionConflictError extends Error {
constructor(
readonly expectedConfigRevision: string,
readonly actualConfigRevision: string,
) {
super("cron job definition no longer matches the loaded version");
}
}
function cronJobReadView(job: CronJob) {
return {
...job,
configRevision: resolveCronJobConfigRevision(job),
nextRunAtMs: job.state.nextRunAtMs,
lastRunAtMs: job.state.lastRunAtMs,
lastRunStatus: job.state.lastRunStatus ?? job.state.lastStatus,
@@ -620,6 +631,7 @@ export const cronHandlers: GatewayRequestHandlers = {
id?: string;
jobId?: string;
patch: Record<string, unknown>;
expectedConfigRevision?: string;
};
const callerScope = readCronCallerScope(client);
const jobId = p.id ?? p.jobId;
@@ -694,6 +706,15 @@ export const cronHandlers: GatewayRequestHandlers = {
) {
throw new Error(`unknown cron job id: ${jobId}`);
}
if (p.expectedConfigRevision !== undefined) {
const actualConfigRevision = resolveCronJobConfigRevision(lockedJob);
if (actualConfigRevision !== p.expectedConfigRevision) {
throw new CronJobConfigRevisionConflictError(
p.expectedConfigRevision,
actualConfigRevision,
);
}
}
await assertValidCronUpdatePatch({
cfg,
defaultAgentId: context.cron.getDefaultAgentId(),
@@ -702,6 +723,24 @@ export const cronHandlers: GatewayRequestHandlers = {
});
});
} catch (err) {
if (err instanceof CronJobConfigRevisionConflictError) {
respond(
false,
undefined,
errorShape(
ErrorCodes.INVALID_REQUEST,
"cron job definition no longer matches the loaded version; review the latest version before retrying",
{
details: {
code: "CRON_JOB_CHANGED",
expectedConfigRevision: err.expectedConfigRevision,
actualConfigRevision: err.actualConfigRevision,
},
},
),
);
return;
}
if (
!(err instanceof TypeError) &&
!(err instanceof RangeError) &&
@@ -720,7 +759,7 @@ export const cronHandlers: GatewayRequestHandlers = {
return;
}
context.logGateway.info("cron: job updated", { jobId });
respond(true, job, undefined);
respond(true, cronJobReadView(job), undefined);
},
"cron.remove": async ({ params, respond, context, client }) => {
if (!validateCronRemoveParams(params)) {
@@ -357,6 +357,14 @@ function expectCronSuccess(respond: ReturnType<typeof vi.fn>): void {
expect(respond).toHaveBeenCalledWith(true, expect.objectContaining({ id: "cron-1" }), undefined);
}
function expectCronReadSuccess(respond: ReturnType<typeof vi.fn>, job: CronJob): void {
expect(respond).toHaveBeenCalledWith(
true,
expect.objectContaining({ ...job, configRevision: expect.stringMatching(/^sha256:/) }),
undefined,
);
}
function requireRecord(value: unknown, label: string): Record<string, unknown> {
if (!value || typeof value !== "object" || Array.isArray(value)) {
throw new Error(`expected ${label} to be an object`);
@@ -501,7 +509,7 @@ describe("cron method validation", () => {
const { context, respond } = await invokeCronGet({ id: "cron-42" }, job);
expect(context.cron.readJob).toHaveBeenCalledWith("cron-42");
expect(respond).toHaveBeenCalledWith(true, job, undefined);
expectCronReadSuccess(respond, job);
});
it("allows caller-scoped cron.get for the same agent", async () => {
@@ -511,7 +519,7 @@ describe("cron method validation", () => {
client: callerClient("ops"),
});
expect(respond).toHaveBeenCalledWith(true, job, undefined);
expectCronReadSuccess(respond, job);
});
it("hides caller-scoped cron.get for a foreign agent", async () => {
+82 -5
View File
@@ -169,6 +169,12 @@ type DirectCronState = GatewayCronState & {
type CronBroadcast = (event: string, payload: unknown) => void;
type DirectCronResponse = {
ok: boolean;
payload?: unknown;
error?: { code?: string; message?: string; details?: unknown };
};
async function createDirectCronState(params?: {
broadcast?: CronBroadcast;
}): Promise<DirectCronState> {
@@ -243,12 +249,14 @@ async function directCronReq(
cronState: DirectCronState,
method: string,
params: Record<string, unknown>,
): Promise<{ ok: boolean; payload?: unknown; error?: { code?: string; message?: string } }> {
): Promise<DirectCronResponse> {
const { cronHandlers } = await import("./server-methods/cron.js");
let result:
| { ok: boolean; payload?: unknown; error?: { code?: string; message?: string } }
| undefined;
const respond = (ok: boolean, payload?: unknown, error?: { code?: string; message?: string }) => {
let result: DirectCronResponse | undefined;
const respond = (
ok: boolean,
payload?: unknown,
error?: { code?: string; message?: string; details?: unknown },
) => {
result = {
ok,
payload,
@@ -915,6 +923,75 @@ describe("gateway server cron", () => {
}
});
test("atomically rejects stale config revisions without conflicting on runtime state", async () => {
const { prevSkipCron } = await setupCronTestRun({
tempPrefix: "openclaw-gw-cron-update-revision-",
cronEnabled: false,
});
const cronState = await createDirectCronState();
try {
const added = await directCronReq(cronState, "cron.add", {
name: "revision protected",
enabled: true,
schedule: { kind: "every", everyMs: 60_000 },
sessionTarget: "main",
wakeMode: "next-heartbeat",
payload: { kind: "systemEvent", text: "original" },
});
expect(added.ok).toBe(true);
const addedJob = added.payload as { id: string };
const initial = await directCronReq(cronState, "cron.get", { id: addedJob.id });
const initialJob = initial.payload as {
id: string;
configRevision: string;
updatedAtMs: number;
};
const runtimeOnly = await directCronReq(cronState, "cron.update", {
id: initialJob.id,
patch: { state: { lastRunAtMs: 1_700_000_000_000 } },
});
expect(runtimeOnly.ok).toBe(true);
expect(runtimeOnly.payload).toMatchObject({
configRevision: initialJob.configRevision,
});
const first = await directCronReq(cronState, "cron.update", {
id: initialJob.id,
expectedConfigRevision: initialJob.configRevision,
patch: { description: "first writer" },
});
expect(first.ok).toBe(true);
const firstJob = first.payload as { configRevision: string; updatedAtMs: number };
expect(firstJob.configRevision).not.toBe(initialJob.configRevision);
expect(firstJob.updatedAtMs).toBeGreaterThan(initialJob.updatedAtMs);
const stale = await directCronReq(cronState, "cron.update", {
id: initialJob.id,
expectedConfigRevision: initialJob.configRevision,
patch: { description: "stale writer" },
});
expect(stale.ok).toBe(false);
expect(stale.error).toMatchObject({
code: "INVALID_REQUEST",
details: {
code: "CRON_JOB_CHANGED",
expectedConfigRevision: initialJob.configRevision,
actualConfigRevision: firstJob.configRevision,
},
});
const current = await directCronReq(cronState, "cron.get", { id: initialJob.id });
expect(current.payload).toMatchObject({
description: "first writer",
updatedAtMs: firstJob.updatedAtMs,
});
} finally {
await cleanupCronTestRun({ cronState, prevSkipCron });
}
});
test("accepts opaque custom session ids on add and update", async () => {
const { prevSkipCron } = await setupCronTestRun({
tempPrefix: "openclaw-gw-cron-opaque-session-target-",