fix(compute): show queued jobs on the notebook bar and drain them on restart

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<void> {
+    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.

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<boolean>,
+  timeoutMs = 2000
+): Promise<void> {
+  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')
 

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)

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<void> {
+    await this.tryDispatchNext()
+  }
+
   // Query session status (active/queued counts, limits, provider ceilings).
   async getStatus(sessionId: string): Promise<SessionStatus> {
     const sessionLimit = this.sessionLimits.get(sessionId) ?? null

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)