From 8fcbf6d592526b461d840e6bfcb3c275fffb9501 Mon Sep 17 00:00:00 2001 From: wen2zhou Date: Thu, 23 Jul 2026 08:25:56 +0000 Subject: [PATCH 1/4] fix(compute): show queued jobs on notebook bar MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When jobs exceed concurrency limits and enter queued status, they were invisible on the notebook bar badge. Root cause had two layers: 1. Main process: submitJob only broadcast job-updated for submitted jobs (via dispatchJob), while queued jobs were written to DB but never broadcast. The renderer store relies purely on broadcasts after initial hydrate, so queued jobs never reached the UI. 2. Renderer: RemoteJobBadge only counted running+submitted, ignoring queued. Fixes: - Broadcast job-updated immediately after creating any job (queued or submitted) so the renderer store sees it instantly - Badge now displays 'N running · M queued' when queued jobs exist, stays amber (not gray idle), and tooltip shows queued rows with 'queued' label - Subscribe to jobsById map (not stable fn ref) so badge re-renders on any job status change, not just the 1s elapsed-time tick Tests: 16 badge render tests + 1 integration test for queued broadcast. All compute-service/concurrency suites pass (138 tests, no regressions). Co-Authored-By: Claude Opus 4.8 (1M context) --- src/main/compute/compute-service.ts | 12 ++ .../compute/concurrency-integration.test.ts | 33 +++++ .../components/RemoteJobBadge.render.test.tsx | 120 +++++++++++++++++- .../src/components/RemoteJobBadge.tsx | 52 ++++++-- .../src/components/remote-job-badge-utils.ts | 8 ++ 5 files changed, 210 insertions(+), 15 deletions(-) diff --git a/src/main/compute/compute-service.ts b/src/main/compute/compute-service.ts index 1e4e5746..b7931bd6 100644 --- a/src/main/compute/compute-service.ts +++ b/src/main/compute/compute-service.ts @@ -1287,6 +1287,18 @@ export class ComputeService { await createRow('submitted') } + // ── BROADCAST THE NEW ROW ────────────────────────────────────────────────────── + // The renderer job store hydrates once then stays live purely off job-updated broadcasts, so + // the badge/feed only ever sees a job that was broadcast at least once. The dispatcher broadcasts + // submitted→running→terminal transitions, but a QUEUED job is never dispatched (it waits for a + // slot) and would otherwise never reach the renderer. Broadcast the freshly created row here so + // queued jobs appear on the notebook bar immediately; this also closes the brief invisible window + // for submitted jobs before the async dispatcher emits its first transition. + if (this.onJobUpdated) { + const createdJob = await this.jobRepository.get(jobId) + if (createdJob) this.onJobUpdated(createdJob) + } + // ── BACKGROUND DISPATCH (non-blocking, only for submitted jobs) ──────────────── // Fire-and-forget. Errors are persisted to the job row by the dispatcher. // Queued jobs are NOT dispatched here — they wait for ConcurrencyManager.tryDispatchNext(). diff --git a/src/main/compute/concurrency-integration.test.ts b/src/main/compute/concurrency-integration.test.ts index 42babfd3..f75a422d 100644 --- a/src/main/compute/concurrency-integration.test.ts +++ b/src/main/compute/concurrency-integration.test.ts @@ -157,6 +157,39 @@ describe('ConcurrencyManager integration with ComputeService', () => { expect(job2?.status).toBe('queued') }) + it('broadcasts a job-updated event when a job is created in queued status', async () => { + const providerId = computeProviderId('test-host') + await service.setSessionConcurrencyLimit('session-1', 1) + + // First job is submitted (dispatch broadcasts happen via the dispatcher, mocked here). + await service.submitJob( + providerId, + 'job 1', + 'echo one', + {}, + { sessionId: 'session-1', projectId: 'project-1' } + ) + + onJobUpdatedSpy.mockClear() + + // Second job queues — the renderer store only learns about jobs via job-updated broadcasts, + // so a queued job MUST broadcast on creation or it never appears on the notebook bar. + const result2 = await service.submitJob( + providerId, + 'job 2', + 'echo two', + {}, + { sessionId: 'session-1', projectId: 'project-1' } + ) + expect(result2.status).toBe('queued') + + const broadcastForQueued = onJobUpdatedSpy.mock.calls.find( + ([job]) => job?.job_id === result2.job_id + ) + expect(broadcastForQueued).toBeDefined() + expect(broadcastForQueued?.[0].status).toBe('queued') + }) + it('should submit job with status=queued when provider ceiling reached', async () => { const providerId = computeProviderId('test-host') diff --git a/src/renderer/src/components/RemoteJobBadge.render.test.tsx b/src/renderer/src/components/RemoteJobBadge.render.test.tsx index a594b0d4..ada47b07 100644 --- a/src/renderer/src/components/RemoteJobBadge.render.test.tsx +++ b/src/renderer/src/components/RemoteJobBadge.render.test.tsx @@ -5,9 +5,19 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' import type { JobSummary } from '../../../shared/compute' import { RemoteJobBadge } from './RemoteJobBadge' -import { formatDuration } from './remote-job-badge-utils' +import { formatDuration, jobRowDuration } from './remote-job-badge-utils' import { createInitialSessionJobState, useSessionJobStore } from '@/stores/session-job-store' +// Radix's tooltip open/close relies on pointer + rAF timing that jsdom does not drive, so its +// portaled content never mounts under synthetic events. Render the primitives inline (content +// always visible) so tests can assert RemoteJobBadge's own tooltip JSX rather than Radix internals. +vi.mock('@/components/ui/tooltip', () => ({ + TooltipProvider: ({ children }: { children: React.ReactNode }) => children, + Tooltip: ({ children }: { children: React.ReactNode }) => children, + TooltipTrigger: ({ children }: { children: React.ReactNode }) => children, + TooltipContent: ({ children }: { children: React.ReactNode }) => children +})) + let container: HTMLDivElement let root: Root @@ -59,6 +69,20 @@ describe('formatDuration', () => { }) }) +describe('jobRowDuration', () => { + it('labels a queued job "queued" instead of an elapsed time (it has not started)', () => { + const now = Date.now() + const queued = makeJob({ status: 'queued', started_at: undefined, created_at: now - 5_000 }) + expect(jobRowDuration(queued, now)).toBe('queued') + }) + + it('renders the elapsed duration for a running job', () => { + const now = Date.now() + const running = makeJob({ status: 'running', started_at: now - 90_000 }) + expect(jobRowDuration(running, now)).toBe('1m 30s') + }) +}) + describe('RemoteJobBadge — 0 running', () => { it('renders nothing when there are no running jobs', () => { act(() => { @@ -136,6 +160,100 @@ describe('RemoteJobBadge — N running', () => { }) }) +describe('RemoteJobBadge — queued jobs (concurrency limit)', () => { + it('counts queued jobs alongside running in an active amber badge', () => { + useSessionJobStore.getState().applyUpdate(makeJob({ job_id: 'r1', status: 'running' })) + useSessionJobStore.getState().applyUpdate(makeJob({ job_id: 'q1', status: 'queued' })) + useSessionJobStore.getState().applyUpdate(makeJob({ job_id: 'q2', status: 'queued' })) + + act(() => { + root.render() + }) + + expect(container.textContent).toContain('1 running') + expect(container.textContent).toContain('2 queued') + + // Badge stays amber/active (carries the --session-waiting style), not gray idle. + const btn = container.querySelector('button') + expect(btn?.getAttribute('style')).toContain('--session-waiting') + }) + + it('stays amber and shows "N queued" when every in-flight job is queued (no running)', () => { + useSessionJobStore.getState().applyUpdate(makeJob({ job_id: 'q1', status: 'queued' })) + useSessionJobStore.getState().applyUpdate(makeJob({ job_id: 'q2', status: 'queued' })) + + act(() => { + root.render() + }) + + // Active (amber), not the gray idle badge. + expect(container.textContent).toContain('2 queued') + expect(container.textContent).not.toContain('running') + const btn = container.querySelector('button') + expect(btn?.getAttribute('style')).toContain('--session-waiting') + + // The accessible label must not claim jobs are "running" when none are. + expect(btn?.getAttribute('aria-label')).not.toContain('running') + expect(btn?.getAttribute('aria-label')).toContain('queued') + }) + + it('drops the queued segment once a queued job is dispatched to running', () => { + useSessionJobStore.getState().applyUpdate(makeJob({ job_id: 'r1', status: 'running' })) + useSessionJobStore.getState().applyUpdate(makeJob({ job_id: 'q1', status: 'queued' })) + + act(() => { + root.render() + }) + expect(container.textContent).toContain('1 running') + expect(container.textContent).toContain('1 queued') + + // A slot frees up: the queued job is promoted to running. + act(() => { + useSessionJobStore.getState().applyUpdate(makeJob({ job_id: 'q1', status: 'running' })) + }) + + expect(container.textContent).toContain('2 running') + expect(container.textContent).not.toContain('queued') + }) + + it('falls back to the gray "N jobs" badge once every job is terminal', () => { + useSessionJobStore.getState().applyUpdate(makeJob({ job_id: 'r1', status: 'running' })) + useSessionJobStore.getState().applyUpdate(makeJob({ job_id: 'q1', status: 'queued' })) + + act(() => { + root.render() + }) + + // Both jobs finish. + act(() => { + useSessionJobStore.getState().applyUpdate(makeJob({ job_id: 'r1', status: 'success' })) + useSessionJobStore.getState().applyUpdate(makeJob({ job_id: 'q1', status: 'failed' })) + }) + + expect(container.textContent).toContain('2 jobs') + expect(container.textContent).not.toContain('running') + expect(container.textContent).not.toContain('queued') + const btn = container.querySelector('button') + expect(btn?.getAttribute('style')).not.toContain('--session-waiting') + }) + + it('lists queued jobs in the tooltip with a "queued" label instead of an elapsed time', () => { + useSessionJobStore.getState().applyUpdate(makeJob({ job_id: 'r1', status: 'running' })) + useSessionJobStore + .getState() + .applyUpdate(makeJob({ job_id: 'q1', status: 'queued', intent: 'Train model' })) + + act(() => { + root.render() + }) + + // Tooltip content renders inline (primitives mocked above). The queued job appears as a row + // with its intent and a literal "queued" label rather than an elapsed duration. + expect(container.textContent).toContain('Train model') + expect(container.textContent).toContain('queued') + }) +}) + describe('RemoteJobBadge — click interaction', () => { it('triggers onOpenJobList when clicked', () => { const handleOpen = vi.fn() diff --git a/src/renderer/src/components/RemoteJobBadge.tsx b/src/renderer/src/components/RemoteJobBadge.tsx index b18b82b5..da097d3a 100644 --- a/src/renderer/src/components/RemoteJobBadge.tsx +++ b/src/renderer/src/components/RemoteJobBadge.tsx @@ -1,10 +1,10 @@ -import { useEffect, useState } from 'react' +import { useEffect, useMemo, useState } from 'react' import { Zap } from 'lucide-react' import type { JobSummary } from '../../../shared/compute' import { Tooltip, TooltipContent, TooltipProvider, TooltipTrigger } from '@/components/ui/tooltip' import { useSessionJobStore } from '@/stores/session-job-store' -import { formatDuration, jobElapsedMs } from './remote-job-badge-utils' +import { formatDuration, jobElapsedMs, jobRowDuration } from './remote-job-badge-utils' // Badge props — sessionId is used to scope the running job list to the active session. type RemoteJobBadgeProps = { @@ -13,11 +13,19 @@ type RemoteJobBadgeProps = { } // Amber capsule badge placed on the notebook bar right side (design.md §4). -// Shows running count + elapsed time when jobs are running; shows gray "N jobs" when all finished. -// Hidden only when session has no jobs at all. -// Hover reveals a tooltip listing each running job's host + intent + duration. -export const RemoteJobBadge = ({ sessionId, onOpenJobList }: RemoteJobBadgeProps): React.JSX.Element | null => { - const allJobsForSession = useSessionJobStore((state) => state.allJobsForSession) +// While any job is in-flight it shows the active counts + elapsed time ("N running · M queued · +// elapsed"); queued jobs (waiting for a concurrency slot) keep the badge amber so they are never +// hidden. Shows a gray "N jobs" once every job is terminal, and is hidden only when the session has +// no jobs at all. Hover reveals a tooltip listing each in-flight job's host + intent + duration +// (queued rows show "queued" instead of an elapsed time). +export const RemoteJobBadge = ({ + sessionId, + onOpenJobList +}: RemoteJobBadgeProps): React.JSX.Element | null => { + // Subscribe to the raw job map (not the allJobsForSession selector fn, whose reference is stable) + // so the badge re-renders the instant a job is added or changes status — applyUpdate replaces the + // map on every broadcast. Without this the badge would only refresh on the 1s elapsed-time tick. + const jobsById = useSessionJobStore((state) => state.jobsById) const [now, setNow] = useState(() => Date.now()) // Tick every second to keep elapsed times fresh. @@ -26,10 +34,21 @@ export const RemoteJobBadge = ({ sessionId, onOpenJobList }: RemoteJobBadgeProps return () => clearInterval(id) }, []) - const allJobs = allJobsForSession(sessionId) + // Jobs for this session, newest first — recomputed whenever the job map changes. + const allJobs = useMemo( + () => + Array.from(jobsById.values()) + .filter((j) => j.session_id === sessionId) + .sort((a, b) => b.created_at - a.created_at), + [jobsById, sessionId] + ) - // Count active jobs (running + submitted) for display - const activeJobs = allJobs.filter((j) => j.status === 'running' || j.status === 'submitted') + // Running jobs are dispatched to the remote host (running or submitted there); + // queued jobs are waiting locally for a free concurrency slot. Both are in-flight and + // keep the badge amber. Terminal jobs (success/failed/timeout/error) only count toward "N jobs". + const runningJobs = allJobs.filter((j) => j.status === 'running' || j.status === 'submitted') + const queuedJobs = allJobs.filter((j) => j.status === 'queued') + const activeJobs = [...runningJobs, ...queuedJobs] // Hidden only when session has no jobs at all. if (allJobs.length === 0) return null @@ -37,7 +56,7 @@ export const RemoteJobBadge = ({ sessionId, onOpenJobList }: RemoteJobBadgeProps const isActive = activeJobs.length > 0 if (isActive) { - // Active state: amber badge with "N running · elapsed" + // Active state: amber badge with "N running [· N queued] · elapsed". const oldest = activeJobs.reduce((a, b) => { const aStart = a.started_at ?? a.created_at const bStart = b.started_at ?? b.created_at @@ -47,6 +66,11 @@ export const RemoteJobBadge = ({ sessionId, onOpenJobList }: RemoteJobBadgeProps const elapsedMs = jobElapsedMs(oldest, now) const elapsedStr = formatDuration(elapsedMs) + // Count segments — only non-zero groups appear, joined with " · ". + const segments: string[] = [] + if (runningJobs.length > 0) segments.push(`${runningJobs.length} running`) + if (queuedJobs.length > 0) segments.push(`${queuedJobs.length} queued`) + return ( @@ -66,11 +90,11 @@ export const RemoteJobBadge = ({ sessionId, onOpenJobList }: RemoteJobBadgeProps gap: '4px', cursor: onOpenJobList ? 'pointer' : 'default' }} - aria-label={`${activeJobs.length} running remote job${activeJobs.length !== 1 ? 's' : ''}`} + aria-label={`${segments.join(', ')} remote job${activeJobs.length !== 1 ? 's' : ''}`} > - {activeJobs.length} running · {elapsedStr} + {segments.join(' · ')} · {elapsedStr} @@ -85,7 +109,7 @@ export const RemoteJobBadge = ({ sessionId, onOpenJobList }: RemoteJobBadgeProps {job.display_name} {job.intent} - {formatDuration(jobElapsedMs(job, now))} + {jobRowDuration(job, now)} ))} diff --git a/src/renderer/src/components/remote-job-badge-utils.ts b/src/renderer/src/components/remote-job-badge-utils.ts index fdd6df15..58288f73 100644 --- a/src/renderer/src/components/remote-job-badge-utils.ts +++ b/src/renderer/src/components/remote-job-badge-utils.ts @@ -14,3 +14,11 @@ export const jobElapsedMs = (job: JobSummary, now: number): number => { const start = job.started_at ?? job.created_at return now - start } + +// Right-aligned label for a job row / tooltip entry. A queued job has not started running on the +// remote host, so an elapsed time would be misleading — it shows the literal status "queued" +// instead. Every other status shows its elapsed duration. +export const jobRowDuration = (job: JobSummary, now: number): string => { + if (job.status === 'queued') return 'queued' + return formatDuration(jobElapsedMs(job, now)) +} From ca7df03c4c6a4db9854742761e009d427b23a755 Mon Sep 17 00:00:00 2001 From: wen2zhou Date: Thu, 23 Jul 2026 05:42:49 +0000 Subject: [PATCH 2/4] fix(compute): drain queued jobs on startup so restart does not orphan queued tasks - Add drainOnStartup() to ConcurrencyManager (delegates to tryDispatchNext) - Add drainQueuedJobs() facade on ComputeService - Wire void computeService.drainQueuedJobs() in ipc.ts after jobPoller.start() - Tests: drainOnStartup unit test + integration test for restart scenario --- src/main/compute/compute-service.ts | 9 ++++ .../compute/concurrency-integration.test.ts | 44 +++++++++++++++++++ src/main/compute/concurrency-manager.test.ts | 36 +++++++++++++++ src/main/compute/concurrency-manager.ts | 7 +++ src/main/ipc.ts | 3 ++ 5 files changed, 99 insertions(+) diff --git a/src/main/compute/compute-service.ts b/src/main/compute/compute-service.ts index b7931bd6..679bda8c 100644 --- a/src/main/compute/compute-service.ts +++ b/src/main/compute/compute-service.ts @@ -1478,6 +1478,15 @@ export class ComputeService { return status } + // Drains queued jobs after a process restart. Should be called once during startup after the poller + // has been initialized: any jobs that were queued when the process last exited will be dispatched + // if capacity is now available (e.g. previously-active jobs completed while the app was down). + async drainQueuedJobs(): Promise { + if (this.concurrencyManager) { + await this.concurrencyManager.drainOnStartup() + } + } + // Internal callback wrapper: when a job transitions to a terminal state, notify ConcurrencyManager // to trigger auto-dispatch of queued jobs. This is called by the JobPoller via onJobUpdated. // Exposed as a method so the JobPoller (or IPC layer) can wire it in production. diff --git a/src/main/compute/concurrency-integration.test.ts b/src/main/compute/concurrency-integration.test.ts index f75a422d..da8d9cbb 100644 --- a/src/main/compute/concurrency-integration.test.ts +++ b/src/main/compute/concurrency-integration.test.ts @@ -413,6 +413,50 @@ describe('ConcurrencyManager integration with ComputeService', () => { expect(status.provider_ceilings[providerId]).toBe(10) // default ceiling }) + it('queued job is dispatched on drainOnStartup when no active jobs', async () => { + const providerId = computeProviderId('test-host') + + // Set provider ceiling to 1 so the second submission queues + await hostRepo.updateConcurrencyLimit(providerId, 1) + + // Submit first job (should be submitted — fills the single slot) + const result1 = await service.submitJob( + providerId, + 'job 1', + 'echo one', + {}, + { sessionId: 'session-1', projectId: 'project-1' } + ) + expect(result1.status).toBe('submitted') + + // Submit second job from a different session (should queue because provider ceiling is hit) + const result2 = await service.submitJob( + providerId, + 'job 2', + 'echo two', + {}, + { sessionId: 'session-2', projectId: 'project-1' } + ) + expect(result2.status).toBe('queued') + + // Simulate restart: mark job-1 as done externally (as if the app exited while it was running) + await jobRepo.update(result1.job_id, { + status: 'success', + finishedAt: new Date(), + exitCode: 0 + }) + + // No active jobs remain — call drainQueuedJobs to simulate startup drain + await service.drainQueuedJobs() + + // Wait for async dispatch to complete + await new Promise((resolve) => setTimeout(resolve, 200)) + + // job-2 should now be submitted + const job2Updated = await jobRepo.get(result2.job_id) + expect(job2Updated?.status).toBe('submitted') + }) + it('should not dispatch queued job if status is not terminal', async () => { const providerId = computeProviderId('test-host') diff --git a/src/main/compute/concurrency-manager.test.ts b/src/main/compute/concurrency-manager.test.ts index 5b56b041..20d14440 100644 --- a/src/main/compute/concurrency-manager.test.ts +++ b/src/main/compute/concurrency-manager.test.ts @@ -608,6 +608,42 @@ describe('ConcurrencyManager', () => { }) }) + describe('drainOnStartup', () => { + it('drainOnStartup dispatches eligible queued jobs', async () => { + const queuedJobs: ComputeJob[] = [ + { + job_id: 'job-1', + session_id: 'session-1', + provider_id: 'ssh:cluster-a', + created_at: 1000, + status: 'queued' + } as ComputeJob + ] + + vi.mocked(jobRepo.findQueuedJobs).mockResolvedValue(queuedJobs) + vi.mocked(jobRepo.countActiveBySession).mockResolvedValue(0) + vi.mocked(jobRepo.countActiveByProvider).mockResolvedValue(0) + vi.mocked(hostRepo.get).mockResolvedValue({ + concurrencyLimit: 10 + } as ComputeHost) + vi.mocked(jobRepo.update).mockResolvedValue({} as ComputeJob) + + await manager.drainOnStartup() + + expect(jobRepo.update).toHaveBeenCalledWith('job-1', { status: 'submitted' }) + expect(dispatchJob).toHaveBeenCalledWith('job-1') + }) + + it('drainOnStartup is a no-op when no queued jobs exist', async () => { + vi.mocked(jobRepo.findQueuedJobs).mockResolvedValue([]) + + await manager.drainOnStartup() + + expect(jobRepo.update).not.toHaveBeenCalled() + expect(dispatchJob).not.toHaveBeenCalled() + }) + }) + describe('admit - atomic status decision + row commit', () => { it('calls commit with submitted when limits are clear', async () => { vi.mocked(jobRepo.countQueuedJobs).mockResolvedValue(0) diff --git a/src/main/compute/concurrency-manager.ts b/src/main/compute/concurrency-manager.ts index a5d78072..e1b91fc1 100644 --- a/src/main/compute/concurrency-manager.ts +++ b/src/main/compute/concurrency-manager.ts @@ -132,6 +132,13 @@ export class ConcurrencyManager { await this.tryDispatchNext() } + // Drains the queue after a process restart: attempts to dispatch any queued jobs that can now + // fill available capacity slots. Should be called once during startup after the poller has been + // initialized. + async drainOnStartup(): Promise { + await this.tryDispatchNext() + } + // Query session status (active/queued counts, limits, provider ceilings). async getStatus(sessionId: string): Promise { const sessionLimit = this.sessionLimits.get(sessionId) ?? null diff --git a/src/main/ipc.ts b/src/main/ipc.ts index db081d65..d45ad242 100644 --- a/src/main/ipc.ts +++ b/src/main/ipc.ts @@ -380,6 +380,9 @@ const registerIpcHandlers = async ({ }) }) jobPoller.start() + // Drain queued jobs that were pending when the process last exited. Called after the poller starts + // so the poller's onJobUpdated hook is wired before any dispatch fires. + void computeService.drainQueuedJobs() // Augment computeService with getEnabledComputeHosts so the RPC server can serve list_compute. // Must preserve ComputeService's prototype methods (list/getDetails/submitJob/...) — see the helper. const computeServiceWithRegistry = attachEnabledComputeHosts(computeService, hostsRegistry) From 85eb5a3ca991147a110087929b14cb478b8ffe4a Mon Sep 17 00:00:00 2001 From: wen2zhou Date: Thu, 23 Jul 2026 09:01:17 +0000 Subject: [PATCH 3/4] refactor(compute): scope RemoteJobBadge job subscription to its session MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Subscribe to only the active session's job slice via useShallow instead of the whole jobsById map. applyUpdate swaps the map on every broadcast, so a job update in another session no longer re-renders this badge — only this session's own add/status changes do. The 1s elapsed-time tick still drives fresh durations independently. Co-Authored-By: Claude Opus 4.8 --- .../src/components/RemoteJobBadge.tsx | 29 ++++++++++--------- 1 file changed, 15 insertions(+), 14 deletions(-) diff --git a/src/renderer/src/components/RemoteJobBadge.tsx b/src/renderer/src/components/RemoteJobBadge.tsx index da097d3a..473bdb70 100644 --- a/src/renderer/src/components/RemoteJobBadge.tsx +++ b/src/renderer/src/components/RemoteJobBadge.tsx @@ -1,5 +1,6 @@ -import { useEffect, useMemo, useState } from 'react' +import { useEffect, useState } from 'react' import { Zap } from 'lucide-react' +import { useShallow } from 'zustand/react/shallow' import type { JobSummary } from '../../../shared/compute' import { Tooltip, TooltipContent, TooltipProvider, TooltipTrigger } from '@/components/ui/tooltip' @@ -22,10 +23,19 @@ export const RemoteJobBadge = ({ sessionId, onOpenJobList }: RemoteJobBadgeProps): React.JSX.Element | null => { - // Subscribe to the raw job map (not the allJobsForSession selector fn, whose reference is stable) - // so the badge re-renders the instant a job is added or changes status — applyUpdate replaces the - // map on every broadcast. Without this the badge would only refresh on the 1s elapsed-time tick. - const jobsById = useSessionJobStore((state) => state.jobsById) + // Subscribe to THIS session's jobs only. applyUpdate replaces the whole jobsById map on every + // broadcast, so subscribing to the stable allJobsForSession fn ref (as before) never re-rendered. + // useShallow compares the filtered+sorted array element-by-element: a broadcast for ANOTHER session + // keeps this slice's object references unchanged → no re-render, while any add or status change in + // THIS session produces a different element and re-renders immediately. The 1s elapsed-time tick + // below drives fresh durations independently of this subscription. + const allJobs = useSessionJobStore( + useShallow((state) => + Array.from(state.jobsById.values()) + .filter((j) => j.session_id === sessionId) + .sort((a, b) => b.created_at - a.created_at) + ) + ) const [now, setNow] = useState(() => Date.now()) // Tick every second to keep elapsed times fresh. @@ -34,15 +44,6 @@ export const RemoteJobBadge = ({ return () => clearInterval(id) }, []) - // Jobs for this session, newest first — recomputed whenever the job map changes. - const allJobs = useMemo( - () => - Array.from(jobsById.values()) - .filter((j) => j.session_id === sessionId) - .sort((a, b) => b.created_at - a.created_at), - [jobsById, sessionId] - ) - // Running jobs are dispatched to the remote host (running or submitted there); // queued jobs are waiting locally for a free concurrency slot. Both are in-flight and // keep the badge amber. Terminal jobs (success/failed/timeout/error) only count toward "N jobs". From 6b372bafd546749418ed401b16ec403fdee64794 Mon Sep 17 00:00:00 2001 From: wen2zhou Date: Thu, 23 Jul 2026 09:01:29 +0000 Subject: [PATCH 4/4] test(compute): harden queued-job drain tests - Replace the fixed 200ms setTimeout wait in the drainOnStartup integration test with a polling waitFor helper so it does not flake under CI load. - Add a multi-job drainOnStartup unit test asserting it drains every queued job up to the provider ceiling and leaves the remainder queued, making the drain contract explicit. Co-Authored-By: Claude Opus 4.8 --- .../compute/concurrency-integration.test.ts | 19 ++++++- src/main/compute/concurrency-manager.test.ts | 55 +++++++++++++++++++ 2 files changed, 72 insertions(+), 2 deletions(-) diff --git a/src/main/compute/concurrency-integration.test.ts b/src/main/compute/concurrency-integration.test.ts index da8d9cbb..4eb4b7bd 100644 --- a/src/main/compute/concurrency-integration.test.ts +++ b/src/main/compute/concurrency-integration.test.ts @@ -16,6 +16,20 @@ import type { SshRunner } from './ssh-runner' import type { ScpRunner } from './scp-runner' import { computeProviderId, type ComputeJob } from '../../shared/compute' +// Polls `predicate` until it returns true or `timeoutMs` elapses. Replaces fixed `setTimeout` waits +// for async dispatch, which flake under CI load when the work outruns the hard-coded delay. +async function waitFor( + predicate: () => boolean | Promise, + timeoutMs = 2000 +): Promise { + const deadline = Date.now() + timeoutMs + while (Date.now() < deadline) { + if (await predicate()) return + await new Promise((resolve) => setTimeout(resolve, 20)) + } + throw new Error(`waitFor timed out after ${timeoutMs}ms`) +} + // Mock the job-dispatcher module to prevent real SSH dispatches vi.mock('./job-dispatcher', async () => { const actual = await vi.importActual('./job-dispatcher') @@ -449,8 +463,9 @@ describe('ConcurrencyManager integration with ComputeService', () => { // No active jobs remain — call drainQueuedJobs to simulate startup drain await service.drainQueuedJobs() - // Wait for async dispatch to complete - await new Promise((resolve) => setTimeout(resolve, 200)) + // Wait for async dispatch to promote job-2 to 'submitted' — poll rather than a fixed delay so the + // test does not flake if dispatch is slower under CI load. + await waitFor(async () => (await jobRepo.get(result2.job_id))?.status === 'submitted') // job-2 should now be submitted const job2Updated = await jobRepo.get(result2.job_id) diff --git a/src/main/compute/concurrency-manager.test.ts b/src/main/compute/concurrency-manager.test.ts index 20d14440..4926321d 100644 --- a/src/main/compute/concurrency-manager.test.ts +++ b/src/main/compute/concurrency-manager.test.ts @@ -642,6 +642,61 @@ describe('ConcurrencyManager', () => { expect(jobRepo.update).not.toHaveBeenCalled() expect(dispatchJob).not.toHaveBeenCalled() }) + + it('drains every queued job up to the provider ceiling and leaves the rest queued', async () => { + // Drain must iterate the WHOLE queue (not stop after one promotion) while still respecting + // capacity: with a ceiling of 2 and 3 queued jobs, the first two are dispatched and the third + // stays queued. The active counter mirrors what a real DB read returns after each reservation. + const providerCeiling = 2 + vi.mocked(hostRepo.get).mockResolvedValue({ + concurrencyLimit: providerCeiling + } as ComputeHost) + + const queuedJobs: ComputeJob[] = [ + { + job_id: 'job-1', + session_id: 'session-1', + provider_id: 'ssh:cluster-a', + created_at: 1000, + status: 'queued' + } as ComputeJob, + { + job_id: 'job-2', + session_id: 'session-1', + provider_id: 'ssh:cluster-a', + created_at: 2000, + status: 'queued' + } as ComputeJob, + { + job_id: 'job-3', + session_id: 'session-1', + provider_id: 'ssh:cluster-a', + created_at: 3000, + status: 'queued' + } as ComputeJob + ] + vi.mocked(jobRepo.findQueuedJobs).mockResolvedValue(queuedJobs) + + // Active count rises as each job is reserved (status → 'submitted'), mirroring a real DB read. + let active = 0 + vi.mocked(jobRepo.countActiveBySession).mockImplementation(async () => active) + vi.mocked(jobRepo.countActiveByProvider).mockImplementation(async () => active) + vi.mocked(jobRepo.update).mockImplementation(async (_id, u) => { + if ((u as { status?: string }).status === 'submitted') active++ + return {} as ComputeJob + }) + + await manager.drainOnStartup() + + // job-1 and job-2 won slots; job-3 hit the ceiling and was left queued. + expect(jobRepo.update).toHaveBeenCalledWith('job-1', { status: 'submitted' }) + expect(jobRepo.update).toHaveBeenCalledWith('job-2', { status: 'submitted' }) + expect(jobRepo.update).not.toHaveBeenCalledWith('job-3', expect.anything()) + expect(dispatchJob).toHaveBeenCalledWith('job-1') + expect(dispatchJob).toHaveBeenCalledWith('job-2') + expect(dispatchJob).not.toHaveBeenCalledWith('job-3') + expect(active).toBe(providerCeiling) + }) }) describe('admit - atomic status decision + row commit', () => {