diff --git a/src/main/compute/compute-service.ts b/src/main/compute/compute-service.ts index 1e4e57466..679bda8c9 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(). @@ -1466,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 42babfd3a..4eb4b7bd5 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') @@ -157,6 +171,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') @@ -380,6 +427,51 @@ 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 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) + 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 5b56b0412..4926321db 100644 --- a/src/main/compute/concurrency-manager.test.ts +++ b/src/main/compute/concurrency-manager.test.ts @@ -608,6 +608,97 @@ 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() + }) + + 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', () => { 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 a5d78072f..e1b91fc12 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 db081d65a..d45ad242b 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) diff --git a/src/renderer/src/components/RemoteJobBadge.render.test.tsx b/src/renderer/src/components/RemoteJobBadge.render.test.tsx index a594b0d4b..ada47b07a 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 b18b82b53..473bdb706 100644 --- a/src/renderer/src/components/RemoteJobBadge.tsx +++ b/src/renderer/src/components/RemoteJobBadge.tsx @@ -1,10 +1,11 @@ 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' 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 +14,28 @@ 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 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. @@ -26,10 +44,12 @@ export const RemoteJobBadge = ({ sessionId, onOpenJobList }: RemoteJobBadgeProps return () => clearInterval(id) }, []) - const allJobs = allJobsForSession(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 +57,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 +67,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 +91,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 +110,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 fdd6df158..58288f734 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)) +}