diff --git a/extensions/qa-lab/src/scenario-lane.test.ts b/extensions/qa-lab/src/scenario-lane.test.ts index 2958f2cf4cee..ae894f8826f9 100644 --- a/extensions/qa-lab/src/scenario-lane.test.ts +++ b/extensions/qa-lab/src/scenario-lane.test.ts @@ -1,5 +1,6 @@ // Qa Lab tests cover canonical scenario lane matching behavior. import { describe, expect, it } from "vitest"; +import type { QaProviderMode } from "./model-selection.js"; import { readQaScenarioById, readQaScenarioPack } from "./scenario-catalog.js"; import { describeQaProviderLaneMismatches, @@ -32,6 +33,43 @@ describe("QA scenario lane matching", () => { ).toBe(false); }); + it.each([ + { + id: "cron-empty-response-after-write-recovery", + allowedProviderMode: "mock-openai" as const, + rejectedProviderMode: "live-frontier" as const, + }, + { + id: "cron-explicit-authority-execution", + allowedProviderMode: "live-frontier" as const, + rejectedProviderMode: "mock-openai" as const, + }, + ])( + "keeps $id on its required $allowedProviderMode provider lane", + ({ id, allowedProviderMode, rejectedProviderMode }) => { + const scenario = readQaScenarioById(id); + const modelForMode = (mode: QaProviderMode) => + mode === "mock-openai" ? "mock-openai/gpt-5.6-luna" : "openai/gpt-5.6-luna"; + + expect( + scenarioMatchesQaProviderLane({ + scenario, + providerMode: allowedProviderMode, + primaryModel: modelForMode(allowedProviderMode), + }), + ).toBe(true); + const rejected = { + scenario, + providerMode: rejectedProviderMode, + primaryModel: modelForMode(rejectedProviderMode), + }; + expect(scenarioMatchesQaProviderLane(rejected)).toBe(false); + expect(describeQaProviderLaneMismatches(rejected)).toContain( + `providerMode=${allowedProviderMode}`, + ); + }, + ); + it("reports every declared mismatch in one decision", () => { const scenario = makeQaSuiteTestScenario("strict-live-lane", { channel: "matrix", diff --git a/extensions/workboard/src/dispatcher.races.test.ts b/extensions/workboard/src/dispatcher.races.test.ts index b57892582d6d..9c6a8ce8050e 100644 --- a/extensions/workboard/src/dispatcher.races.test.ts +++ b/extensions/workboard/src/dispatcher.races.test.ts @@ -73,4 +73,56 @@ describe("Workboard dispatcher lifecycle races", () => { expect(current?.status).toBe(transition.status); } }); + + it("does not spend worker attempts on cards that change before they can be claimed", async () => { + const store = new WorkboardStore(createMemoryStore()); + const first = await store.create({ + title: "First stale dispatch", + status: "ready", + priority: "urgent", + agentId: "first-worker", + workspaceAccess: { unrestricted: true }, + }); + const second = await store.create({ + title: "Second stale dispatch", + status: "ready", + priority: "high", + agentId: "second-worker", + workspaceAccess: { unrestricted: true }, + }); + const healthy = await store.create({ + title: "Healthy dispatch", + status: "ready", + priority: "normal", + agentId: "healthy-worker", + workspaceAccess: { unrestricted: true }, + }); + const originalClaim = store.claim.bind(store); + const staleCardIds = new Set([first.id, second.id]); + vi.spyOn(store, "claim").mockImplementation(async (id, input, options) => { + if (staleCardIds.delete(id)) { + await store.archive(id, true); + } + return await originalClaim(id, input, options); + }); + const run = vi.fn().mockResolvedValue({ runId: "healthy-run" }); + + const result = await dispatchAndStartWorkboardCards({ + store, + subagent: { run }, + options: { maxStarts: 1, workspaceAccess: { unrestricted: true } }, + }); + + expect(result.startFailures.map((failure) => failure.cardId)).toEqual([first.id, second.id]); + expect(result.started).toEqual([ + expect.objectContaining({ cardId: healthy.id, runId: "healthy-run" }), + ]); + expect(run).toHaveBeenCalledOnce(); + await expect(store.get(healthy.id)).resolves.toMatchObject({ status: "running" }); + for (const cardId of [first.id, second.id]) { + const archived = await store.get(cardId); + expect(archived?.metadata?.archivedAt).toBeGreaterThan(0); + expect(archived?.metadata?.claim).toBeUndefined(); + } + }); }); diff --git a/extensions/workboard/src/dispatcher.ts b/extensions/workboard/src/dispatcher.ts index dd9ee822d560..15c0fa08f082 100644 --- a/extensions/workboard/src/dispatcher.ts +++ b/extensions/workboard/src/dispatcher.ts @@ -439,8 +439,6 @@ async function runWorkboardDispatch( continue; } } - // Invalid workspace preflights must not spend the provider outage budget. - attemptedStarts += 1; try { const claimed = await params.store.claim( card.id, @@ -457,6 +455,9 @@ async function runWorkboardDispatch( }, ); claimValue = claimed.token; + // Racing card changes never reached a worker and must not consume the + // provider-outage budget or starve a later healthy candidate. + attemptedStarts += 1; const context = await params.store.buildWorkerContext(card.id); const materialized = await materializeWorkspace({ card: claimed.card, diff --git a/extensions/workboard/src/store-notifications.ts b/extensions/workboard/src/store-notifications.ts index 1c2a0780dcfb..1c98bcb967f4 100644 --- a/extensions/workboard/src/store-notifications.ts +++ b/extensions/workboard/src/store-notifications.ts @@ -117,20 +117,19 @@ export class WorkboardNotificationStore extends WorkboardWorkflowStore { if (subscription?.eventKinds?.length && !subscription.eventKinds.includes(event.kind)) { continue; } - const eventSequence = notificationSequence(event); - if (subscription?.lastEventSequence && eventSequence !== undefined) { - if ( - eventSequence < subscription.lastEventSequence || - (eventSequence === subscription.lastEventSequence && - event.id <= (subscription.lastEventId ?? "")) - ) { - continue; - } - } else if ( - subscription?.lastEventAt && - (event.createdAt < subscription.lastEventAt || - (event.createdAt === subscription.lastEventAt && - event.id <= (subscription.lastEventId ?? ""))) + // Cursor advancement must use the same mixed-sequence ordering as + // event delivery or valid same-millisecond notifications disappear. + if ( + subscription?.lastEventAt !== undefined && + compareNotifications(event, { + id: subscription.lastEventId ?? "", + kind: event.kind, + createdAt: subscription.lastEventAt, + ...(subscription.lastEventSequence !== undefined + ? { sequence: subscription.lastEventSequence } + : {}), + message: "", + }) <= 0 ) { continue; } diff --git a/extensions/workboard/src/store-workflow.ts b/extensions/workboard/src/store-workflow.ts index ac08bb2c3b69..d96a42d76d0d 100644 --- a/extensions/workboard/src/store-workflow.ts +++ b/extensions/workboard/src/store-workflow.ts @@ -24,6 +24,7 @@ import { import { addWorkboardDurationMs, DEFAULT_CLAIM_TTL_MS, + isWorkboardClaimReclaimable, MAX_CARD_ARTIFACTS, MAX_CARD_COMMENTS, MAX_CARD_NOTIFICATIONS, @@ -114,7 +115,11 @@ export class WorkboardWorkflowStore extends WorkboardPromoteStore { } const existingClaim = guarded.metadata?.claim; const activeClaim = - existingClaim && isFutureDateTimestampMs(existingClaim.expiresAt, { nowMs: now }) + existingClaim && + (isFutureDateTimestampMs(existingClaim.expiresAt, { nowMs: now }) || + // Direct claims must honor the same running-worker heartbeat grace + // as dispatcher recovery; otherwise they silently steal live tokens. + (guarded.status === "running" && !isWorkboardClaimReclaimable(existingClaim, now))) ? existingClaim : undefined; if (cardParentIds(guarded).length > 0 && guarded.status !== "ready" && !activeClaim) { diff --git a/extensions/workboard/src/store.test.ts b/extensions/workboard/src/store.test.ts index 8b59a6345239..b16b8694042d 100644 --- a/extensions/workboard/src/store.test.ts +++ b/extensions/workboard/src/store.test.ts @@ -1545,6 +1545,44 @@ describe("WorkboardStore", () => { ); }); + it("protects a running worker's expired claim throughout its heartbeat grace period", async () => { + vi.useFakeTimers(); + try { + vi.setSystemTime(1_000); + const store = new WorkboardStore(createMemoryStore()); + const card = await store.create({ title: "Grace-protected worker", status: "ready" }); + const claimed = await store.claim(card.id, { ownerId: "original", ttlSeconds: 1 }); + const expiresAt = claimed.card.metadata?.claim?.expiresAt; + if (expiresAt === undefined) { + throw new Error("expected a timed worker claim"); + } + + vi.setSystemTime(expiresAt + 1); + await expect(store.claim(card.id, { ownerId: "replacement" })).rejects.toThrow( + "card already claimed by original.", + ); + const renewed = await store.heartbeat(card.id, { + ownerId: "original", + token: claimed.token, + }); + const renewedExpiresAt = renewed.metadata?.claim?.expiresAt; + if (renewedExpiresAt === undefined) { + throw new Error("expected the worker heartbeat to renew its claim"); + } + + vi.setSystemTime(renewedExpiresAt + 5 * 60_000); + await expect(store.claim(card.id, { ownerId: "replacement" })).rejects.toThrow( + "card already claimed by original.", + ); + + vi.setSystemTime(renewedExpiresAt + 5 * 60_000 + 1); + const replacement = await store.claim(card.id, { ownerId: "replacement" }); + expect(replacement.card.metadata?.claim?.ownerId).toBe("replacement"); + } finally { + vi.useRealTimers(); + } + }); + it("preserves scheduled and retry-budget errors when a claim is active", async () => { vi.useFakeTimers(); try { @@ -3083,6 +3121,62 @@ describe("WorkboardStore", () => { expect(second.events).toEqual([expect.objectContaining({ id: "a-event" })]); }); + it("does not skip unsequenced notifications after a sequenced same-millisecond event", async () => { + const store = new WorkboardStore(createMemoryStore(), { + subscriptions: createMemoryStore(), + }); + await store.create({ + title: "Sequenced notification", + boardId: "ops", + metadata: { + notifications: [ + { + id: "z-event", + kind: "completed", + createdAt: 1234, + sequence: 1234000, + message: "First", + }, + ], + }, + }); + await store.create({ + title: "Unsequenced notification", + boardId: "ops", + metadata: { + notifications: [ + { + id: "a-event", + kind: "completed", + createdAt: 1234, + message: "Second", + }, + ], + }, + }); + const subscription = await store.subscribeNotifications({ + boardId: "ops", + target: "session:operator", + eventKinds: ["completed"], + }); + + const first = await store.advanceNotificationEvents({ + subscriptionId: subscription.id, + limit: 1, + }); + expect(first.events).toEqual([expect.objectContaining({ id: "z-event" })]); + + const second = await store.advanceNotificationEvents({ + subscriptionId: subscription.id, + limit: 1, + }); + expect(second.events).toEqual([expect.objectContaining({ id: "a-event" })]); + await expect(store.notificationEvents({ subscriptionId: subscription.id })).resolves.toEqual({ + subscription: expect.objectContaining({ id: subscription.id }), + events: [], + }); + }); + it("drains large same-millisecond notification batches without replaying delivered ids", async () => { const store = new WorkboardStore(createMemoryStore(), { subscriptions: createMemoryStore(), diff --git a/qa/scenarios/scheduling/cron-empty-response-after-write-recovery.yaml b/qa/scenarios/scheduling/cron-empty-response-after-write-recovery.yaml index 0be844bedbac..eee72b257ca8 100644 --- a/qa/scenarios/scheduling/cron-empty-response-after-write-recovery.yaml +++ b/qa/scenarios/scheduling/cron-empty-response-after-write-recovery.yaml @@ -22,6 +22,7 @@ scenario: channel: qa-channel summary: Force one resolved announce cron through settled-tool finalization. config: + requiredProviderMode: mock-openai channelId: qa-room promptSnippet: Empty response after write recovery QA check retryNeedle: The previous assistant turn completed its tool calls but did not produce a user-visible answer. diff --git a/qa/scenarios/scheduling/cron-explicit-authority-execution.yaml b/qa/scenarios/scheduling/cron-explicit-authority-execution.yaml index 6ea96d7cd2fd..6a9a5355aa03 100644 --- a/qa/scenarios/scheduling/cron-explicit-authority-execution.yaml +++ b/qa/scenarios/scheduling/cron-explicit-authority-execution.yaml @@ -3,7 +3,6 @@ title: Cron explicit authority execution scenario: id: cron-explicit-authority-execution surface: cron - runtimeParityTier: live-only coverage: primary: - automation.isolated-cron-execution @@ -46,6 +45,7 @@ scenario: timeoutMs: 420000 summary: Create one group-owned job through the operator API, add a sender wildcard restriction, then prove its scheduled write authority survives requester-policy resolution. config: + requiredProviderMode: live-frontier artifactFile: cron-explicit-authority-proof.txt artifactText: CRON_EXPLICIT_AUTHORITY_OK conversationId: qa-cron-authority-execution diff --git a/src/cron/service.runs-one-shot-main-job-disables-it.test.ts b/src/cron/service.runs-one-shot-main-job-disables-it.test.ts index 7b58da8e688f..53944508addf 100644 --- a/src/cron/service.runs-one-shot-main-job-disables-it.test.ts +++ b/src/cron/service.runs-one-shot-main-job-disables-it.test.ts @@ -463,6 +463,38 @@ describe("CronService", () => { await stopCronAndCleanup(cron, store); }); + it("removes a queued main-session event when an immediate heartbeat fails", async () => { + const runHeartbeatOnce = vi.fn(async () => { + throw new Error("heartbeat failed"); + }); + const { store, cron, enqueueSystemEvent, requestHeartbeat } = await createCronHarness({ + runHeartbeatOnce, + useRemovableSystemEventQueue: true, + withEvents: false, + }); + + try { + const job = await addWakeModeNowMainSystemEventJob(cron, { + name: "failed immediate heartbeat", + }); + + await cron.run(job.id, "force"); + + expect(runHeartbeatOnce).toHaveBeenCalledOnce(); + expect(requestHeartbeat).not.toHaveBeenCalled(); + const sessionKeys = getPostedSystemEventSessionKeys(enqueueSystemEvent); + expect(sessionKeys).toHaveLength(1); + expectNoQueuedEvents(sessionKeys); + const updated = (await cron.list({ includeDisabled: true })).find( + (candidate) => candidate.id === job.id, + ); + expect(updated?.state.lastRunStatus).toBe("error"); + expect(updated?.state.lastError).toContain("heartbeat failed"); + } finally { + await stopCronAndCleanup(cron, store); + } + }); + it("rejects sessionTarget main for non-default agents at creation time", async () => { const runHeartbeatOnce = vi.fn(async () => ({ status: "ran" as const, durationMs: 1 })); diff --git a/src/cron/service/timer-execution.ts b/src/cron/service/timer-execution.ts index 180dc807b9cb..bb044ad7f0bf 100644 --- a/src/cron/service/timer-execution.ts +++ b/src/cron/service/timer-execution.ts @@ -272,16 +272,23 @@ async function executeMainSessionCronJob( removeQueuedSystemEventHandle(state, job, queuedSystemEvent); return { status: "error", error: timeoutErrorMessage() }; } - heartbeatResult = await state.deps.runHeartbeatOnce({ - source: "cron", - intent: "immediate", - reason, - agentId: job.agentId, - sessionKey: cronRunSessionKey, - owningCronJobMarker: activeJobMarker, - owningCronLaneTaskMarker, - heartbeat: { target: "last" }, - }); + try { + heartbeatResult = await state.deps.runHeartbeatOnce({ + source: "cron", + intent: "immediate", + reason, + agentId: job.agentId, + sessionKey: cronRunSessionKey, + owningCronJobMarker: activeJobMarker, + owningCronLaneTaskMarker, + heartbeat: { target: "last" }, + }); + } catch (error) { + // A failed immediate heartbeat must not leave its failed run's + // reminder queued for an unrelated future heartbeat. + removeQueuedSystemEventHandle(state, job, queuedSystemEvent); + throw error; + } if (abortSignal?.aborted) { removeQueuedSystemEventHandle(state, job, queuedSystemEvent); return { status: "error", error: timeoutErrorMessage() }; diff --git a/src/tasks/cron-history-retention.ts b/src/tasks/cron-history-retention.ts index 77f047596686..0696f2f26ce1 100644 --- a/src/tasks/cron-history-retention.ts +++ b/src/tasks/cron-history-retention.ts @@ -36,6 +36,7 @@ export function collectCronHistoryOverflowTaskIds(tasks: readonly TaskRecord[]): rows.sort((left, right) => { return ( resolveCronTaskRecordTimestamp(right) - resolveCronTaskRecordTimestamp(left) || + right.createdAt - left.createdAt || right.taskId.localeCompare(left.taskId) ); }); diff --git a/src/tasks/task-registry.maintenance.issue-60299.test.ts b/src/tasks/task-registry.maintenance.issue-60299.test.ts index 69119209b724..50dd21fcc42d 100644 --- a/src/tasks/task-registry.maintenance.issue-60299.test.ts +++ b/src/tasks/task-registry.maintenance.issue-60299.test.ts @@ -923,6 +923,31 @@ describe("task-registry maintenance issue #60299", () => { expect(currentTasks.has(lostTask.taskId)).toBe(true); }); + it("retains the newest-created cron runs when terminal timestamps are identical", async () => { + const now = Date.now(); + const tasks = Array.from({ length: CRON_HISTORY_KEEP_PER_JOB + 1 }, (_, index) => + makeStaleTask({ + taskId: `cron-same-ms-${String(CRON_HISTORY_KEEP_PER_JOB - index).padStart(4, "0")}`, + runtime: "cron", + sourceId: "cron-same-ms-job", + status: "succeeded", + createdAt: now + index, + startedAt: now + index, + endedAt: now + CRON_HISTORY_KEEP_PER_JOB + 1, + lastEventAt: now + CRON_HISTORY_KEEP_PER_JOB + 1, + cleanupAfter: 0, + }), + ); + const { currentTasks } = createTaskRegistryMaintenanceHarness({ tasks }); + + const result = await runTaskRegistryMaintenance(); + + expect(result.pruned).toBe(1); + expect(currentTasks.size).toBe(CRON_HISTORY_KEEP_PER_JOB); + expect(currentTasks.has("cron-same-ms-2000")).toBe(false); + expect(currentTasks.has("cron-same-ms-0000")).toBe(true); + }); + it("scopes same-id cron history retention to each store", async () => { const now = Date.now(); const storeATasks = Array.from({ length: CRON_HISTORY_KEEP_PER_JOB }, (_, index) => diff --git a/ui/src/pages/workboard/workboard-page.test.ts b/ui/src/pages/workboard/workboard-page.test.ts index 83188ad235af..50717268b3dd 100644 --- a/ui/src/pages/workboard/workboard-page.test.ts +++ b/ui/src/pages/workboard/workboard-page.test.ts @@ -3,6 +3,7 @@ import type { ApplicationContext, ApplicationGatewaySnapshot } from "../../app/c import { createWorkboardCapability } from "../../lib/workboard/capability.ts"; import type { WorkboardCapability } from "../../lib/workboard/capability.ts"; import * as workboardLib from "../../lib/workboard/index.ts"; +import type { WorkboardRouteData } from "./route.ts"; const configureLiveRefresh = vi.fn((): boolean => false); const handleChanged = vi.fn(); @@ -15,8 +16,10 @@ await import("./workboard-page.ts"); type WorkboardPageTestElement = HTMLElement & { context: ApplicationContext; + routeData?: WorkboardRouteData; updateComplete: Promise; syncWorkboardAgentScope: () => void; + syncWorkboardBoardFilter: () => void; }; function contextWithWorkboard(workboard: WorkboardCapability): ApplicationContext { @@ -242,4 +245,64 @@ describe("WorkboardPage lifecycle", () => { expect(workboard.state.draftOpen).toBe(true); expect(workboard.state.editingCardId).toBe("writer-card"); }); + + it.each([ + { boardFilter: "product", remainsVisible: false }, + { boardFilter: "__all__", remainsVisible: true }, + ])( + "reconciles existing card overlays when the board route changes to $boardFilter", + async ({ boardFilter, remainsVisible }) => { + const workboard = createWorkboardCapability(); + const page = document.createElement("openclaw-workboard-page") as WorkboardPageTestElement; + page.context = contextWithWorkboard(workboard); + document.body.append(page); + await page.updateComplete; + workboard.state.cards = [ + { + id: "ops-card", + title: "Operations task", + status: "todo", + priority: "normal", + labels: [], + position: 1000, + createdAt: 1, + updatedAt: 1, + metadata: { automation: { boardId: "ops" } }, + }, + ]; + workboard.state.boardFilter = "ops"; + workboard.state.detailCardId = "ops-card"; + workboard.state.detailCommentBody = "draft comment"; + workboard.state.draftOpen = true; + workboard.state.editingCardId = "ops-card"; + page.routeData = { boardFilter, search: "" }; + + page.syncWorkboardBoardFilter(); + + expect(workboard.state.boardFilter).toBe(boardFilter); + expect(workboard.state.detailCardId).toBe(remainsVisible ? "ops-card" : null); + expect(workboard.state.detailCommentBody).toBe(remainsVisible ? "draft comment" : ""); + expect(workboard.state.draftOpen).toBe(remainsVisible); + expect(workboard.state.editingCardId).toBe(remainsVisible ? "ops-card" : null); + }, + ); + + it("preserves a new-card draft when the board route changes", async () => { + const workboard = createWorkboardCapability(); + const page = document.createElement("openclaw-workboard-page") as WorkboardPageTestElement; + page.context = contextWithWorkboard(workboard); + document.body.append(page); + await page.updateComplete; + workboard.state.boardFilter = "ops"; + workboard.state.draftOpen = true; + workboard.state.draftTitle = "New operations task"; + page.routeData = { boardFilter: "product", search: "" }; + + page.syncWorkboardBoardFilter(); + + expect(workboard.state.boardFilter).toBe("product"); + expect(workboard.state.draftOpen).toBe(true); + expect(workboard.state.draftTitle).toBe("New operations task"); + expect(workboard.state.editingCardId).toBeNull(); + }); }); diff --git a/ui/src/pages/workboard/workboard-page.ts b/ui/src/pages/workboard/workboard-page.ts index 36e9a5fa6a3c..c9531241d964 100644 --- a/ui/src/pages/workboard/workboard-page.ts +++ b/ui/src/pages/workboard/workboard-page.ts @@ -31,7 +31,7 @@ import { import { OpenClawLightDomElement } from "../../lit/openclaw-element.ts"; import { SubscriptionsController } from "../../lit/subscriptions-controller.ts"; import { matchesAgentScope } from "./agent-filter.ts"; -import { WORKBOARD_ALL_BOARDS_FILTER } from "./board-filter.ts"; +import { matchesBoardFilter, WORKBOARD_ALL_BOARDS_FILTER } from "./board-filter.ts"; import type { WorkboardRouteData } from "./route.ts"; import { renderWorkboard } from "./view.ts"; @@ -253,7 +253,19 @@ class WorkboardPage extends OpenClawLightDomElement { if (!context || !boardFilter || context.workboard.state.boardFilter === boardFilter) { return; } - context.workboard.state.boardFilter = boardFilter; + const state = context.workboard.state; + const remainsVisible = (cardId: string) => { + const card = state.cards.find((entry) => entry.id === cardId); + return Boolean(card && matchesBoardFilter(card, boardFilter)); + }; + if (state.detailCardId && !remainsVisible(state.detailCardId)) { + state.detailCardId = null; + state.detailCommentBody = ""; + } + if (state.editingCardId && !remainsVisible(state.editingCardId)) { + resetDraftState(state); + } + state.boardFilter = boardFilter; context.workboard.notify(); }