feat(job): introduce JobManager + SchedulerService backbone (Phase 1)

cherry-studio v2 had 6+ ad-hoc queue/scheduler implementations (Knowledge, FileProcessing, agent SchedulerService, TopicQueue, NotificationQueue, protocol heartbeats) with no shared registry, inconsistent cancel and progress semantics, and no cross-restart recovery outside agent_task. This commit lands the unified replacement: JobManager owns Job lifecycle and SchedulerService owns time-only scheduling, both reusable independently. Phase 1 ships the backbone only: jobTable + jobScheduleTable, entity services, 6-state machine with per-handler recovery (abandon/retry/ singleton), catch-up policy (skip-missed/after-startup), retry backoff, GC, idempotencyKey dedup, useJob renderer hook, and 4 reference docs. Four-layer lock model addresses libsql client-ts issue #288 (Layer 0 global dispatch mutex serializes all transactions). croner@^10 is introduced for cron expressions (zero-dep, Electron-friendly). Application.SHUTDOWN_TIMEOUT_MS is promoted to public for JobManager reuse. cron-parser stays until Phase 2 agent task migration. No business migrations in this commit — Knowledge/FileProcessing/agent SchedulerService remain untouched and migrate in Phase 2-4 per docs/references/job-and-scheduler/migration-checklist.md. Follow-ups for Phase 1 completion: DataApi GET /jobs handler, dummy.echo smoke test, integration tests, migration feasibility report.
2026-07-10 17:58:46 +08:00 · 2026-05-19 03:57:18 -07:00
parent e93c8cd943
commit 344c58eb8e
25 changed files with 6958 additions and 26 deletions
--- a/docs/references/job-and-scheduler/README.md
+++ b/docs/references/job-and-scheduler/README.md
@@ -0,0 +1,17 @@
+# Job & Scheduler
+
+Cherry Studio unified background job + time-scheduling system. Phase 1 (basic infrastructure) is implemented; business migrations (agent task, FileProcessing, Knowledge) are scheduled for Phase 2-4.
+
+| Doc | What it covers | Audience |
+|---|---|---|
+| [overview.md](./overview.md) | Architecture, two-service separation, DB-driven dispatch | New contributors |
+| [concurrency-and-locks.md](./concurrency-and-locks.md) | Four-layer lock model + business-level resource locks | Handler authors |
+| [handler-authoring.md](./handler-authoring.md) | How to write a JobHandler (recovery / retry / catchUp / progress) | Handler authors |
+| [migration-checklist.md](./migration-checklist.md) | Step-by-step checklist for migrating existing services | Phase 2-4 migrators |
+
+## Quick navigation
+
+- Need to enqueue background work? → see [overview.md / "When to use JobManager"](./overview.md)
+- Need to schedule a callback (cron / interval)? → see [overview.md / "When to use SchedulerService directly"](./overview.md)
+- Migrating from a custom queue? → see [migration-checklist.md](./migration-checklist.md)
+- Handler tripping over concurrent base writes? → see [concurrency-and-locks.md](./concurrency-and-locks.md)
--- a/docs/references/job-and-scheduler/concurrency-and-locks.md
+++ b/docs/references/job-and-scheduler/concurrency-and-locks.md
@@ -0,0 +1,59 @@
+# Concurrency & Locks (Four-Layer Model)
+
+JobManager uses four orthogonal lock layers to keep correctness under concurrent dispatch and protect business resources from cross-handler interference.
+
+## Layer 0: Global dispatch mutex (held by JobManager singleton)
+
+**Scope**: All dispatch transactions across all queues.
+**Held for**: Microseconds — the duration of one (count → fetch → claim) transaction.
+**Purpose**: libsql async client issue [#288](https://github.com/tursodatabase/libsql-client-ts/issues/288): default `busy_timeout` is not honored, multiple concurrent `db.transaction()` interlock each other into `SQLITE_BUSY`. Layer 0 serializes all dispatch transactions through a single Mutex so libsql never sees concurrent BEGIN.
+
+**Without this layer**: 1000+ queues triggering dispatch simultaneously hit SQLITE_BUSY immediately under high load.
+
+## Layer 1: Per-queue dispatch mutex (per DispatchQueue instance)
+
+**Scope**: Single queue's (count → fetch → claim) section.
+**Held for**: Microseconds.
+**Purpose**: Two concurrent dispatchers on the same queue must not both see "queueActive < concurrency" and both claim the next pending row.
+
+**Why both Layer 0 + Layer 1?** Layer 0 only serializes the tx-begin; Layer 1 ensures the count → claim sequence is atomic for the queue. Acquisition order is fixed (per-queue first, then global) to avoid deadlock between the two layers.
+
+## Layer 2: Queue concurrency limit (`DispatchQueue.concurrency`)
+
+**Scope**: How many handlers in this queue run simultaneously.
+**Held for**: The full handler execution period.
+**Purpose**: Throttle parallelism per queue — e.g., per-base indexing at concurrency=5.
+
+**Critical**: Layer 2 alone does NOT protect against same-resource concurrent writes after restart. If a process crashes mid-write to a vector store, the new handler instance (recovery='retry') starts before the old write is fully observed by the OS. Layer 2 only counts active *jobs*, not in-flight *writes*.
+
+## Layer 3: Business-level mutex (handler-owned)
+
+**Scope**: Resource-specific (vector store write, file IO, external API rate limit).
+**Held for**: Decided by handler.
+**Purpose**: Serialize critical sections that survive process restart.
+
+**Example**: `KnowledgeRuntimeService.runWithBaseWriteLockForBase(baseId, fn)` — singleton-owned mutex keyed by baseId, so even if recovery spawns a new handler instance, both old and new code paths acquire the same lock and write atomically.
+
+## Common trap
+
+**"Setting `queue=base.${baseId}` with `concurrency=1` replaces business-level locks."** WRONG.
+
+Why: After crash + restart, recovery='retry' starts a *new* handler instance for the same job. The new instance enters Layer 2 (queue active count from DB reflects the new running row), but the *old* in-flight write may still be observed by the filesystem / vector store at OS level. Without Layer 3, you have two writers to the same resource.
+
+**Always combine Layer 2 (parallelism cap) with Layer 3 (resource serialization).**
+
+## Summary diagram
+
+```
+┌─ Layer 0: Global dispatch mutex ─────────────┐  Serializes all libsql txs
+│ ┌─ Layer 1: Per-queue dispatch mutex ──────┐ │  Serializes same-queue claim
+│ │ ┌─ Layer 2: Queue concurrency limit ───┐ │ │  N handlers per queue
+│ │ │ ┌─ Layer 3: Business mutex ────────┐ │ │ │  Resource-level serialization
+│ │ │ │ handler.execute() runs           │ │ │ │  (handler-owned, survives restart)
+│ │ │ └──────────────────────────────────┘ │ │ │
+│ │ └──────────────────────────────────────┘ │ │
+│ └──────────────────────────────────────────┘ │
+└──────────────────────────────────────────────┘
+       ↑ outside the orchestrator's lock — long
+       ↑ inside the orchestrator's lock — microseconds
+```
--- a/docs/references/job-and-scheduler/handler-authoring.md
+++ b/docs/references/job-and-scheduler/handler-authoring.md
@@ -0,0 +1,81 @@
+# Handler Authoring
+
+Phase 1 ship guarantees this doc has four sections. Further worked examples (retry / singleton recovery, failure-rate breaker, business-level mutex) are backported during Phase 2-4 business migrations to avoid speculative code that bit-rots before a real consumer appears.
+
+## 1. dummy.echo (minimal handler)
+
+```typescript
+import { jobManager } from '@main/core/job/JobManager'
+
+declare module '@main/core/job/jobRegistry' {
+  interface JobRegistry {
+    'dummy.echo': { message: string }
+  }
+}
+
+jobManager.registerHandler('dummy.echo', {
+  recovery: 'abandon',
+  defaultConcurrency: 1,
+  defaultTimeoutMs: 5000,
+  async execute(ctx) {
+    ctx.logger.info('echo start', { message: ctx.input.message })
+    await new Promise<void>((resolve, reject) => {
+      const t = setTimeout(resolve, 1000)
+      ctx.signal.addEventListener('abort', () => {
+        clearTimeout(t)
+        reject(new Error('aborted'))
+      })
+    })
+    ctx.reportProgress(100, { stage: 'done' })
+    return `echo: ${ctx.input.message}`
+  }
+})
+```
+
+## 2. Remote-poll pattern (cross-restart hand-off)
+
+```typescript
+async execute(ctx: JobContext<RemotePollInput>): Promise<RemoteResult> {
+  let providerTaskId = ctx.metadata.providerTaskId as string | undefined
+  if (!providerTaskId) {
+    providerTaskId = await startRemote(ctx.input, { signal: ctx.signal })
+    // CRITICAL: await — without persistence the restart-recovery will re-submit
+    // the remote job, wasting user quota and producing parallel external tasks.
+    await ctx.patchMetadata({ providerTaskId })
+  }
+  while (!ctx.signal.aborted) {
+    const status = await pollRemote(providerTaskId, { signal: ctx.signal })
+    if (status.done) return status.result
+    ctx.reportProgress(status.percent, { stage: status.stage })
+    await sleep(POLL_INTERVAL_MS, { signal: ctx.signal })
+  }
+  throw new Error('AbortError: cancelled')
+}
+```
+
+Anti-pattern: `while (true)` (cannot be cancelled), `await sleep(N)` without signal (delays cancellation by up to N ms).
+
+## 3. recovery × catchUpPolicy matrix (6 cells)
+
+| Recovery × CatchUp | `skip-missed` | `after-startup` |
+|---|---|---|
+| **abandon** | Pre-existing non-terminal jobs → cancelled on startup. Missed schedule fires emit `onMissed` (observability) but enqueue nothing. | Same as left, PLUS enqueue make-up job after `minutes * 60_000` ms delay. |
+| **retry** | running → pending on startup; delayed kept as-is. Missed fires emit `onMissed` only. | Same as left, PLUS enqueue make-up after N min. |
+| **singleton** | Keep newest non-terminal, cancel the rest. Missed fires emit `onMissed` only. | Same as left, PLUS enqueue make-up after N min (joins the single-instance slot when free). |
+
+## 4. Error codes (renderer maps via i18next)
+
+| Code | Origin | Retryable | Meaning |
+|---|---|---|---|
+| `JOB_UNKNOWN_TYPE` | enqueue | no | No handler registered for this type |
+| `JOB_PAYLOAD_TOO_LARGE` | enqueue | no | Input JSON exceeds 1MB |
+| `JOB_CANCEL_REASON_TOO_LONG` | cancel | no | Cancel reason exceeds 500 chars |
+| `JOB_SCHEDULE_NOT_FOUND_BY_NAME` | schedule by-name API | no | Provided (type, name) does not exist |
+| `JOB_SCHEDULE_NAME_REQUIRED` | schedule by-name API | no | Multi-instance type but no name passed |
+| `JOB_SCHEDULE_NAME_CONFLICT` | schedule create | no | (type, name) already exists |
+| `JOB_SCHEDULE_SINGLETON_EXISTS` | schedule create | no | Unnamed schedule attempted on a type that already has schedules |
+| `JOB_HANDLER_TIMEOUT` | runtime | yes | Handler exceeded `timeoutMs` |
+| `JOB_HANDLER_THREW` | runtime | yes | Handler threw a non-abort error |
+| `JOB_CANCELLED` | recovery / cancel | no | Job cancelled by user, recovery, or shutdown |
+
+Renderer: `t(\`errors.jobs.${code.toLowerCase()}\`, params)`.
--- a/docs/references/job-and-scheduler/migration-checklist.md
+++ b/docs/references/job-and-scheduler/migration-checklist.md
@@ -0,0 +1,48 @@
+# Migration Checklist (Phase 2-4)
+
+Use this checklist when migrating an existing service (KnowledgeRuntime / FileProcessing / agent task / heartbeat) to the unified JobManager. Each phase is a separate project — this doc is the per-handler discipline applied within each.
+
+## Per-handler
+
+- [ ] Choose `recovery` strategy:
+  - `abandon` — fire-and-forget (heartbeats, notifications)
+  - `retry` — "must complete" (ingestion, indexing, model sync)
+  - `singleton` — "at most one active per type" (init, periodic refresh)
+- [ ] Set `defaultQueue` if per-resource serialization is needed (e.g. `base.${baseId}` for per-base writes)
+- [ ] Set `defaultConcurrency` based on resource budget (vector store / GPU / network)
+- [ ] Configure `defaultRetryPolicy` if retries are valuable
+- [ ] Set `defaultTimeoutMs` if the handler can be long-running
+- [ ] Implement `execute`:
+  - Respect `ctx.signal.aborted` in every loop body and every `await`
+  - Use `ctx.patchMetadata` for cross-restart state hand-off (e.g., remote task IDs)
+  - Use `ctx.reportProgress(percent, detail)` for renderer-visible progress
+  - Do NOT use `while (true)` — always `while (!ctx.signal.aborted)`
+- [ ] Implement `onMissed` if business needs catch-up observability or breaker
+- [ ] Implement `onSettled` if business needs failure-rate breaker — query
+      `jobService.listRecentTerminalByScheduleId(scheduleId, N)` for the truth, do NOT build a separate counter table
+- [ ] Add JobRegistry type binding via TypeScript declaration merging
+- [ ] Register handler in the owning service's `onInit`
+
+## Data migration (per business)
+
+- [ ] Map existing rows → `jobTable` / `jobScheduleTable` rows
+- [ ] Run migration via `v2-refactor-temp/tools/data-classify` flow
+- [ ] Update v2 migrators in `src/main/data/migration/v2/migrators/` for clean-restart safety
+- [ ] Add a `v2-refactor-temp/docs/breaking-changes/` entry if user-visible behavior changes (e.g., agent task: per-attempt log → single row per enqueue)
+- [ ] Delete or thin-facade the legacy service (keep IPC entry points; redirect to JobManager)
+
+## Validation per handler
+
+- [ ] Smoke test: enqueue → terminal happy path
+- [ ] Restart test: spawn jobs, `kill -9`, verify recovery acts per `recovery` strategy
+- [ ] Concurrency test: assert per-queue concurrency cap is respected (and per-resource Layer 3 lock for write-heavy handlers)
+- [ ] Cancel test: cancel during run, verify `cancelled` terminal status and handler observed `ctx.signal.aborted`
+- [ ] Catch-up test (if scheduled): freeze time past nextRun, verify `onMissed` event and (for `after-startup`) the make-up job
+
+## Cross-cutting verification (each phase)
+
+- [ ] `pnpm lint` + `pnpm test` + `pnpm format` clean
+- [ ] DataApi paths (if added) registered in `apiPaths.ts` / `apiTypes.ts`
+- [ ] cacheSchemas entries (if any new cache keys) registered
+- [ ] No commits to legacy services unless they're being deleted/refactored
+- [ ] Phase summary added to PR description (what migrated, what stayed)
--- a/docs/references/job-and-scheduler/overview.md
+++ b/docs/references/job-and-scheduler/overview.md
@@ -0,0 +1,82 @@
+# Job & Scheduler — Architecture Overview
+
+Two independent main-process lifecycle services:
+
+| Service | Role | Persistence | Direct consumer |
+|---|---|---|---|
+| **SchedulerService** | "When to fire a callback" — cron / interval / once. Stateless. | None | JobManager + any module needing simple time scheduling |
+| **JobManager** | "Job lifecycle" — registry, persistence, 6-state machine, dispatch, recovery | `jobTable` + `jobScheduleTable` | All background work |
+
+**Layering rule**: SchedulerService is unaware of Jobs. JobManager uses SchedulerService to arm schedules. Business modules pick one based on need:
+
+- Need cron + persistent observability + retry → register a JobHandler + use `jobManager.registerJobSchedule()`
+- Need cron only (heartbeat-style, no persistence) → `schedulerService.registerSchedule()` directly
+- Need recurring service-internal GC / self-check → `BaseService.registerInterval` (project convention, not SchedulerService)
+
+## DB-driven dispatch
+
+`jobTable` is the **single source of truth**. Memory state (handlers Map, queues Map, AbortControllers) is a derived view that JobManager rebuilds on every startup.
+
+Each queue has a `DispatchQueue` instance holding `{ name, concurrency, mutex }`. The dispatch loop:
+
+1. Acquire **Layer 0** global mutex (libsql tx serialization)
+2. Acquire **Layer 1** per-queue mutex (same-queue dispatch serialization)
+3. Inside one DB transaction:
+   - Count queue-active jobs → check queue.concurrency
+   - Count globally-running jobs → check globalMaxConcurrency
+   - SELECT next pending → UPDATE to running (claim)
+4. Release both mutexes
+5. Spawn handler.execute outside the lock
+6. Queue a microtask to dispatch the same queue again (fill next slot)
+
+Spawning happens *outside* the mutex — the handler executes for seconds/minutes while new dispatches proceed.
+
+## Six-state state machine
+
+```
+                  ┌── retry backoff (delayed) ──┐
+                  ▼                              │
+   enqueue → pending → running → completed       │
+                  │       │                      │
+                  │       └→ failed ─────────────┘ (if retryable && attempt < max)
+                  │       └→ cancelled (terminal)
+                  └→ delayed → (scheduledAt ≤ now) → pending
+```
+
+Terminal states (`completed` / `failed` / `cancelled`) are never reopened. Retry re-enters `delayed` then transitions back to `pending` when scheduledAt elapses.
+
+## Why DB-driven and not in-memory queue?
+
+We considered BullMQ / bee-queue / better-queue / agenda / graphile-worker / bree etc. and selected this design because:
+
+- All persistence already in SQLite (no Redis / MongoDB / PostgreSQL dependency)
+- Restart recovery is automatic — memory replays from DB
+- Race safety needs only one mutex pair (Layer 0 + Layer 1) around `count → claim`
+- No double-source-of-truth bookkeeping (PQueue + DB) and its sync discipline
+
+Throughput: ~200 dispatch/s at single-process libsql throughput, well above Cherry Studio's largest scenario (1000+ knowledge bases, each with concurrency=5, never exceeds globalMaxConcurrency=50 simultaneous running jobs).
+
+## Strongly-typed JobRegistry
+
+Business modules use TypeScript declaration merging to register `type → payload` mapping:
+
+```typescript
+declare module '@main/core/job/jobRegistry' {
+  interface JobRegistry {
+    'agent.task': AgentTaskPayload
+    'knowledge.index-leaf': IndexLeafPayload
+  }
+}
+```
+
+After this declaration:
+
+- `jobManager.enqueue('agent.task', payload)` is compile-time type-checked
+- Renaming a type surfaces every call site via the TypeScript error pipeline
+- Wrong payload shape is a compile error
+
+## See also
+
+- [concurrency-and-locks.md](./concurrency-and-locks.md) — The full four-layer lock model
+- [handler-authoring.md](./handler-authoring.md) — How to write a handler
+- [migration-checklist.md](./migration-checklist.md) — Migrating existing services
--- a/migrations/sqlite-drizzle/0025_left_black_tarantula.sql
+++ b/migrations/sqlite-drizzle/0025_left_black_tarantula.sql
@@ -0,0 +1,50 @@
+CREATE TABLE `job_schedule` (
+	`id` text PRIMARY KEY NOT NULL,
+	`type` text NOT NULL,
+	`name` text,
+	`trigger` text NOT NULL,
+	`job_input_template` text NOT NULL,
+	`enabled` integer DEFAULT true NOT NULL,
+	`next_run` integer,
+	`last_run` integer,
+	`catch_up_policy` text NOT NULL,
+	`metadata` text DEFAULT '{}' NOT NULL,
+	`created_at` integer NOT NULL,
+	`updated_at` integer NOT NULL
+);
+--> statement-breakpoint
+CREATE UNIQUE INDEX `job_schedule_type_name_uq` ON `job_schedule` (`type`,`name`);--> statement-breakpoint
+CREATE INDEX `job_schedule_enabled_next_run_idx` ON `job_schedule` (`enabled`,`next_run`);--> statement-breakpoint
+CREATE INDEX `job_schedule_type_idx` ON `job_schedule` (`type`);--> statement-breakpoint
+CREATE TABLE `job` (
+	`id` text PRIMARY KEY NOT NULL,
+	`type` text NOT NULL,
+	`status` text NOT NULL,
+	`priority` integer DEFAULT 0 NOT NULL,
+	`queue` text NOT NULL,
+	`idempotency_key` text,
+	`schedule_id` text,
+	`scheduled_at` integer NOT NULL,
+	`started_at` integer,
+	`finished_at` integer,
+	`attempt` integer DEFAULT 0 NOT NULL,
+	`max_attempts` integer DEFAULT 3 NOT NULL,
+	`input` text NOT NULL,
+	`output` text,
+	`error` text,
+	`parent_id` text,
+	`cancel_requested` integer DEFAULT false NOT NULL,
+	`metadata` text DEFAULT '{}' NOT NULL,
+	`timeout_ms` integer,
+	`created_at` integer NOT NULL,
+	`updated_at` integer NOT NULL,
+	FOREIGN KEY (`schedule_id`) REFERENCES `job_schedule`(`id`) ON UPDATE no action ON DELETE set null,
+	FOREIGN KEY (`parent_id`) REFERENCES `job`(`id`) ON UPDATE no action ON DELETE set null,
+	CONSTRAINT "job_status_check" CHECK("job"."status" IN ('pending','delayed','running','completed','failed','cancelled'))
+);
+--> statement-breakpoint
+CREATE INDEX `job_queue_status_scheduled_at_idx` ON `job` (`queue`,`status`,`scheduled_at`);--> statement-breakpoint
+CREATE INDEX `job_status_idx` ON `job` (`status`);--> statement-breakpoint
+CREATE INDEX `job_schedule_id_finished_at_idx` ON `job` (`schedule_id`,`finished_at`);--> statement-breakpoint
+CREATE INDEX `job_parent_id_idx` ON `job` (`parent_id`);--> statement-breakpoint
+CREATE UNIQUE INDEX `job_idempotency_key_partial_uq` ON `job` (`idempotency_key`) WHERE "job"."idempotency_key" IS NOT NULL AND "job"."status" NOT IN ('completed','failed','cancelled');
--- a/migrations/sqlite-drizzle/meta/0025_snapshot.json
+++ b/migrations/sqlite-drizzle/meta/0025_snapshot.json
--- a/migrations/sqlite-drizzle/meta/_journal.json
+++ b/migrations/sqlite-drizzle/meta/_journal.json
@@ -175,6 +175,13 @@
      "when": 1778738430302,
      "tag": "0024_eager_patriot",
      "breakpoints": true
+    },
+    {
+      "idx": 25,
+      "version": "6",
+      "when": 1779183817228,
+      "tag": "0025_left_black_tarantula",
+      "breakpoints": true
    }
  ],
  "version": "7"
--- a/package.json
+++ b/package.json
@@ -315,6 +315,7 @@
    "commander": "^14.0.2",
    "concurrently": "^9.2.1",
    "cors": "2.8.5",
+    "croner": "^10.0.1",
    "country-flag-emoji-polyfill": "0.1.8",
    "dayjs": "^1.11.11",
    "dexie": "^4.0.8",
--- a/packages/shared/data/api/schemas/jobs.ts
+++ b/packages/shared/data/api/schemas/jobs.ts
@@ -0,0 +1,238 @@
+/**
+ * Jobs / Schedules domain API Schema definitions.
+ *
+ * Entity schemas live here (Rule C/D in api/README.md). Field atoms,
+ * discriminated unions for Trigger / CatchUpPolicy / RetryPolicy, and the
+ * Job + JobSchedule snapshots that both the DataApi response and the renderer
+ * useJob hook consume.
+ *
+ * NOTE: Handler runtime types (JobHandler / JobContext / JobMissEvent /
+ * JobSettledEvent) are NOT in shared — they belong to main-process internals
+ * at src/main/core/job/types.ts. Renderer never instantiates them.
+ */
+
+import * as z from 'zod'
+
+// ============================================================================
+// Field atoms
+// ============================================================================
+
+export const JobStatusAtomSchema = z.enum(['pending', 'delayed', 'running', 'completed', 'failed', 'cancelled'])
+export type JobStatus = z.infer<typeof JobStatusAtomSchema>
+
+/** Terminal states: jobs in these states are finished and never resume. */
+export const TERMINAL_JOB_STATUSES = ['completed', 'failed', 'cancelled'] as const satisfies readonly JobStatus[]
+
+export const isTerminalStatus = (status: JobStatus): boolean =>
+  (TERMINAL_JOB_STATUSES as readonly JobStatus[]).includes(status)
+
+/**
+ * Stable error structure crossing the IPC boundary. `code` is an English
+ * constant; renderer maps it to a localized message via
+ * `t(\`errors.jobs.${code.toLowerCase()}\`, params)`.
+ */
+export const JobErrorSchema = z.strictObject({
+  code: z.string(),
+  message: z.string(),
+  params: z.record(z.string(), z.unknown()).optional(),
+  retryable: z.boolean()
+})
+export type JobError = z.infer<typeof JobErrorSchema>
+
+// ============================================================================
+// Trigger (discriminated union: cron / interval / once)
+// ============================================================================
+
+export const CronTriggerSchema = z.strictObject({
+  kind: z.literal('cron'),
+  expr: z.string().min(1),
+  timezone: z.string().optional(),
+  /** Stop after N firings (croner maxRuns). For trial/test windows. */
+  limit: z.number().int().min(1).optional()
+})
+
+export const IntervalTriggerSchema = z.strictObject({
+  kind: z.literal('interval'),
+  ms: z.number().int().min(1),
+  anchor: z.enum(['createdAt', 'lastRun']).optional()
+})
+
+export const OnceTriggerSchema = z.strictObject({
+  kind: z.literal('once'),
+  /** Unix ms timestamp at which to fire exactly once. */
+  at: z.number().int().min(0)
+})
+
+export const TriggerSchema = z.discriminatedUnion('kind', [CronTriggerSchema, IntervalTriggerSchema, OnceTriggerSchema])
+export type Trigger = z.infer<typeof TriggerSchema>
+
+// ============================================================================
+// CatchUpPolicy (Phase 1: skip-missed / after-startup; after-idle descoped)
+// ============================================================================
+
+export const CatchUpPolicySchema = z.discriminatedUnion('kind', [
+  z.strictObject({ kind: z.literal('skip-missed') }),
+  z.strictObject({ kind: z.literal('after-startup'), minutes: z.number().int().min(0) })
+])
+export type CatchUpPolicy = z.infer<typeof CatchUpPolicySchema>
+
+// ============================================================================
+// RetryPolicy
+// ============================================================================
+
+export const RetryPolicySchema = z.strictObject({
+  maxAttempts: z.number().int().min(1),
+  backoff: z.enum(['exponential', 'fixed', 'none']),
+  baseDelayMs: z.number().int().min(0),
+  maxDelayMs: z.number().int().min(0)
+})
+export type RetryPolicy = z.infer<typeof RetryPolicySchema>
+
+// ============================================================================
+// Job entity (renderer-visible snapshot)
+// ============================================================================
+
+export const JobSnapshotSchema = z.strictObject({
+  id: z.string(),
+  type: z.string(),
+  status: JobStatusAtomSchema,
+  priority: z.number().int(),
+  queue: z.string(),
+  idempotencyKey: z.string().nullable(),
+  scheduleId: z.string().nullable(),
+  scheduledAt: z.string(),
+  startedAt: z.string().nullable(),
+  finishedAt: z.string().nullable(),
+  attempt: z.number().int().min(0),
+  maxAttempts: z.number().int().min(1),
+  input: z.unknown(),
+  output: z.unknown().nullable(),
+  error: JobErrorSchema.nullable(),
+  parentId: z.string().nullable(),
+  cancelRequested: z.boolean(),
+  metadata: z.record(z.string(), z.unknown()),
+  timeoutMs: z.number().int().nullable(),
+  createdAt: z.string(),
+  updatedAt: z.string()
+})
+export type JobSnapshot = z.infer<typeof JobSnapshotSchema>
+
+// ============================================================================
+// JobSchedule entity
+// ============================================================================
+
+export const JobScheduleSnapshotSchema = z.strictObject({
+  id: z.string(),
+  type: z.string(),
+  name: z.string().nullable(),
+  trigger: TriggerSchema,
+  jobInputTemplate: z.unknown(),
+  enabled: z.boolean(),
+  nextRun: z.string().nullable(),
+  lastRun: z.string().nullable(),
+  catchUpPolicy: CatchUpPolicySchema,
+  metadata: z.record(z.string(), z.unknown()),
+  createdAt: z.string(),
+  updatedAt: z.string()
+})
+export type JobScheduleSnapshot = z.infer<typeof JobScheduleSnapshotSchema>
+
+// ============================================================================
+// JobProgress (cache value at jobs.progress.${id}, never DB-persisted)
+// ============================================================================
+
+export const JobProgressSchema = z.strictObject({
+  progress: z.number().min(0).max(100),
+  detail: z.unknown().optional()
+})
+export type JobProgress = z.infer<typeof JobProgressSchema>
+
+// ============================================================================
+// DTO schemas (request payloads) — used by DataApi handlers + JobManager IPC
+// ============================================================================
+
+/**
+ * Enqueue input. `type` is checked against the runtime handler registry by
+ * JobManager.enqueue; the schema only enforces shape. `input` is unknown here
+ * because JobRegistry compile-time mapping is main-process only.
+ */
+export const EnqueueJobInputSchema = z.strictObject({
+  type: z.string().min(1),
+  input: z.unknown(),
+  queue: z.string().optional(),
+  priority: z.number().int().optional(),
+  idempotencyKey: z.string().optional(),
+  scheduledAt: z.string().optional(),
+  parentId: z.string().optional(),
+  timeoutMs: z.number().int().min(1).optional(),
+  maxAttempts: z.number().int().min(1).optional(),
+  metadata: z.record(z.string(), z.unknown()).optional()
+})
+export type EnqueueJobDto = z.infer<typeof EnqueueJobInputSchema>
+
+export const CancelJobInputSchema = z.strictObject({
+  /** Max 500 chars; enforced again in JobManager IPC validation. */
+  reason: z.string().max(500).optional()
+})
+export type CancelJobDto = z.infer<typeof CancelJobInputSchema>
+
+/**
+ * Name soft-constraint validator. Length 1-200, no control chars, trim
+ * surrounding whitespace, no `__` prefix (reserved for system schedules).
+ * Allows Unicode (中文/emoji ok) — name is a user-facing label.
+ */
+export const JobScheduleNameAtomSchema = z
+  .string()
+  .min(1)
+  .max(200)
+  .refine((s) => s === s.trim(), { message: 'name must be trimmed' })
+  .refine(
+    (s) => {
+      for (let i = 0; i < s.length; i++) {
+        const code = s.charCodeAt(i)
+        if (code === 0 || code === 9 || code === 10 || code === 13) return false
+      }
+      return true
+    },
+    { message: 'name cannot contain control characters (NUL/TAB/LF/CR)' }
+  )
+  .refine((s) => !s.startsWith('__'), { message: 'name cannot start with "__" (reserved)' })
+
+export const CreateJobScheduleInputSchema = z.strictObject({
+  type: z.string().min(1),
+  name: JobScheduleNameAtomSchema.nullable().optional(),
+  trigger: TriggerSchema,
+  jobInputTemplate: z.unknown(),
+  catchUpPolicy: CatchUpPolicySchema,
+  metadata: z.record(z.string(), z.unknown()).optional(),
+  enabled: z.boolean().optional()
+})
+export type CreateJobScheduleDto = z.infer<typeof CreateJobScheduleInputSchema>
+
+export const UpdateJobScheduleInputSchema = z.strictObject({
+  name: JobScheduleNameAtomSchema.nullable().optional(),
+  trigger: TriggerSchema.optional(),
+  jobInputTemplate: z.unknown().optional(),
+  catchUpPolicy: CatchUpPolicySchema.optional(),
+  enabled: z.boolean().optional(),
+  metadata: z.record(z.string(), z.unknown()).optional()
+})
+export type UpdateJobScheduleDto = z.infer<typeof UpdateJobScheduleInputSchema>
+
+// ============================================================================
+// Error codes (constants for JobManager + DataApi handler + renderer i18n)
+// ============================================================================
+
+export const JOB_ERROR_CODES = {
+  UNKNOWN_TYPE: 'JOB_UNKNOWN_TYPE',
+  PAYLOAD_TOO_LARGE: 'JOB_PAYLOAD_TOO_LARGE',
+  CANCEL_REASON_TOO_LONG: 'JOB_CANCEL_REASON_TOO_LONG',
+  SCHEDULE_NOT_FOUND_BY_NAME: 'JOB_SCHEDULE_NOT_FOUND_BY_NAME',
+  SCHEDULE_NAME_REQUIRED: 'JOB_SCHEDULE_NAME_REQUIRED',
+  SCHEDULE_NAME_CONFLICT: 'JOB_SCHEDULE_NAME_CONFLICT',
+  SCHEDULE_SINGLETON_EXISTS: 'JOB_SCHEDULE_SINGLETON_EXISTS',
+  HANDLER_TIMEOUT: 'JOB_HANDLER_TIMEOUT',
+  HANDLER_THREW: 'JOB_HANDLER_THREW',
+  CANCELLED: 'JOB_CANCELLED'
+} as const
+export type JobErrorCode = (typeof JOB_ERROR_CODES)[keyof typeof JOB_ERROR_CODES]
--- a/packages/shared/data/cache/cacheSchemas.ts
+++ b/packages/shared/data/cache/cacheSchemas.ts
@@ -1,3 +1,4 @@
+import type { JobProgress, JobSnapshot } from '@shared/data/api/schemas/jobs'
 import type { MiniAppRegion } from '@shared/data/types/miniApp'

 import type * as CacheValueTypes from './cacheValueTypes'
@@ -248,13 +249,24 @@ export type SharedCacheSchema = {
  // API key rotation state (cross-window, tracks last used key per provider)
  'web_search.provider.last_used_key.${providerId}': string
  'ocr.provider.last_used_key.${providerId}': string
+  // Job system: state snapshot + progress, broadcast main → all windows. TTL 60s
+  // (JobManager sets ttl when calling setShared, so cache miss after a job
+  // terminates is acceptable — useJob falls back to dataApi.get).
+  // Value is nullable: template default is `null`, replaced by JobSnapshot when
+  // a concrete job exists. Renderer treats null as cache miss.
+  'jobs.state.${jobId}': JobSnapshot | null
+  'jobs.progress.${jobId}': JobProgress
 }

 export const DefaultSharedCache: SharedCacheSchema = {
  'chat.web_search.active_searches': {},
  'feature.openclaw.gateway_status': 'stopped',
  'web_search.provider.last_used_key.${providerId}': '',
-  'ocr.provider.last_used_key.${providerId}': ''
+  'ocr.provider.last_used_key.${providerId}': '',
+  // Template defaults are placeholders never consumed at runtime — concrete
+  // keys are populated by JobManager when actual jobs exist.
+  'jobs.state.${jobId}': null,
+  'jobs.progress.${jobId}': { progress: 0 }
 }

 /**
--- a/pnpm-lock.yaml
+++ b/pnpm-lock.yaml
--- a/src/main/core/application/Application.ts
+++ b/src/main/core/application/Application.ts
@@ -29,7 +29,7 @@ interface QuitPreventionHold extends Disposable {
 * Manages services, windows, and Electron app events
 */
 export class Application {
-  private static readonly SHUTDOWN_TIMEOUT_MS = 5000
+  public static readonly SHUTDOWN_TIMEOUT_MS = 5000

  private static instance: Application | null = null
  private container: ServiceContainer
--- a/src/main/core/application/serviceRegistry.ts
+++ b/src/main/core/application/serviceRegistry.ts
@@ -2,6 +2,8 @@ import { CacheService } from '@data/CacheService'
 import { DataApiService } from '@data/DataApiService'
 import { DbService } from '@data/db/DbService'
 import { PreferenceService } from '@data/PreferenceService'
+import { JobManager } from '@main/core/job/JobManager'
+import { SchedulerService } from '@main/core/scheduler/SchedulerService'
 import { WindowManager } from '@main/core/window/WindowManager'
 import { AgentBootstrapService } from '@main/services/AgentBootstrapService'
 import { AnalyticsService } from '@main/services/AnalyticsService'
@@ -107,7 +109,9 @@ export const services = {
  KnowledgeRuntimeService,
  AgentBootstrapService,
  ApiServerService,
-  AppUpdaterService
+  AppUpdaterService,
+  SchedulerService,
+  JobManager
 } as const

 /** Auto-derived service name to instance type mapping */
--- a/src/main/core/job/JobManager.ts
+++ b/src/main/core/job/JobManager.ts
@@ -0,0 +1,840 @@
+import { application } from '@application'
+import type { InsertJobRow, JobRow } from '@data/db/schemas/job'
+import { jobScheduleService } from '@data/services/JobScheduleService'
+import { jobService } from '@data/services/JobService'
+import { loggerService } from '@logger'
+import { Application } from '@main/core/application/Application'
+import { BaseService, DependsOn, Injectable, Phase, ServicePhase } from '@main/core/lifecycle'
+import type { Disposable } from '@main/core/lifecycle/event'
+import {
+  JOB_ERROR_CODES,
+  type JobError,
+  type JobScheduleSnapshot,
+  type JobSnapshot,
+  type RetryPolicy,
+  type Trigger
+} from '@shared/data/api/schemas/jobs'
+import { Mutex } from 'async-mutex'
+
+import type { JobPayloadOf, JobType } from './jobRegistry'
+import { computeCatchUpAction } from './runtime/catchUp'
+import { DispatchQueue } from './runtime/DispatchQueue'
+import { runStartupRecovery } from './runtime/recovery'
+import {
+  type EnqueueOptions,
+  JOB_PROGRESS_KEY_PREFIX,
+  JOB_STATE_KEY_PREFIX,
+  type JobContext,
+  type JobHandle,
+  type JobHandler,
+  type JobScheduleRegistrationInput
+} from './types'
+
+const logger = loggerService.withContext('JobManager')
+
+/** Default retry policy used when handler does not declare one. */
+const DEFAULT_RETRY_POLICY: RetryPolicy = {
+  maxAttempts: 3,
+  backoff: 'exponential',
+  baseDelayMs: 1000,
+  maxDelayMs: 60_000
+}
+
+const MAX_INPUT_BYTES = 1_048_576 // 1MB
+const MAX_CANCEL_REASON_CHARS = 500
+
+const DEFAULT_GLOBAL_MAX_CONCURRENCY = 50
+const GC_INTERVAL_MS = 60 * 60 * 1000 // 1h
+const GC_TERMINAL_TTL_MS = 7 * 24 * 60 * 60 * 1000 // 7 days
+const GC_KEEP_PER_TYPE = 100
+const DELAYED_PROMOTION_INTERVAL_MS = 5 * 60 * 1000 // 5min
+
+interface FinishedResolver {
+  resolve: (snapshot: JobSnapshot) => void
+  promise: Promise<JobSnapshot>
+}
+
+/**
+ * Job orchestration: registers handlers, persists job rows, runs DB-driven
+ * dispatch with Layer 0 + Layer 1 mutex, executes handler callbacks, manages
+ * 6-state state machine, schedule registry, retry backoff, startup recovery,
+ * catch-up detection, GC.
+ *
+ * Phase 1 boundaries (see plan):
+ *   - GC hardcoded (1h sweep + 100 per type + 7-day TTL)
+ *   - globalMaxConcurrency = 50 (constructor default, not configurable)
+ *   - No worker / child_process executor
+ *   - No DAG / DLQ / priority preemption
+ */
+@Injectable('JobManager')
+@ServicePhase(Phase.WhenReady)
+@DependsOn(['DbService', 'CacheService', 'SchedulerService'])
+export class JobManager extends BaseService {
+  private readonly handlers = new Map<string, JobHandler>()
+  private readonly queues = new Map<string, DispatchQueue>()
+  private readonly globalDispatchMutex = new Mutex()
+  private readonly abortControllers = new Map<string, AbortController>()
+  private readonly finishedResolvers = new Map<string, FinishedResolver>()
+  private readonly scheduleDisposables = new Map<string, Disposable>()
+  private readonly globalMaxConcurrency = DEFAULT_GLOBAL_MAX_CONCURRENCY
+
+  // ---------------- Lifecycle ----------------
+
+  protected override onInit(): void {
+    logger.info('JobManager initialized')
+  }
+
+  protected override async onReady(): Promise<void> {
+    const stats = await runStartupRecovery(this.handlers)
+    logger.info('Startup recovery complete', stats)
+
+    // Arm all enabled schedules so SchedulerService can fire them.
+    const schedules = await jobScheduleService.listEnabled()
+    for (const schedule of schedules) {
+      this.armSchedule(schedule)
+    }
+
+    await this.detectAndDispatchOverdue(schedules)
+
+    // GC + delayed-promotion ticks. registerInterval auto-unrefs + clears on stop.
+    this.registerInterval(() => void this.runGC(), GC_INTERVAL_MS)
+    this.registerInterval(async () => {
+      const promoted = await jobService.promoteDelayedDue(Date.now())
+      if (promoted > 0) {
+        logger.debug('Promoted delayed jobs', { count: promoted })
+        this.dispatchAll()
+      }
+    }, DELAYED_PROMOTION_INTERVAL_MS)
+
+    // Kick the queues once — any job reset by recovery should start now.
+    this.dispatchAll()
+    logger.info('JobManager ready', { schedules: schedules.length })
+  }
+
+  protected override async onStop(): Promise<void> {
+    const inFlight = Array.from(this.abortControllers.keys())
+    for (const controller of this.abortControllers.values()) {
+      controller.abort(new Error('JobManager shutdown'))
+    }
+    for (const disposable of this.scheduleDisposables.values()) {
+      disposable.dispose()
+    }
+    this.scheduleDisposables.clear()
+
+    if (inFlight.length === 0) {
+      logger.info('JobManager.onStop: no in-flight jobs')
+    } else {
+      const pendingPromises = inFlight
+        .map((id) => this.finishedResolvers.get(id)?.promise)
+        .filter((p): p is Promise<JobSnapshot> => p !== undefined)
+
+      const timeout = new Promise<'timeout'>((resolve) =>
+        setTimeout(() => resolve('timeout'), Application.SHUTDOWN_TIMEOUT_MS)
+      )
+      const winner = await Promise.race([Promise.allSettled(pendingPromises).then(() => 'done' as const), timeout])
+
+      if (winner === 'timeout') {
+        logger.warn('JobManager.onStop timed out — pending jobs will be recovered on next start', {
+          inFlight: inFlight.length,
+          timeoutMs: Application.SHUTDOWN_TIMEOUT_MS
+        })
+      } else {
+        logger.info('JobManager.onStop: all in-flight jobs settled')
+      }
+    }
+
+    // Critical anti-leak: discard unresolved finished resolvers without
+    // rejecting their promises. Callers awaiting them keep an unsettled
+    // Promise — their responsibility to wrap in a timeout / race.
+    this.finishedResolvers.clear()
+    this.abortControllers.clear()
+  }
+
+  protected override onDestroy(): void {
+    this.handlers.clear()
+    this.queues.clear()
+    this.abortControllers.clear()
+    this.finishedResolvers.clear()
+    this.scheduleDisposables.clear()
+  }
+
+  // ---------------- Handler registry ----------------
+
+  registerHandler<K extends JobType>(type: K, handler: JobHandler<JobPayloadOf<K>>): void {
+    if (this.handlers.has(type)) {
+      throw new Error(`JobManager: handler for type "${type}" is already registered`)
+    }
+    this.handlers.set(type, handler as JobHandler)
+    logger.info('Handler registered', { type, recovery: handler.recovery })
+  }
+
+  hasHandler(type: string): boolean {
+    return this.handlers.has(type)
+  }
+
+  // ---------------- enqueue / cancel / list / get ----------------
+
+  async enqueue<K extends JobType>(type: K, input: JobPayloadOf<K>, opts: EnqueueOptions = {}): Promise<JobHandle> {
+    const handler = this.handlers.get(type)
+    if (!handler) {
+      throw this.makeError(JOB_ERROR_CODES.UNKNOWN_TYPE, `No handler registered for type "${type}"`, {
+        type,
+        knownTypes: Array.from(this.handlers.keys())
+      })
+    }
+
+    const inputJson = JSON.stringify(input ?? null)
+    if (inputJson.length > MAX_INPUT_BYTES) {
+      throw this.makeError(JOB_ERROR_CODES.PAYLOAD_TOO_LARGE, 'Job input payload exceeds 1MB', {
+        type,
+        sizeBytes: inputJson.length
+      })
+    }
+
+    if (opts.idempotencyKey) {
+      const existing = await jobService.findActiveByIdempotencyKey(opts.idempotencyKey)
+      if (existing) {
+        logger.info('idempotencyKey match — returning existing handle', {
+          type,
+          key: opts.idempotencyKey,
+          existingId: existing.id
+        })
+        return this.handleFor(existing)
+      }
+    }
+
+    const queueName = opts.queue ?? handler.defaultQueue?.(input as never) ?? type
+    const concurrency = handler.defaultConcurrency ?? 1
+    this.ensureQueue(queueName, concurrency)
+
+    const now = Date.now()
+    const scheduledAt = opts.scheduledAt ?? now
+    const status = scheduledAt > now ? 'delayed' : 'pending'
+    const maxAttempts = opts.maxAttempts ?? handler.defaultRetryPolicy?.maxAttempts ?? DEFAULT_RETRY_POLICY.maxAttempts
+
+    const insertRow: InsertJobRow = {
+      type,
+      status,
+      priority: opts.priority ?? 0,
+      queue: queueName,
+      idempotencyKey: opts.idempotencyKey ?? null,
+      scheduleId: opts.scheduleId ?? null,
+      scheduledAt,
+      attempt: 0,
+      maxAttempts,
+      input: inputJson,
+      parentId: opts.parentId ?? null,
+      cancelRequested: false,
+      metadata: JSON.stringify(opts.metadata ?? {}),
+      timeoutMs: opts.timeoutMs ?? handler.defaultTimeoutMs ?? null
+    }
+
+    const snapshot = await jobService.create(insertRow)
+    this.publishState(snapshot)
+    const handle = this.handleFor(snapshot)
+
+    if (snapshot.status === 'pending') {
+      void this.dispatch(queueName)
+    } else if (snapshot.status === 'delayed') {
+      this.armDelayedJob(snapshot)
+    }
+
+    logger.info('Job enqueued', {
+      id: snapshot.id,
+      type,
+      queue: queueName,
+      status: snapshot.status,
+      scheduledAt
+    })
+    return handle
+  }
+
+  async cancel(jobId: string, reason?: string): Promise<void> {
+    if (reason !== undefined && reason.length > MAX_CANCEL_REASON_CHARS) {
+      throw this.makeError(JOB_ERROR_CODES.CANCEL_REASON_TOO_LONG, 'Cancel reason exceeds 500 characters', {
+        length: reason.length
+      })
+    }
+
+    const db = application.get('DbService').getDb()
+    await db.transaction(async (tx) => {
+      await jobService.setCancelRequestedTx(tx, jobId)
+    })
+
+    const controller = this.abortControllers.get(jobId)
+    if (controller) {
+      controller.abort(new Error(`Job cancelled${reason ? `: ${reason}` : ''}`))
+      // Wait up to handler.cancelTimeoutMs (default 30s) for the handler to
+      // acknowledge. If the timeout fires the handler is misbehaving — force
+      // the job into 'cancelled' so the dispatch queue slot frees up.
+      const snapshot = await jobService.getById(jobId)
+      const handler = snapshot ? this.handlers.get(snapshot.type) : undefined
+      const graceMs = handler?.cancelTimeoutMs ?? 30_000
+      const finished = this.finishedResolvers.get(jobId)?.promise
+      if (finished) {
+        const timeout = new Promise<'timeout'>((resolve) => setTimeout(() => resolve('timeout'), graceMs))
+        const winner = await Promise.race([finished.then(() => 'done' as const), timeout])
+        if (winner === 'timeout') {
+          logger.warn('cancel timed out — forcing terminal state', { jobId, graceMs })
+          await this.finalizeJob(jobId, 'cancelled', undefined, {
+            code: JOB_ERROR_CODES.CANCELLED,
+            message: `Cancel timed out after ${graceMs}ms${reason ? ` (reason: ${reason})` : ''}`,
+            retryable: false
+          })
+        }
+      }
+    } else {
+      // Not in-flight — pending / delayed → finalize directly as cancelled
+      const snapshot = await jobService.getById(jobId)
+      if (snapshot && (snapshot.status === 'pending' || snapshot.status === 'delayed')) {
+        await this.finalizeJob(jobId, 'cancelled', undefined, {
+          code: JOB_ERROR_CODES.CANCELLED,
+          message: reason ?? 'Cancelled by user',
+          retryable: false
+        })
+      }
+    }
+  }
+
+  /**
+   * Cancel all non-terminal jobs matching the filter. Aborts in-flight
+   * AbortControllers in this process and transitions pending/delayed rows
+   * directly to 'cancelled'. Covers reset() semantics for Phase 4 Knowledge
+   * reset and Phase 3 FileProcessing batch cancellation.
+   *
+   * Empty filter (no queue and no type) is rejected — preventing accidental
+   * "cancel all jobs in the system".
+   *
+   * Returns:
+   *   - `aborted`: in-flight controllers aborted in this process
+   *   - `transitioned`: pending/delayed rows finalized synchronously
+   *
+   * Running jobs settle asynchronously through the normal handler-execute
+   * flow (handler observes signal.aborted) and are NOT counted as transitioned.
+   */
+  async cancelMany(
+    filter: { queue?: string; type?: string },
+    reason?: string
+  ): Promise<{ aborted: number; transitioned: number }> {
+    if (!filter.queue && !filter.type) {
+      throw new Error('cancelMany: filter must specify queue or type (empty filter rejected)')
+    }
+    if (reason !== undefined && reason.length > MAX_CANCEL_REASON_CHARS) {
+      throw this.makeError(JOB_ERROR_CODES.CANCEL_REASON_TOO_LONG, 'Cancel reason exceeds 500 characters', {
+        length: reason.length
+      })
+    }
+    const db = application.get('DbService').getDb()
+    const result = await db.transaction(async (tx) =>
+      jobService.cancelManyTx(tx, filter, {
+        code: JOB_ERROR_CODES.CANCELLED,
+        message: reason ?? 'Cancelled by cancelMany',
+        retryable: false
+      })
+    )
+    let aborted = 0
+    for (const id of result.runningIds) {
+      const controller = this.abortControllers.get(id)
+      if (controller) {
+        controller.abort(new Error(`Job cancelled${reason ? `: ${reason}` : ''}`))
+        aborted++
+      }
+    }
+    return { aborted, transitioned: result.transitioned }
+  }
+
+  async get(jobId: string): Promise<JobSnapshot | null> {
+    return jobService.getById(jobId)
+  }
+
+  async list(filter: Parameters<typeof jobService.list>[0] = {}): Promise<JobSnapshot[]> {
+    return jobService.list(filter)
+  }
+
+  // ---------------- Schedule registry (dual API: type+name / by id) ----------------
+
+  async registerJobSchedule<K extends JobType>(input: JobScheduleRegistrationInput<K>): Promise<{ id: string }> {
+    if (!this.handlers.has(input.type)) {
+      throw this.makeError(JOB_ERROR_CODES.UNKNOWN_TYPE, `No handler for schedule type "${input.type}"`, {
+        type: input.type
+      })
+    }
+    const snapshot = await jobScheduleService.create({
+      type: input.type,
+      name: input.name ?? null,
+      trigger: input.trigger,
+      jobInputTemplate: input.jobInputTemplate,
+      catchUpPolicy: input.catchUpPolicy,
+      metadata: input.metadata,
+      enabled: input.enabled
+    })
+    this.armSchedule(snapshot)
+    logger.info('Schedule registered', { id: snapshot.id, type: input.type, name: snapshot.name })
+    return { id: snapshot.id }
+  }
+
+  async pauseJobScheduleById(id: string): Promise<boolean> {
+    const disp = this.scheduleDisposables.get(id)
+    if (disp) {
+      disp.dispose()
+      this.scheduleDisposables.delete(id)
+    }
+    return jobScheduleService.setEnabled(id, false)
+  }
+
+  async resumeJobScheduleById(id: string): Promise<boolean> {
+    const updated = await jobScheduleService.setEnabled(id, true)
+    if (updated) {
+      const snapshot = await jobScheduleService.getById(id)
+      if (snapshot) this.armSchedule(snapshot)
+    }
+    return updated
+  }
+
+  async triggerJobScheduleNowById(id: string): Promise<boolean> {
+    const schedule = await jobScheduleService.getById(id)
+    if (!schedule) return false
+    // Try Scheduler's native trigger first (works for cron triggers).
+    const triggered = await application.get('SchedulerService').triggerNow(`schedule:${id}`)
+    if (triggered) return true
+    // For interval/once schedules, enqueue directly using the schedule's template.
+    await this.enqueue(schedule.type as JobType, schedule.jobInputTemplate as never, {
+      scheduleId: schedule.id
+    })
+    return true
+  }
+
+  async unregisterJobScheduleById(id: string): Promise<boolean> {
+    const disp = this.scheduleDisposables.get(id)
+    if (disp) {
+      disp.dispose()
+      this.scheduleDisposables.delete(id)
+    }
+    return jobScheduleService.delete(id)
+  }
+
+  async getJobScheduleById(id: string): Promise<JobScheduleSnapshot | null> {
+    return jobScheduleService.getById(id)
+  }
+
+  async listJobSchedules(
+    filter: Parameters<typeof jobScheduleService.listAll>[0] = {}
+  ): Promise<JobScheduleSnapshot[]> {
+    return jobScheduleService.listAll(filter)
+  }
+
+  // By-name flavor — internal resolves to by-id.
+
+  async pauseJobSchedule<K extends JobType>(type: K, name?: string | null): Promise<boolean> {
+    const id = await this.resolveScheduleIdByName(type, name)
+    return this.pauseJobScheduleById(id)
+  }
+
+  async resumeJobSchedule<K extends JobType>(type: K, name?: string | null): Promise<boolean> {
+    const id = await this.resolveScheduleIdByName(type, name)
+    return this.resumeJobScheduleById(id)
+  }
+
+  async triggerJobScheduleNow<K extends JobType>(type: K, name?: string | null): Promise<boolean> {
+    const id = await this.resolveScheduleIdByName(type, name)
+    return this.triggerJobScheduleNowById(id)
+  }
+
+  async unregisterJobSchedule<K extends JobType>(type: K, name?: string | null): Promise<boolean> {
+    const id = await this.resolveScheduleIdByName(type, name)
+    return this.unregisterJobScheduleById(id)
+  }
+
+  async getJobSchedule<K extends JobType>(type: K, name?: string | null): Promise<JobScheduleSnapshot | null> {
+    const id = await this.resolveScheduleIdByName(type, name).catch((err) => {
+      // For "name required" errors, surface them. For "not found", return null.
+      if (err instanceof Error && err.message.includes(JOB_ERROR_CODES.SCHEDULE_NOT_FOUND_BY_NAME)) return null
+      throw err
+    })
+    if (typeof id !== 'string') return null
+    return jobScheduleService.getById(id)
+  }
+
+  private async resolveScheduleIdByName(type: string, name?: string | null): Promise<string> {
+    if (name == null) {
+      const candidates = await jobScheduleService.listAll({ type })
+      if (candidates.length > 1) {
+        throw this.makeError(
+          JOB_ERROR_CODES.SCHEDULE_NAME_REQUIRED,
+          `Type "${type}" has multiple schedules — name required`,
+          { type, knownNames: candidates.map((c) => c.name) }
+        )
+      }
+      if (candidates.length === 1) return candidates[0].id
+    }
+    const snapshot = await jobScheduleService.getByTypeAndName(type, name)
+    if (!snapshot) {
+      const knownNames = await jobScheduleService.listNamesForType(type)
+      throw this.makeError(JOB_ERROR_CODES.SCHEDULE_NOT_FOUND_BY_NAME, `Schedule not found for type "${type}"`, {
+        type,
+        name: name ?? null,
+        knownNames
+      })
+    }
+    return snapshot.id
+  }
+
+  // ---------------- Dispatch + execute ----------------
+
+  /**
+   * Get or create a DispatchQueue.
+   *
+   * Concurrency is set at first creation and NOT updated on subsequent calls
+   * with the same queueName. If the same queueName is reached via different
+   * handlers with different `defaultConcurrency`, the first enqueue wins.
+   * Phase 1 follows the plan convention "one type ↔ one queue ↔ one
+   * concurrency", so this is unobservable in practice. Documented for the
+   * future-proof reader.
+   */
+  private ensureQueue(name: string, concurrency: number): DispatchQueue {
+    let queue = this.queues.get(name)
+    if (!queue) {
+      queue = new DispatchQueue(name, concurrency)
+      this.queues.set(name, queue)
+    }
+    return queue
+  }
+
+  /**
+   * Try to claim one pending job in `queueName` and spawn its handler. Releases
+   * both mutex layers before invoking the handler. Schedules a microtask
+   * recursion to fill the next slot if one was claimed.
+   */
+  async dispatch(queueName: string): Promise<void> {
+    const queue = this.queues.get(queueName)
+    if (!queue) return
+
+    const releaseQueue = await queue.mutex.acquire()
+    const releaseGlobal = await this.globalDispatchMutex.acquire()
+    let claimed: JobRow | null = null
+
+    try {
+      const db = application.get('DbService').getDb()
+      claimed = await db.transaction(async (tx) => {
+        const queueActive = await jobService.countActiveByQueueTx(tx, queueName)
+        if (queueActive >= queue.concurrency) return null
+        const globalActive = await jobService.countActiveGlobalTx(tx)
+        if (globalActive >= this.globalMaxConcurrency) return null
+        return jobService.claimNextPendingTx(tx, queueName)
+      })
+    } catch (err) {
+      logger.error('dispatch transaction failed', { queue: queueName, error: err })
+    } finally {
+      releaseGlobal()
+      releaseQueue()
+    }
+
+    if (!claimed) return
+
+    // Spawn handler outside of the mutex.
+    this.spawnExecute(claimed)
+    queueMicrotask(() => void this.dispatch(queueName))
+  }
+
+  private dispatchAll(): void {
+    for (const name of this.queues.keys()) {
+      void this.dispatch(name)
+    }
+  }
+
+  /**
+   * Build context, spawn handler.execute, transition state on terminal or
+   * schedule retry on retryable failure. Errors thrown synchronously by
+   * handler before its first await are caught inside the same task.
+   */
+  private spawnExecute(row: JobRow): void {
+    const handler = this.handlers.get(row.type)
+    if (!handler) {
+      logger.error('spawnExecute: missing handler — finalizing as failed', { type: row.type, id: row.id })
+      void this.finalizeJob(row.id, 'failed', undefined, {
+        code: JOB_ERROR_CODES.UNKNOWN_TYPE,
+        message: `No handler registered for type "${row.type}"`,
+        retryable: false
+      })
+      return
+    }
+
+    const controller = new AbortController()
+    this.abortControllers.set(row.id, controller)
+
+    let timeoutHandle: ReturnType<typeof setTimeout> | undefined
+    if (row.timeoutMs && row.timeoutMs > 0) {
+      timeoutHandle = setTimeout(() => {
+        controller.abort(new Error('JobHandlerTimeout'))
+      }, row.timeoutMs)
+      timeoutHandle.unref?.()
+    }
+
+    const ctx: JobContext = {
+      jobId: row.id,
+      input: this.parseJson(row.input, undefined),
+      attempt: row.attempt,
+      signal: controller.signal,
+      metadata: (this.parseJson(row.metadata, {}) ?? {}) as Record<string, unknown>,
+      patchMetadata: async (patch) => {
+        const current = (this.parseJson(row.metadata, {}) ?? {}) as Record<string, unknown>
+        const merged = { ...current, ...patch }
+        row.metadata = JSON.stringify(merged)
+        const db = application.get('DbService').getDb()
+        await db.transaction(async (tx) => {
+          await jobService.setMetadataTx(tx, row.id, row.metadata)
+        })
+      },
+      reportProgress: (progress, detail) => {
+        application.get('CacheService').setShared(`${JOB_PROGRESS_KEY_PREFIX}${row.id}`, { progress, detail }, 60_000)
+      },
+      logger: loggerService.withContext('JobExec', { jobId: row.id, type: row.type })
+    }
+
+    void (async () => {
+      try {
+        const output = await handler.execute(ctx)
+        if (timeoutHandle) clearTimeout(timeoutHandle)
+        await this.finalizeJob(row.id, 'completed', output, null)
+      } catch (err) {
+        if (timeoutHandle) clearTimeout(timeoutHandle)
+        const isAbort = this.isAbortError(err)
+        const error: JobError = isAbort
+          ? {
+              code: JOB_ERROR_CODES.CANCELLED,
+              message: (err as Error).message || 'Cancelled',
+              retryable: false
+            }
+          : {
+              code: this.isTimeoutError(err) ? JOB_ERROR_CODES.HANDLER_TIMEOUT : JOB_ERROR_CODES.HANDLER_THREW,
+              message: (err as Error).message || String(err),
+              retryable: true
+            }
+
+        const retryPolicy = handler.defaultRetryPolicy ?? DEFAULT_RETRY_POLICY
+        const canRetry = !isAbort && error.retryable && row.attempt + 1 < row.maxAttempts
+
+        if (canRetry) {
+          const backoffMs = this.computeBackoff(retryPolicy, row.attempt + 1)
+          const scheduledAt = Date.now() + backoffMs
+          await this.scheduleRetry(row.id, row.attempt + 1, scheduledAt, error, row.queue)
+        } else {
+          await this.finalizeJob(row.id, isAbort ? 'cancelled' : 'failed', undefined, error)
+        }
+      } finally {
+        this.abortControllers.delete(row.id)
+      }
+    })()
+  }
+
+  private async finalizeJob(
+    jobId: string,
+    status: 'completed' | 'failed' | 'cancelled',
+    output: unknown | undefined,
+    error: JobError | null
+  ): Promise<void> {
+    const db = application.get('DbService').getDb()
+    try {
+      await db.transaction(async (tx) => {
+        await jobService.setTerminalTx(tx, jobId, status, output, error)
+      })
+    } catch (err) {
+      logger.error('finalizeJob: tx failed — job may stay running until next recovery', { jobId, status, err })
+      return
+    }
+
+    const snapshot = await jobService.getById(jobId)
+    if (!snapshot) return
+
+    this.publishState(snapshot)
+
+    const resolver = this.finishedResolvers.get(jobId)
+    if (resolver) {
+      resolver.resolve(snapshot)
+      this.finishedResolvers.delete(jobId)
+    }
+
+    const handler = this.handlers.get(snapshot.type)
+    if (handler?.onSettled) {
+      try {
+        await handler.onSettled({
+          jobId,
+          type: snapshot.type,
+          scheduleId: snapshot.scheduleId,
+          status,
+          output: snapshot.output,
+          error: snapshot.error,
+          attempt: snapshot.attempt
+        })
+      } catch (settledErr) {
+        logger.warn('handler.onSettled threw — ignoring', {
+          jobId,
+          err: (settledErr as Error).message
+        })
+      }
+    }
+
+    // A slot just freed — dispatch in case another job is waiting.
+    void this.dispatch(snapshot.queue)
+  }
+
+  private async scheduleRetry(
+    jobId: string,
+    nextAttempt: number,
+    scheduledAt: number,
+    error: JobError,
+    queue: string
+  ): Promise<void> {
+    const db = application.get('DbService').getDb()
+    await db.transaction(async (tx) => {
+      await jobService.setDelayedRetryTx(tx, jobId, nextAttempt, scheduledAt, error)
+    })
+
+    const scheduler = application.get('SchedulerService')
+    const retryId = `retry:${jobId}:${nextAttempt}`
+    scheduler.registerSchedule(retryId, { kind: 'once', at: scheduledAt }, async () => {
+      await jobService.promoteDelayedDue(Date.now())
+      void this.dispatch(queue)
+    })
+    logger.info('Retry scheduled', { jobId, nextAttempt, scheduledAt, queue })
+  }
+
+  // ---------------- Schedule arming + catch-up ----------------
+
+  private armSchedule(schedule: JobScheduleSnapshot): void {
+    if (!schedule.enabled) return
+    if (this.scheduleDisposables.has(schedule.id)) {
+      // Replace existing disposable (e.g. on edit / re-enable)
+      this.scheduleDisposables.get(schedule.id)?.dispose()
+      this.scheduleDisposables.delete(schedule.id)
+    }
+
+    const scheduler = application.get('SchedulerService')
+    const scheduleKey = `schedule:${schedule.id}`
+    const trigger: Trigger = schedule.trigger
+    const disp = scheduler.registerSchedule(scheduleKey, trigger, async () => {
+      try {
+        await this.enqueue(schedule.type as JobType, schedule.jobInputTemplate as never, {
+          scheduleId: schedule.id
+        })
+        const nextRun = scheduler.getNextRun(scheduleKey)
+        await jobScheduleService.markFired(schedule.id, Date.now(), nextRun?.getTime() ?? null)
+      } catch (err) {
+        logger.error('Schedule fire failed', { scheduleId: schedule.id, err: (err as Error).message })
+      }
+    })
+    this.scheduleDisposables.set(schedule.id, disp)
+  }
+
+  /**
+   * Schedule a `once` registration in the Scheduler so the delayed job gets
+   * promoted + dispatched at its scheduledAt time. For `pending` jobs this
+   * fast-path is skipped — the normal dispatch loop handles them.
+   */
+  private armDelayedJob(snapshot: JobSnapshot): void {
+    const scheduler = application.get('SchedulerService')
+    const jobKey = `job:${snapshot.id}`
+    const scheduledMs = Date.parse(snapshot.scheduledAt)
+    scheduler.registerSchedule(jobKey, { kind: 'once', at: scheduledMs }, async () => {
+      await jobService.promoteDelayedDue(Date.now())
+      void this.dispatch(snapshot.queue)
+    })
+  }
+
+  private async detectAndDispatchOverdue(schedules: JobScheduleSnapshot[]): Promise<void> {
+    const nowMs = Date.now()
+    for (const schedule of schedules) {
+      const handler = this.handlers.get(schedule.type)
+      if (!handler) continue
+      const action = computeCatchUpAction(schedule, handler, nowMs)
+      if (action.missEvent && handler.onMissed) {
+        try {
+          await handler.onMissed(action.missEvent)
+        } catch (err) {
+          logger.warn('handler.onMissed threw — ignoring', {
+            scheduleId: schedule.id,
+            err: (err as Error).message
+          })
+        }
+      }
+      if (action.shouldEnqueue) {
+        const scheduledAt = nowMs + action.enqueueDelayMs
+        await this.enqueue(schedule.type as JobType, schedule.jobInputTemplate as never, {
+          scheduleId: schedule.id,
+          scheduledAt
+        })
+        logger.info('Catch-up enqueued', { scheduleId: schedule.id, type: schedule.type, scheduledAt })
+      }
+    }
+  }
+
+  // ---------------- GC ----------------
+
+  private async runGC(): Promise<void> {
+    const cutoff = Date.now() - GC_TERMINAL_TTL_MS
+    const byTtl = await jobService.pruneTerminalOlderThan(cutoff)
+    const byCount = await jobService.pruneTerminalKeepLatestPerType(GC_KEEP_PER_TYPE)
+    if (byTtl + byCount > 0) {
+      logger.info('GC pass', { byTtl, byCount })
+    }
+  }
+
+  // ---------------- Helpers ----------------
+
+  private handleFor(snapshot: JobSnapshot): JobHandle {
+    const existing = this.finishedResolvers.get(snapshot.id)
+    if (existing) {
+      return { id: snapshot.id, snapshot, finished: existing.promise }
+    }
+    if (this.isTerminal(snapshot.status)) {
+      return { id: snapshot.id, snapshot, finished: Promise.resolve(snapshot) }
+    }
+    let resolve!: (s: JobSnapshot) => void
+    const promise = new Promise<JobSnapshot>((res) => {
+      resolve = res
+    })
+    this.finishedResolvers.set(snapshot.id, { resolve, promise })
+    return { id: snapshot.id, snapshot, finished: promise }
+  }
+
+  private isTerminal(status: JobSnapshot['status']): boolean {
+    return status === 'completed' || status === 'failed' || status === 'cancelled'
+  }
+
+  private publishState(snapshot: JobSnapshot): void {
+    application.get('CacheService').setShared(`${JOB_STATE_KEY_PREFIX}${snapshot.id}`, snapshot, 60_000)
+  }
+
+  private computeBackoff(policy: RetryPolicy, attempt: number): number {
+    if (policy.backoff === 'none') return 0
+    if (policy.backoff === 'fixed') return Math.min(policy.baseDelayMs, policy.maxDelayMs)
+    // exponential: base * 2^(attempt-1)
+    const exp = policy.baseDelayMs * Math.pow(2, Math.max(0, attempt - 1))
+    return Math.min(exp, policy.maxDelayMs)
+  }
+
+  private parseJson(raw: string | null, fallback: unknown): unknown {
+    if (raw == null) return fallback
+    try {
+      return JSON.parse(raw)
+    } catch {
+      return fallback
+    }
+  }
+
+  private isAbortError(err: unknown): boolean {
+    return err instanceof Error && (err.name === 'AbortError' || err.message.toLowerCase().includes('abort'))
+  }
+
+  private isTimeoutError(err: unknown): boolean {
+    return err instanceof Error && err.message.includes('Timeout')
+  }
+
+  private makeError(code: string, message: string, params?: Record<string, unknown>): Error {
+    const err = new Error(`${code}: ${message}`)
+    ;(err as Error & { code: string; params?: Record<string, unknown>; retryable: boolean }).code = code
+    ;(err as Error & { code: string; params?: Record<string, unknown>; retryable: boolean }).params = params
+    ;(err as Error & { code: string; params?: Record<string, unknown>; retryable: boolean }).retryable = false
+    return err
+  }
+}
--- a/src/main/core/job/jobRegistry.ts
+++ b/src/main/core/job/jobRegistry.ts
@@ -0,0 +1,33 @@
+/**
+ * Compile-time type catalog for Job payload binding.
+ *
+ * Business modules extend this interface via TypeScript declaration merging to
+ * register their type → payload mapping. The RUNTIME handler Map is stored
+ * inside JobManager and is unrelated to this interface — do not confuse them.
+ *
+ * Example (in a business module that handles 'foo.bar' type jobs):
+ *
+ *   declare module '@main/core/job/jobRegistry' {
+ *     interface JobRegistry {
+ *       'foo.bar': { itemId: string; threshold: number }
+ *     }
+ *   }
+ *
+ * After this declaration:
+ *   - `jobManager.enqueue('foo.bar', { itemId, threshold })` is type-checked
+ *   - Passing `{ wrong: 'shape' }` is a compile error
+ *   - Renaming 'foo.bar' surfaces every call site via the type system
+ *
+ * Path-alias note: the `declare module` path MUST match the tsconfig paths
+ * alias exactly — project uses `@main/*` → `./src/main/*`, so this file is
+ * `@main/core/job/jobRegistry`. Mismatches make declaration merging silently
+ * no-op (no compile error, just lost type binding).
+ */
+
+export interface JobRegistry {}
+
+/** All type strings registered in JobRegistry (compile-time enumeration). */
+export type JobType = keyof JobRegistry & string
+
+/** Payload type for a given job type. */
+export type JobPayloadOf<K extends JobType> = JobRegistry[K]
--- a/src/main/core/job/runtime/DispatchQueue.ts
+++ b/src/main/core/job/runtime/DispatchQueue.ts
@@ -0,0 +1,20 @@
+import { Mutex } from 'async-mutex'
+
+/**
+ * Thin per-queue state holder. The dispatch loop itself lives on JobManager —
+ * DispatchQueue just owns the per-queue concurrency cap and the Layer 1
+ * mutex that serializes the (count → fetch → claim) section for this queue.
+ *
+ * No waiting list, no active job map. All job state lives in jobTable; the
+ * dispatch loop re-queries on every tick to find work. The total memory
+ * footprint per queue is one Map entry + one Mutex (~hundreds of bytes),
+ * which keeps the 1000+ knowledge-base scenario tractable.
+ */
+export class DispatchQueue {
+  readonly mutex = new Mutex()
+
+  constructor(
+    readonly name: string,
+    readonly concurrency: number
+  ) {}
+}
--- a/src/main/core/job/runtime/catchUp.ts
+++ b/src/main/core/job/runtime/catchUp.ts
@@ -0,0 +1,68 @@
+import { loggerService } from '@logger'
+import type { JobScheduleSnapshot } from '@shared/data/api/schemas/jobs'
+
+import type { JobHandler, JobMissEvent } from '../types'
+
+const logger = loggerService.withContext('JobCatchUp')
+
+/**
+ * What the caller (JobManager) should do for this schedule.
+ *
+ * `shouldEnqueue=false + missEvent=null` means the schedule is fine — no
+ * action needed. `shouldEnqueue=false + missEvent != null` means notify the
+ * handler that a fire was missed but do not enqueue (skip-missed policy).
+ */
+export interface CatchUpAction {
+  scheduleId: string
+  type: string
+  shouldEnqueue: boolean
+  /** Delay before enqueuing the make-up job. Used by `after-startup`. */
+  enqueueDelayMs: number
+  missEvent: JobMissEvent | null
+}
+
+/**
+ * Compute the catch-up action for one schedule given current time and the
+ * handler registered for its type. Pure function — JobManager owns the
+ * iteration and the actual side-effect dispatch.
+ *
+ * "Overdue" is decided by `schedule.nextRun <= now` — JobManager / Scheduler
+ * keep `nextRun` updated. If `nextRun` is null (schedule was never armed or
+ * one-shot already fired), the schedule is considered not overdue.
+ *
+ * Phase 1 catch-up policies:
+ *   - skip-missed   : do NOT enqueue, but still emit missEvent (observability).
+ *   - after-startup : enqueue once after `minutes * 60_000` ms; emit missEvent.
+ *
+ * `after-idle` is deferred (requires PowerMonitor.getSystemIdleTime API).
+ */
+export function computeCatchUpAction(schedule: JobScheduleSnapshot, handler: JobHandler, nowMs: number): CatchUpAction {
+  const lastRunMs = schedule.lastRun ? Date.parse(schedule.lastRun) : null
+  const nextRunMs = schedule.nextRun ? Date.parse(schedule.nextRun) : null
+  const isOverdue = nextRunMs !== null && nextRunMs <= nowMs
+
+  if (!isOverdue) {
+    return { scheduleId: schedule.id, type: schedule.type, shouldEnqueue: false, enqueueDelayMs: 0, missEvent: null }
+  }
+
+  // Always build miss event when overdue + handler defines onMissed.
+  // skip-missed still wants the event for observability / breaker logic.
+  const missEvent: JobMissEvent | null = handler.onMissed
+    ? {
+        scheduleId: schedule.id,
+        type: schedule.type,
+        missedCount: 1,
+        lastFireAt: lastRunMs
+      }
+    : null
+
+  if (schedule.catchUpPolicy.kind === 'skip-missed') {
+    logger.debug('Catch-up: skip-missed', { scheduleId: schedule.id, type: schedule.type })
+    return { scheduleId: schedule.id, type: schedule.type, shouldEnqueue: false, enqueueDelayMs: 0, missEvent }
+  }
+
+  // after-startup
+  const delayMs = schedule.catchUpPolicy.minutes * 60_000
+  logger.debug('Catch-up: after-startup', { scheduleId: schedule.id, type: schedule.type, delayMs })
+  return { scheduleId: schedule.id, type: schedule.type, shouldEnqueue: true, enqueueDelayMs: delayMs, missEvent }
+}
--- a/src/main/core/job/runtime/recovery.ts
+++ b/src/main/core/job/runtime/recovery.ts
@@ -0,0 +1,109 @@
+import { jobService } from '@data/services/JobService'
+import { loggerService } from '@logger'
+import { JOB_ERROR_CODES, type JobError } from '@shared/data/api/schemas/jobs'
+
+import type { JobHandler } from '../types'
+
+const logger = loggerService.withContext('JobRecovery')
+
+export interface RecoveryStats {
+  cancelled: number
+  pendingReset: number
+  delayedKept: number
+  singletonKept: number
+}
+
+/**
+ * Per-processor declarative recovery. Walks the registered handlers, applies
+ * each type's `recovery` strategy to its non-terminal jobs, then handles
+ * orphan running jobs whose handler is no longer registered.
+ *
+ * Step order matters: cancelRequested=true overrides any strategy — those
+ * jobs are cancelled regardless. The strategy then applies to the rest.
+ */
+export async function runStartupRecovery(handlers: ReadonlyMap<string, JobHandler>): Promise<RecoveryStats> {
+  const stats: RecoveryStats = { cancelled: 0, pendingReset: 0, delayedKept: 0, singletonKept: 0 }
+  const cancelledByRecovery: JobError = {
+    code: JOB_ERROR_CODES.CANCELLED,
+    message: 'Cancelled by startup recovery',
+    retryable: false
+  }
+
+  for (const [type, handler] of handlers) {
+    const nonTerminal = await jobService.getNonTerminalByType(type)
+    if (nonTerminal.length === 0) continue
+
+    // 1. cancelRequested → cancelled, regardless of strategy.
+    const cancelRequestedIds = nonTerminal
+      .filter((r) => r.cancelRequested && (r.status === 'running' || r.status === 'delayed'))
+      .map((r) => r.id)
+    if (cancelRequestedIds.length) {
+      await jobService.cancelByIds(cancelRequestedIds, cancelledByRecovery)
+      stats.cancelled += cancelRequestedIds.length
+      logger.info('Cancelled jobs with cancelRequested=true', { type, count: cancelRequestedIds.length })
+    }
+
+    const cancelRequestedSet = new Set(cancelRequestedIds)
+    const remaining = nonTerminal.filter((r) => !cancelRequestedSet.has(r.id))
+
+    // 2. Apply strategy to the rest.
+    if (handler.recovery === 'abandon') {
+      const ids = remaining.map((r) => r.id)
+      if (ids.length) {
+        await jobService.cancelByIds(ids, cancelledByRecovery)
+        stats.cancelled += ids.length
+        logger.info('Abandon recovery: cancelled non-terminal', { type, count: ids.length })
+      }
+    } else if (handler.recovery === 'retry') {
+      const runningIds = remaining.filter((r) => r.status === 'running').map((r) => r.id)
+      const delayedCount = remaining.filter((r) => r.status === 'delayed').length
+      if (runningIds.length) {
+        await jobService.resetToPendingByIds(runningIds)
+        stats.pendingReset += runningIds.length
+      }
+      stats.delayedKept += delayedCount
+      if (runningIds.length || delayedCount) {
+        logger.info('Retry recovery', { type, reset: runningIds.length, delayedKept: delayedCount })
+      }
+    } else if (handler.recovery === 'singleton') {
+      if (remaining.length === 0) continue
+      // remaining is already sorted by createdAt DESC (jobService contract).
+      const keep = remaining[0]
+      if (keep.status === 'running') {
+        await jobService.resetToPendingByIds([keep.id])
+        stats.pendingReset += 1
+      }
+      stats.singletonKept += 1
+      const toCancelIds = remaining.slice(1).map((r) => r.id)
+      if (toCancelIds.length) {
+        await jobService.cancelByIds(toCancelIds, cancelledByRecovery)
+        stats.cancelled += toCancelIds.length
+      }
+      logger.info('Singleton recovery', {
+        type,
+        keptId: keep.id,
+        keptStatus: keep.status,
+        cancelled: toCancelIds.length
+      })
+    }
+  }
+
+  // 3. Orphan running jobs — handler no longer registered. Cancel them so they
+  // do not hang in 'running' forever (would block dispatch via active count).
+  const allRunning = await jobService.getStaleRunning()
+  const orphanIds = allRunning.filter((r) => !handlers.has(r.type)).map((r) => r.id)
+  if (orphanIds.length) {
+    await jobService.cancelByIds(orphanIds, {
+      code: JOB_ERROR_CODES.CANCELLED,
+      message: 'Orphan job: handler no longer registered',
+      retryable: false
+    })
+    stats.cancelled += orphanIds.length
+    logger.warn('Cancelled orphan running jobs (no handler registered)', {
+      count: orphanIds.length,
+      types: Array.from(new Set(allRunning.filter((r) => !handlers.has(r.type)).map((r) => r.type)))
+    })
+  }
+
+  return stats
+}
--- a/src/main/core/job/types.ts
+++ b/src/main/core/job/types.ts
@@ -0,0 +1,142 @@
+import type { LoggerService } from '@main/core/logger/LoggerService'
+import type {
+  CatchUpPolicy,
+  JobError,
+  JobSnapshot,
+  JobStatus,
+  RetryPolicy,
+  Trigger
+} from '@shared/data/api/schemas/jobs'
+
+import type { JobPayloadOf, JobType } from './jobRegistry'
+
+/**
+ * Startup recovery strategy declared at handler registration. Determines how
+ * JobManager treats this type's non-terminal jobs after process restart.
+ *
+ *   - 'abandon'   : all non-terminal jobs → cancelled
+ *   - 'retry'     : running → pending (attempt unchanged); delayed stays as-is
+ *   - 'singleton' : keep the newest non-terminal; cancel the rest
+ */
+export type RecoveryStrategy = 'abandon' | 'retry' | 'singleton'
+
+/**
+ * Per-job context passed to handler.execute. Provides cancel signal, mutable
+ * metadata, progress reporting, and a scoped logger.
+ */
+export interface JobContext<TPayload = unknown> {
+  jobId: string
+  input: TPayload
+  attempt: number
+  signal: AbortSignal
+  /** Read-only view of jobTable.metadata at the start of this attempt. */
+  metadata: Readonly<Record<string, unknown>>
+  /**
+   * Shallow-merge a metadata patch and persist immediately. Useful for
+   * cross-restart hand-offs (e.g. remote-poll providerTaskId — see
+   * handler-authoring.md). Awaiting this call is required before the value is
+   * guaranteed durable.
+   */
+  patchMetadata(patch: Record<string, unknown>): Promise<void>
+  /**
+   * Report progress (0-100) with optional free-shape detail. Reaches the
+   * renderer via CacheService.subscribeSharedChange at jobs.progress.${jobId}.
+   */
+  reportProgress(progress: number, detail?: unknown): void
+  /** Logger pre-bound to { jobId, type }. */
+  logger: LoggerService
+}
+
+export interface JobMissEvent {
+  scheduleId: string
+  type: string
+  /** Number of fires that were missed during the down period. */
+  missedCount: number
+  /** Last successful fire time (ms epoch), or null if never run before. */
+  lastFireAt: number | null
+}
+
+export interface JobSettledEvent {
+  jobId: string
+  type: string
+  scheduleId: string | null
+  status: Extract<JobStatus, 'completed' | 'failed' | 'cancelled'>
+  output?: unknown
+  error: JobError | null
+  attempt: number
+}
+
+export interface JobHandler<TPayload = unknown> {
+  /** Required: startup recovery behavior for non-terminal jobs of this type. */
+  recovery: RecoveryStrategy
+  /**
+   * Default queue resolver. When enqueue does not pass `opts.queue` the handler
+   * may pick one based on input (e.g. `base.${input.baseId}` for per-base
+   * serialization). Defaults to the type string itself.
+   */
+  defaultQueue?: (input: TPayload) => string
+  /**
+   * Per-type concurrency cap. Applies to the type's default queue and any
+   * queue resolved by defaultQueue. Defaults to 1.
+   */
+  defaultConcurrency?: number
+  /** Default retry policy. */
+  defaultRetryPolicy?: RetryPolicy
+  /** Default per-job timeout (ms). */
+  defaultTimeoutMs?: number
+  /** Grace period to wait for handler to react to AbortSignal after cancel. Defaults to 30000ms. */
+  cancelTimeoutMs?: number
+  /** Execute one job attempt. Throw to fail; reject with AbortError to cancel. */
+  execute(ctx: JobContext<TPayload>): Promise<unknown>
+  /** Optional. Called when a schedule fire was missed. */
+  onMissed?(event: JobMissEvent): void | Promise<void>
+  /** Optional. Called when the job reaches a terminal state. Errors are caught + logged. */
+  onSettled?(event: JobSettledEvent): void | Promise<void>
+}
+
+/** Strongly-typed handler for a registered job type. */
+export type JobHandlerFor<K extends JobType> = JobHandler<JobPayloadOf<K>>
+
+/** Result of `jobManager.enqueue` — id + initial snapshot + terminal promise. */
+export interface JobHandle {
+  id: string
+  snapshot: JobSnapshot
+  /**
+   * Resolves with the terminal JobSnapshot. Never rejects under normal
+   * operation — cancellation / failure surface through `status` on the
+   * resolved snapshot. JobManager.onDestroy abandons unresolved Promises
+   * (does not reject) — callers that await across shutdown should race with
+   * an external timeout.
+   */
+  finished: Promise<JobSnapshot>
+}
+
+export interface EnqueueOptions {
+  queue?: string
+  priority?: number
+  idempotencyKey?: string
+  /** ms epoch — when to first attempt (default: now). */
+  scheduledAt?: number
+  /** When this enqueue is the consequence of a schedule fire. */
+  scheduleId?: string
+  parentId?: string
+  timeoutMs?: number
+  maxAttempts?: number
+  metadata?: Record<string, unknown>
+}
+
+export interface JobScheduleRegistrationInput<K extends JobType = JobType> {
+  type: K
+  trigger: Trigger
+  jobInputTemplate: JobPayloadOf<K>
+  catchUpPolicy: CatchUpPolicy
+  /** Required for multi-instance types; omit (or null) for single-instance types. */
+  name?: string | null
+  metadata?: Record<string, unknown>
+  /** Default true. */
+  enabled?: boolean
+}
+
+/** Cache key template constants (registered in cacheSchemas.ts). */
+export const JOB_STATE_KEY_PREFIX = 'jobs.state.' as const
+export const JOB_PROGRESS_KEY_PREFIX = 'jobs.progress.' as const
--- a/src/main/core/scheduler/SchedulerService.ts
+++ b/src/main/core/scheduler/SchedulerService.ts
@@ -0,0 +1,213 @@
+import { loggerService } from '@logger'
+import { BaseService, Injectable, Phase, ServicePhase } from '@main/core/lifecycle'
+import type { Disposable } from '@main/core/lifecycle/event'
+import type { Trigger } from '@shared/data/api/schemas/jobs'
+import { Cron } from 'croner'
+
+const logger = loggerService.withContext('SchedulerService')
+
+export type ScheduleCallback = () => void | Promise<void>
+
+interface IntervalEntry {
+  handle: ReturnType<typeof setTimeout>
+  ms: number
+  callback: ScheduleCallback
+}
+
+/**
+ * General-purpose stateless scheduler. Knows about "when" to fire a callback,
+ * nothing about what the callback does. JobManager is its primary consumer but
+ * any business module can use it directly for simple cron/interval/once needs
+ * — see plan section "强制原则: SchedulerService 是项目内唯一的通用调度器".
+ *
+ * Internals:
+ *   - `cron` and `once` Triggers are backed by croner instances (pause/resume,
+ *     timezone via Intl API, .trigger() for manual fire).
+ *   - `interval` Trigger uses a chained setTimeout (more flexible than
+ *     setInterval — handles slow callbacks without overlap and unrefs the loop).
+ *   - No persistence. JobManager re-registers all schedules on startup.
+ */
+@Injectable('SchedulerService')
+@ServicePhase(Phase.WhenReady)
+export class SchedulerService extends BaseService {
+  private cronJobs = new Map<string, Cron>()
+  private intervalHandles = new Map<string, IntervalEntry>()
+
+  protected override onInit(): void {
+    logger.info('SchedulerService initialized')
+  }
+
+  protected override onStop(): void {
+    this.clearAll()
+  }
+
+  protected override onDestroy(): void {
+    this.clearAll()
+  }
+
+  /**
+   * Register a callback to fire on schedule. Returns a Disposable that
+   * unregisters when disposed. Calling registerSchedule twice with the same
+   * id replaces the previous registration.
+   */
+  registerSchedule(id: string, trigger: Trigger, callback: ScheduleCallback): Disposable {
+    if (this.has(id)) this.unregister(id)
+
+    if (trigger.kind === 'cron') {
+      this.scheduleCron(id, trigger, callback)
+    } else if (trigger.kind === 'once') {
+      this.scheduleOnce(id, trigger.at, callback)
+    } else {
+      this.scheduleInterval(id, trigger.ms, callback)
+    }
+
+    logger.debug('Scheduled', { id, kind: trigger.kind })
+    return this.registerDisposable(() => this.unregister(id))
+  }
+
+  /**
+   * Pause a cron schedule. No-op (with warn log) for interval / once: those use
+   * chained setTimeout and cannot be paused — to "pause" an interval, callers
+   * should unregister and re-register when ready.
+   */
+  pause(id: string): void {
+    const cron = this.cronJobs.get(id)
+    if (cron) {
+      cron.pause()
+      logger.debug('Paused cron', { id })
+      return
+    }
+    if (this.intervalHandles.has(id)) {
+      logger.warn('pause is a no-op for interval/once schedules — use unregister + re-register to "pause" instead', {
+        id
+      })
+      return
+    }
+    logger.warn('pause called on unknown schedule id', { id })
+  }
+
+  resume(id: string): void {
+    const cron = this.cronJobs.get(id)
+    if (cron) {
+      cron.resume()
+      logger.debug('Resumed cron', { id })
+      return
+    }
+    if (this.intervalHandles.has(id)) {
+      logger.warn('resume is a no-op for interval/once schedules', { id })
+      return
+    }
+    logger.warn('resume called on unknown schedule id', { id })
+  }
+
+  unregister(id: string): void {
+    const cron = this.cronJobs.get(id)
+    if (cron) {
+      cron.stop()
+      this.cronJobs.delete(id)
+      logger.debug('Unregistered cron', { id })
+      return
+    }
+    const interval = this.intervalHandles.get(id)
+    if (interval) {
+      clearTimeout(interval.handle)
+      this.intervalHandles.delete(id)
+      logger.debug('Unregistered interval/once', { id })
+    }
+  }
+
+  /**
+   * Trigger a cron schedule immediately (extra one-shot fire — does not affect
+   * the natural fire schedule). croner's .trigger() returns a Promise that
+   * resolves after the callback finishes, so this method awaits it.
+   *
+   * Returns false if no cron schedule exists for `id` — interval/once cannot be
+   * manually triggered (no croner instance backing them).
+   */
+  async triggerNow(id: string): Promise<boolean> {
+    const cron = this.cronJobs.get(id)
+    if (!cron) {
+      logger.warn('triggerNow only supported for cron schedules', { id })
+      return false
+    }
+    await cron.trigger()
+    return true
+  }
+
+  /** Next scheduled fire time for cron schedules, or null otherwise. */
+  getNextRun(id: string): Date | null {
+    const cron = this.cronJobs.get(id)
+    if (cron) return cron.nextRun() ?? null
+    return null
+  }
+
+  has(id: string): boolean {
+    return this.cronJobs.has(id) || this.intervalHandles.has(id)
+  }
+
+  // ---------------- Private ----------------
+
+  private scheduleCron(id: string, trigger: Extract<Trigger, { kind: 'cron' }>, callback: ScheduleCallback): void {
+    const job = new Cron(
+      trigger.expr,
+      {
+        protect: true,
+        maxRuns: trigger.limit,
+        timezone: trigger.timezone,
+        catch: (err) => logger.error('cron callback error', { id, error: err })
+      },
+      callback
+    )
+    this.cronJobs.set(id, job)
+  }
+
+  private scheduleOnce(id: string, atMs: number, callback: ScheduleCallback): void {
+    const delay = Math.max(0, atMs - Date.now())
+    const handle = setTimeout(async () => {
+      // Self-clean before invoking so a re-entrant registerSchedule(id, ...)
+      // from inside the callback can install a new entry without conflict.
+      this.intervalHandles.delete(id)
+      try {
+        await callback()
+      } catch (err) {
+        logger.error('once-schedule callback error', { id, error: err })
+      }
+    }, delay)
+    handle.unref?.()
+    this.intervalHandles.set(id, { handle, ms: delay, callback })
+  }
+
+  private scheduleInterval(id: string, ms: number, callback: ScheduleCallback): void {
+    const fire = async (): Promise<void> => {
+      try {
+        await callback()
+      } catch (err) {
+        logger.error('interval-schedule callback error', { id, error: err })
+      }
+      // Re-arm only if not unregistered during callback. The map entry was set
+      // before the previous setTimeout fired; if it's still there, we're free
+      // to re-arm. unregister() during callback would have deleted the entry.
+      if (!this.intervalHandles.has(id)) return
+      const nextHandle = setTimeout(fire, ms)
+      nextHandle.unref?.()
+      this.intervalHandles.set(id, { handle: nextHandle, ms, callback })
+    }
+
+    const handle = setTimeout(fire, ms)
+    handle.unref?.()
+    this.intervalHandles.set(id, { handle, ms, callback })
+  }
+
+  private clearAll(): void {
+    for (const [id, job] of this.cronJobs) {
+      job.stop()
+      logger.debug('Stopped cron on shutdown', { id })
+    }
+    this.cronJobs.clear()
+    for (const [id, entry] of this.intervalHandles) {
+      clearTimeout(entry.handle)
+      logger.debug('Cleared interval/once on shutdown', { id })
+    }
+    this.intervalHandles.clear()
+  }
+}
--- a/src/main/data/db/schemas/job.ts
+++ b/src/main/data/db/schemas/job.ts
@@ -0,0 +1,101 @@
+import { sql } from 'drizzle-orm'
+import { check, foreignKey, index, integer, sqliteTable, text, uniqueIndex } from 'drizzle-orm/sqlite-core'
+
+import { createUpdateTimestamps, uuidPrimaryKey, uuidPrimaryKeyOrdered } from './_columnHelpers'
+
+/**
+ * Persistent schedule registry for recurring/once jobs.
+ *
+ * Each row maps a logical `type` (+ optional `name` for multi-instance types)
+ * to a `Trigger` (cron / interval / once) and a `jobInputTemplate` that becomes
+ * the `input` of every job spawned by this schedule. JobManager.registerJobSchedule
+ * owns the lifecycle; SchedulerService receives a `() => jobManager.enqueue(...)`
+ * callback and does not look at this table.
+ *
+ * NOTE: `(type, name)` is unique only for non-NULL `name`. Single-instance types
+ * (where name is omitted) are guarded at the application layer by
+ * JobScheduleService.create — SQLite treats every NULL name as distinct, so the
+ * "one schedule per single-instance type" invariant cannot be enforced by the DB.
+ */
+export const jobScheduleTable = sqliteTable(
+  'job_schedule',
+  {
+    id: uuidPrimaryKey(),
+    type: text().notNull(),
+    name: text(),
+    trigger: text().notNull(),
+    jobInputTemplate: text().notNull(),
+    enabled: integer({ mode: 'boolean' }).notNull().default(true),
+    nextRun: integer(),
+    lastRun: integer(),
+    catchUpPolicy: text().notNull(),
+    metadata: text().notNull().default('{}'),
+    ...createUpdateTimestamps
+  },
+  (t) => [
+    uniqueIndex('job_schedule_type_name_uq').on(t.type, t.name),
+    index('job_schedule_enabled_next_run_idx').on(t.enabled, t.nextRun),
+    index('job_schedule_type_idx').on(t.type)
+  ]
+)
+
+/**
+ * Single source of truth for every job's lifecycle. Six states:
+ *   pending → running → completed
+ *           ↘ failed
+ *           ↘ cancelled
+ *           ↘ delayed (scheduledAt > now or retry backoff) → pending
+ *
+ * dispatch path reads: WHERE queue=? AND status='pending' AND scheduledAt<=now()
+ * ORDER BY priority ASC, scheduledAt ASC LIMIT 1. The composite index
+ * job_queue_status_scheduled_at_idx covers it.
+ *
+ * idempotencyKey partial unique guarantees at most one non-terminal job per key.
+ * scheduleId index by (scheduleId, finishedAt) supports
+ * "last N terminal jobs by schedule" — used by handler.onSettled for circuit
+ * breaker logic (no separate tracker table needed).
+ */
+export const jobTable = sqliteTable(
+  'job',
+  {
+    id: uuidPrimaryKeyOrdered(),
+    type: text().notNull(),
+    status: text().notNull(),
+    priority: integer().notNull().default(0),
+    queue: text().notNull(),
+    idempotencyKey: text(),
+    scheduleId: text().references(() => jobScheduleTable.id, { onDelete: 'set null' }),
+    scheduledAt: integer().notNull(),
+    startedAt: integer(),
+    finishedAt: integer(),
+    attempt: integer().notNull().default(0),
+    maxAttempts: integer().notNull().default(3),
+    input: text().notNull(),
+    output: text(),
+    error: text(),
+    parentId: text(),
+    cancelRequested: integer({ mode: 'boolean' }).notNull().default(false),
+    metadata: text().notNull().default('{}'),
+    timeoutMs: integer(),
+    ...createUpdateTimestamps
+  },
+  (t) => [
+    check('job_status_check', sql`${t.status} IN ('pending','delayed','running','completed','failed','cancelled')`),
+    index('job_queue_status_scheduled_at_idx').on(t.queue, t.status, t.scheduledAt),
+    index('job_status_idx').on(t.status),
+    index('job_schedule_id_finished_at_idx').on(t.scheduleId, t.finishedAt),
+    index('job_parent_id_idx').on(t.parentId),
+    uniqueIndex('job_idempotency_key_partial_uq')
+      .on(t.idempotencyKey)
+      .where(sql`${t.idempotencyKey} IS NOT NULL AND ${t.status} NOT IN ('completed','failed','cancelled')`),
+    foreignKey({
+      columns: [t.parentId],
+      foreignColumns: [t.id]
+    }).onDelete('set null')
+  ]
+)
+
+export type JobRow = typeof jobTable.$inferSelect
+export type InsertJobRow = typeof jobTable.$inferInsert
+export type JobScheduleRow = typeof jobScheduleTable.$inferSelect
+export type InsertJobScheduleRow = typeof jobScheduleTable.$inferInsert
--- a/src/main/data/services/JobScheduleService.ts
+++ b/src/main/data/services/JobScheduleService.ts
@@ -0,0 +1,267 @@
+import { application } from '@application'
+import { type InsertJobScheduleRow, type JobScheduleRow, jobScheduleTable } from '@data/db/schemas/job'
+import { defaultHandlersFor, withSqliteErrors } from '@data/db/sqliteErrors'
+import type { DbOrTx } from '@data/db/types'
+import { timestampToISO } from '@data/services/utils/rowMappers'
+import { loggerService } from '@logger'
+import { DataApiErrorFactory } from '@shared/data/api'
+import {
+  type CatchUpPolicy,
+  type CreateJobScheduleDto,
+  JOB_ERROR_CODES,
+  JobScheduleNameAtomSchema,
+  type JobScheduleSnapshot,
+  type Trigger,
+  type UpdateJobScheduleDto
+} from '@shared/data/api/schemas/jobs'
+import { and, asc, eq, type SQL } from 'drizzle-orm'
+
+const logger = loggerService.withContext('JobScheduleService')
+
+export interface JobScheduleListFilter {
+  type?: string
+  enabled?: boolean
+  limit?: number
+  offset?: number
+}
+
+/**
+ * Owning entity service for `jobScheduleTable`. JobManager and DataApi handlers
+ * reach the table through this service — no direct Drizzle access elsewhere.
+ *
+ * Single-instance invariant: when `name` is null the type must have at most one
+ * schedule. SQLite treats every NULL as distinct so the DB unique index on
+ * `(type, name)` cannot enforce this; we check at the application layer in
+ * `create()` instead.
+ */
+export class JobScheduleService {
+  private getDb(): DbOrTx {
+    return application.get('DbService').getDb()
+  }
+
+  // ---------------- Read ----------------
+
+  async listAll(filter: JobScheduleListFilter = {}): Promise<JobScheduleSnapshot[]> {
+    const db = this.getDb()
+    const conditions: SQL[] = []
+    if (filter.type) conditions.push(eq(jobScheduleTable.type, filter.type))
+    if (filter.enabled !== undefined) conditions.push(eq(jobScheduleTable.enabled, filter.enabled))
+
+    const baseQuery = conditions.length
+      ? db
+          .select()
+          .from(jobScheduleTable)
+          .where(and(...conditions))
+          .orderBy(asc(jobScheduleTable.createdAt))
+      : db.select().from(jobScheduleTable).orderBy(asc(jobScheduleTable.createdAt))
+
+    const rows =
+      filter.limit !== undefined
+        ? filter.offset !== undefined
+          ? await baseQuery.limit(filter.limit).offset(filter.offset)
+          : await baseQuery.limit(filter.limit)
+        : await baseQuery
+
+    return rows.map((r) => this.rowToSnapshot(r))
+  }
+
+  async listEnabled(): Promise<JobScheduleSnapshot[]> {
+    const rows = await this.getDb()
+      .select()
+      .from(jobScheduleTable)
+      .where(eq(jobScheduleTable.enabled, true))
+      .orderBy(asc(jobScheduleTable.createdAt))
+    return rows.map((r) => this.rowToSnapshot(r))
+  }
+
+  async getById(id: string): Promise<JobScheduleSnapshot | null> {
+    const [row] = await this.getDb().select().from(jobScheduleTable).where(eq(jobScheduleTable.id, id)).limit(1)
+    return row ? this.rowToSnapshot(row) : null
+  }
+
+  /**
+   * Resolve a schedule by (type, name?). Returns null when not found so the
+   * caller (JobManager) can wrap absence into a typed error with a
+   * `knownNames` list for better DX.
+   *
+   * When `name` is omitted:
+   *   - 0 schedules → null
+   *   - exactly 1 → return it
+   *   - more than 1 → throw — caller must pass an explicit name
+   */
+  async getByTypeAndName(type: string, name?: string | null): Promise<JobScheduleSnapshot | null> {
+    const db = this.getDb()
+    if (name == null) {
+      const rows = await db.select().from(jobScheduleTable).where(eq(jobScheduleTable.type, type)).limit(2)
+      if (rows.length === 0) return null
+      if (rows.length > 1) {
+        throw DataApiErrorFactory.invalidOperation(
+          `getByTypeAndName: type "${type}" has multiple schedules — name is required to disambiguate`
+        )
+      }
+      return this.rowToSnapshot(rows[0])
+    }
+    const [row] = await db
+      .select()
+      .from(jobScheduleTable)
+      .where(and(eq(jobScheduleTable.type, type), eq(jobScheduleTable.name, name)))
+      .limit(1)
+    return row ? this.rowToSnapshot(row) : null
+  }
+
+  /** All known names for a type — JobManager uses this to build error context. */
+  async listNamesForType(type: string): Promise<Array<string | null>> {
+    const rows = await this.getDb()
+      .select({ name: jobScheduleTable.name })
+      .from(jobScheduleTable)
+      .where(eq(jobScheduleTable.type, type))
+    return rows.map((r) => r.name)
+  }
+
+  /**
+   * Schedules eligible for catch-up evaluation. Phase 1 returns all enabled
+   * schedules — JobManager applies the per-schedule policy. Filtering more
+   * aggressively here would duplicate policy knowledge from JobManager.
+   */
+  async getCatchUpCandidates(): Promise<JobScheduleSnapshot[]> {
+    return this.listEnabled()
+  }
+
+  // ---------------- Write ----------------
+
+  async create(dto: CreateJobScheduleDto): Promise<JobScheduleSnapshot> {
+    if (dto.name != null) {
+      const parsed = JobScheduleNameAtomSchema.safeParse(dto.name)
+      if (!parsed.success) {
+        throw DataApiErrorFactory.invalidOperation(
+          `Invalid schedule name: ${parsed.error.issues.map((i) => i.message).join('; ')}`
+        )
+      }
+    } else {
+      // Single-instance invariant: when name is null this type must have no schedules.
+      const existing = await this.getDb()
+        .select({ id: jobScheduleTable.id })
+        .from(jobScheduleTable)
+        .where(eq(jobScheduleTable.type, dto.type))
+        .limit(1)
+      if (existing[0]) {
+        throw DataApiErrorFactory.invalidOperation(
+          `${JOB_ERROR_CODES.SCHEDULE_SINGLETON_EXISTS}: Cannot create unnamed schedule for type "${dto.type}" — it already has schedules. Provide a name to make it multi-instance.`
+        )
+      }
+    }
+
+    const insertData: InsertJobScheduleRow = {
+      type: dto.type,
+      name: dto.name ?? null,
+      trigger: JSON.stringify(dto.trigger),
+      jobInputTemplate: JSON.stringify(dto.jobInputTemplate),
+      catchUpPolicy: JSON.stringify(dto.catchUpPolicy),
+      enabled: dto.enabled ?? true,
+      metadata: JSON.stringify(dto.metadata ?? {})
+    }
+
+    const result = await withSqliteErrors(() => this.getDb().insert(jobScheduleTable).values(insertData).returning(), {
+      ...defaultHandlersFor('JobSchedule', '<auto>'),
+      unique: () =>
+        DataApiErrorFactory.conflict(
+          'JobSchedule',
+          `${JOB_ERROR_CODES.SCHEDULE_NAME_CONFLICT}: name "${dto.name ?? '<unnamed>'}" already exists for type "${dto.type}"`
+        )
+    })
+    const row = result[0]
+    if (!row) throw new Error('jobScheduleService.create returned no row')
+    logger.info('JobSchedule created', { id: row.id, type: dto.type, name: dto.name })
+    return this.rowToSnapshot(row)
+  }
+
+  async update(id: string, patch: UpdateJobScheduleDto): Promise<JobScheduleSnapshot | null> {
+    if (patch.name != null) {
+      const parsed = JobScheduleNameAtomSchema.safeParse(patch.name)
+      if (!parsed.success) {
+        throw DataApiErrorFactory.invalidOperation(
+          `Invalid schedule name: ${parsed.error.issues.map((i) => i.message).join('; ')}`
+        )
+      }
+    }
+
+    const updateData: Partial<InsertJobScheduleRow> = { updatedAt: Date.now() }
+    if (patch.name !== undefined) updateData.name = patch.name
+    if (patch.trigger !== undefined) updateData.trigger = JSON.stringify(patch.trigger)
+    if (patch.jobInputTemplate !== undefined) updateData.jobInputTemplate = JSON.stringify(patch.jobInputTemplate)
+    if (patch.catchUpPolicy !== undefined) updateData.catchUpPolicy = JSON.stringify(patch.catchUpPolicy)
+    if (patch.enabled !== undefined) updateData.enabled = patch.enabled
+    if (patch.metadata !== undefined) updateData.metadata = JSON.stringify(patch.metadata)
+
+    const result = await withSqliteErrors(
+      () => this.getDb().update(jobScheduleTable).set(updateData).where(eq(jobScheduleTable.id, id)).returning(),
+      defaultHandlersFor('JobSchedule', id)
+    )
+    const row = result[0]
+    if (!row) return null
+    logger.info('JobSchedule updated', { id })
+    return this.rowToSnapshot(row)
+  }
+
+  async setEnabled(id: string, enabled: boolean): Promise<boolean> {
+    const result = await this.getDb()
+      .update(jobScheduleTable)
+      .set({ enabled, updatedAt: Date.now() })
+      .where(eq(jobScheduleTable.id, id))
+    return result.rowsAffected > 0
+  }
+
+  async delete(id: string): Promise<boolean> {
+    const result = await this.getDb().delete(jobScheduleTable).where(eq(jobScheduleTable.id, id))
+    logger.info('JobSchedule deleted', { id, deleted: result.rowsAffected > 0 })
+    return result.rowsAffected > 0
+  }
+
+  /**
+   * Record a fire event: set lastRun to the actual fire timestamp and nextRun
+   * to the next expected fire (or null for terminal one-shot / no-more-runs).
+   * Called from the SchedulerService callback after each fire.
+   */
+  async markFired(id: string, lastRun: number, nextRun: number | null): Promise<void> {
+    await this.getDb()
+      .update(jobScheduleTable)
+      .set({ lastRun, nextRun, updatedAt: Date.now() })
+      .where(eq(jobScheduleTable.id, id))
+  }
+
+  // ---------------- Row → Entity ----------------
+
+  /**
+   * Row → entity mapping. Like JobService.rowToSnapshot, this is written
+   * explicitly rather than via `{...nullsToUndefined(row)}` because the
+   * snapshot fields are `.nullable()` (not `.optional()`) to keep the IPC
+   * boundary clean. See JobService.rowToSnapshot for the rationale.
+   */
+  rowToSnapshot(row: JobScheduleRow): JobScheduleSnapshot {
+    return {
+      id: row.id,
+      type: row.type,
+      name: row.name,
+      trigger: this.parseJson(row.trigger) as Trigger,
+      jobInputTemplate: this.parseJson(row.jobInputTemplate),
+      enabled: row.enabled,
+      nextRun: row.nextRun != null ? timestampToISO(row.nextRun) : null,
+      lastRun: row.lastRun != null ? timestampToISO(row.lastRun) : null,
+      catchUpPolicy: this.parseJson(row.catchUpPolicy) as CatchUpPolicy,
+      metadata: (this.parseJson(row.metadata) as Record<string, unknown>) ?? {},
+      createdAt: timestampToISO(row.createdAt),
+      updatedAt: timestampToISO(row.updatedAt)
+    }
+  }
+
+  private parseJson(raw: string): unknown {
+    try {
+      return JSON.parse(raw)
+    } catch (err) {
+      logger.warn('Failed to parse JSON field', { rawHead: raw.slice(0, 100), error: (err as Error).message })
+      return {}
+    }
+  }
+}
+
+export const jobScheduleService = new JobScheduleService()
--- a/src/main/data/services/JobService.ts
+++ b/src/main/data/services/JobService.ts
@@ -0,0 +1,407 @@
+import { application } from '@application'
+import { type InsertJobRow, type JobRow, jobTable } from '@data/db/schemas/job'
+import { defaultHandlersFor, withSqliteErrors } from '@data/db/sqliteErrors'
+import type { DbOrTx } from '@data/db/types'
+import { timestampToISO } from '@data/services/utils/rowMappers'
+import { loggerService } from '@logger'
+import { type JobError, type JobSnapshot, type JobStatus, TERMINAL_JOB_STATUSES } from '@shared/data/api/schemas/jobs'
+import { and, asc, count, desc, eq, inArray, lte, type SQL } from 'drizzle-orm'
+
+const logger = loggerService.withContext('JobService')
+
+const NON_TERMINAL_STATUSES = ['pending', 'delayed', 'running'] as const satisfies readonly JobStatus[]
+
+export interface JobListFilter {
+  status?: JobStatus[]
+  queue?: string
+  type?: string
+  scheduleId?: string
+  limit?: number
+  offset?: number
+}
+
+/**
+ * Owning entity service for `jobTable`. JobManager and DataApi handlers reach
+ * the table through this service — no direct Drizzle access elsewhere.
+ *
+ * Tx-scoped methods (suffix `Tx`) accept a `DbOrTx` so JobManager can call them
+ * inside its dispatch transaction (Layer 0 + Layer 1 mutex protect the section).
+ * Non-tx methods use the singleton db handle via `this.getDb()`.
+ */
+export class JobService {
+  private getDb(): DbOrTx {
+    return application.get('DbService').getDb()
+  }
+
+  // ---------------- Read ----------------
+
+  async list(filter: JobListFilter = {}): Promise<JobSnapshot[]> {
+    const db = this.getDb()
+    const conditions: SQL[] = []
+    if (filter.status?.length) conditions.push(inArray(jobTable.status, filter.status))
+    if (filter.queue) conditions.push(eq(jobTable.queue, filter.queue))
+    if (filter.type) conditions.push(eq(jobTable.type, filter.type))
+    if (filter.scheduleId) conditions.push(eq(jobTable.scheduleId, filter.scheduleId))
+
+    const baseQuery = conditions.length
+      ? db
+          .select()
+          .from(jobTable)
+          .where(and(...conditions))
+          .orderBy(desc(jobTable.createdAt))
+      : db.select().from(jobTable).orderBy(desc(jobTable.createdAt))
+
+    const rows =
+      filter.limit !== undefined
+        ? filter.offset !== undefined
+          ? await baseQuery.limit(filter.limit).offset(filter.offset)
+          : await baseQuery.limit(filter.limit)
+        : await baseQuery
+
+    return rows.map((r) => this.rowToSnapshot(r))
+  }
+
+  async getById(id: string): Promise<JobSnapshot | null> {
+    const [row] = await this.getDb().select().from(jobTable).where(eq(jobTable.id, id)).limit(1)
+    return row ? this.rowToSnapshot(row) : null
+  }
+
+  /**
+   * Find any non-terminal job with the given idempotency key. JobManager.enqueue
+   * calls this for cross-restart deduplication: if a result is returned, reuse
+   * the existing job's handle instead of creating a new row.
+   */
+  async findActiveByIdempotencyKey(key: string): Promise<JobSnapshot | null> {
+    const [row] = await this.getDb()
+      .select()
+      .from(jobTable)
+      .where(and(eq(jobTable.idempotencyKey, key), inArray(jobTable.status, NON_TERMINAL_STATUSES)))
+      .limit(1)
+    return row ? this.rowToSnapshot(row) : null
+  }
+
+  /**
+   * The last N terminal jobs for a schedule, ordered by finishedAt DESC.
+   * Used by handler.onSettled to implement circuit-breaker logic without a
+   * separate tracker table — jobTable is the single source of truth.
+   */
+  async listRecentTerminalByScheduleId(scheduleId: string, limit: number): Promise<JobSnapshot[]> {
+    const rows = await this.getDb()
+      .select()
+      .from(jobTable)
+      .where(and(eq(jobTable.scheduleId, scheduleId), inArray(jobTable.status, TERMINAL_JOB_STATUSES)))
+      .orderBy(desc(jobTable.finishedAt))
+      .limit(limit)
+    return rows.map((r) => this.rowToSnapshot(r))
+  }
+
+  // ---------------- Write (non-tx) ----------------
+
+  async create(dto: InsertJobRow): Promise<JobSnapshot> {
+    const result = await withSqliteErrors(
+      () => this.getDb().insert(jobTable).values(dto).returning(),
+      defaultHandlersFor('Job', dto.id ?? '<auto>')
+    )
+    const row = result[0]
+    if (!row) throw new Error('jobService.create returned no row')
+    return this.rowToSnapshot(row)
+  }
+
+  // ---------------- Tx-scoped (inside JobManager.dispatch transaction) ----------------
+
+  /** Count pending+delayed+running jobs for a queue — checks queue concurrency. */
+  async countActiveByQueueTx(tx: DbOrTx, queue: string): Promise<number> {
+    const [r] = await tx
+      .select({ count: count() })
+      .from(jobTable)
+      .where(and(eq(jobTable.queue, queue), inArray(jobTable.status, NON_TERMINAL_STATUSES)))
+    return r?.count ?? 0
+  }
+
+  /**
+   * Count currently-running jobs across all queues — checks globalMaxConcurrency.
+   * Only `running` counts toward the global cap: pending/delayed do not occupy
+   * worker slots.
+   */
+  async countActiveGlobalTx(tx: DbOrTx): Promise<number> {
+    const [r] = await tx.select({ count: count() }).from(jobTable).where(eq(jobTable.status, 'running'))
+    return r?.count ?? 0
+  }
+
+  /**
+   * Atomically claim the next pending job in a queue and transition it to
+   * running. The double-mutex (Layer 0 global + Layer 1 per-queue) outside this
+   * tx ensures no two callers race on the same queue; the optimistic
+   * `WHERE status='pending'` is a belt-and-suspenders guard. Returns the
+   * claimed row (already updated to `running`) or null if none available.
+   */
+  async claimNextPendingTx(tx: DbOrTx, queue: string): Promise<JobRow | null> {
+    const now = Date.now()
+    const [candidate] = await tx
+      .select()
+      .from(jobTable)
+      .where(and(eq(jobTable.queue, queue), eq(jobTable.status, 'pending'), lte(jobTable.scheduledAt, now)))
+      .orderBy(asc(jobTable.priority), asc(jobTable.scheduledAt))
+      .limit(1)
+    if (!candidate) return null
+
+    const updated = await tx
+      .update(jobTable)
+      .set({ status: 'running', startedAt: now, updatedAt: now })
+      .where(and(eq(jobTable.id, candidate.id), eq(jobTable.status, 'pending')))
+      .returning()
+    return updated[0] ?? null
+  }
+
+  /** Move a job to a terminal state, persisting output and/or error. */
+  async setTerminalTx(
+    tx: DbOrTx,
+    jobId: string,
+    status: 'completed' | 'failed' | 'cancelled',
+    output: unknown | undefined,
+    error: JobError | null
+  ): Promise<void> {
+    const now = Date.now()
+    await tx
+      .update(jobTable)
+      .set({
+        status,
+        finishedAt: now,
+        updatedAt: now,
+        output: output !== undefined ? JSON.stringify(output) : null,
+        error: error ? JSON.stringify(error) : null
+      })
+      .where(eq(jobTable.id, jobId))
+  }
+
+  /**
+   * Re-schedule a failed job for retry. Caller computes `scheduledAt = now + backoff(attempt+1)`.
+   * Resets startedAt; preserves output/scheduleId/idempotencyKey.
+   */
+  async setDelayedRetryTx(
+    tx: DbOrTx,
+    jobId: string,
+    attempt: number,
+    scheduledAt: number,
+    error: JobError | null
+  ): Promise<void> {
+    const now = Date.now()
+    await tx
+      .update(jobTable)
+      .set({
+        status: 'delayed',
+        attempt,
+        scheduledAt,
+        startedAt: null,
+        updatedAt: now,
+        error: error ? JSON.stringify(error) : null
+      })
+      .where(eq(jobTable.id, jobId))
+  }
+
+  async setCancelRequestedTx(tx: DbOrTx, jobId: string): Promise<void> {
+    const now = Date.now()
+    await tx.update(jobTable).set({ cancelRequested: true, updatedAt: now }).where(eq(jobTable.id, jobId))
+  }
+
+  /**
+   * Atomically replace jobTable.metadata. Used by JobContext.patchMetadata
+   * to persist cross-restart state (e.g. remote-poll providerTaskId). The
+   * caller computes the merged JSON outside this method (so the in-memory
+   * row's metadata stays in sync without a re-fetch) and passes the final
+   * stringified payload in.
+   */
+  async setMetadataTx(tx: DbOrTx, jobId: string, mergedMetadataJson: string): Promise<void> {
+    const now = Date.now()
+    await tx.update(jobTable).set({ metadata: mergedMetadataJson, updatedAt: now }).where(eq(jobTable.id, jobId))
+  }
+
+  // ---------------- Startup recovery (JobManager.onReady) ----------------
+
+  /** All jobs currently marked `running` — typically orphans from a crash. */
+  async getStaleRunning(): Promise<JobRow[]> {
+    return this.getDb().select().from(jobTable).where(eq(jobTable.status, 'running'))
+  }
+
+  /** All non-terminal jobs for a type, sorted newest first — singleton recovery uses this. */
+  async getNonTerminalByType(type: string): Promise<JobRow[]> {
+    return this.getDb()
+      .select()
+      .from(jobTable)
+      .where(and(eq(jobTable.type, type), inArray(jobTable.status, NON_TERMINAL_STATUSES)))
+      .orderBy(desc(jobTable.createdAt))
+  }
+
+  async resetToPendingByIds(jobIds: string[]): Promise<void> {
+    if (jobIds.length === 0) return
+    const now = Date.now()
+    await this.getDb()
+      .update(jobTable)
+      .set({ status: 'pending', startedAt: null, updatedAt: now })
+      .where(inArray(jobTable.id, jobIds))
+  }
+
+  async cancelByIds(jobIds: string[], error: JobError | null): Promise<void> {
+    if (jobIds.length === 0) return
+    const now = Date.now()
+    await this.getDb()
+      .update(jobTable)
+      .set({
+        status: 'cancelled',
+        finishedAt: now,
+        updatedAt: now,
+        error: error ? JSON.stringify(error) : null
+      })
+      .where(inArray(jobTable.id, jobIds))
+  }
+
+  /**
+   * Cancel all non-terminal jobs matching a queue/type filter. Splits targets:
+   *   - running rows: mark cancelRequested=true. Caller aborts the in-flight
+   *     AbortController; handler observes signal.aborted and terminates;
+   *     normal finalize transitions the row to 'cancelled'.
+   *   - pending/delayed rows: transition directly to 'cancelled'.
+   *
+   * Returns `runningIds` (so the caller can abort their controllers) and
+   * `transitioned` (count of pending/delayed rows finalized synchronously).
+   * Used by JobManager.cancelMany — covers Phase 4 Knowledge reset() and
+   * FileProcessing batch cancellation semantics.
+   */
+  async cancelManyTx(
+    tx: DbOrTx,
+    filter: { queue?: string; type?: string },
+    error: JobError | null
+  ): Promise<{ runningIds: string[]; transitioned: number }> {
+    const conditions: SQL[] = [inArray(jobTable.status, NON_TERMINAL_STATUSES)]
+    if (filter.queue) conditions.push(eq(jobTable.queue, filter.queue))
+    if (filter.type) conditions.push(eq(jobTable.type, filter.type))
+
+    const matching = await tx
+      .select()
+      .from(jobTable)
+      .where(and(...conditions))
+    const runningIds = matching.filter((r) => r.status === 'running').map((r) => r.id)
+    const nonRunningIds = matching.filter((r) => r.status !== 'running').map((r) => r.id)
+
+    const now = Date.now()
+    if (runningIds.length) {
+      await tx.update(jobTable).set({ cancelRequested: true, updatedAt: now }).where(inArray(jobTable.id, runningIds))
+    }
+    let transitioned = 0
+    if (nonRunningIds.length) {
+      const result = await tx
+        .update(jobTable)
+        .set({
+          status: 'cancelled',
+          finishedAt: now,
+          updatedAt: now,
+          error: error ? JSON.stringify(error) : null
+        })
+        .where(inArray(jobTable.id, nonRunningIds))
+      transitioned = result.rowsAffected
+    }
+    return { runningIds, transitioned }
+  }
+
+  // ---------------- Delayed → pending promotion ----------------
+
+  /**
+   * Promote delayed jobs whose `scheduledAt` has passed into `pending` so the
+   * dispatch loop picks them up. Returns the count of rows promoted.
+   */
+  async promoteDelayedDue(now: number): Promise<number> {
+    const result = await this.getDb()
+      .update(jobTable)
+      .set({ status: 'pending', updatedAt: now })
+      .where(and(eq(jobTable.status, 'delayed'), lte(jobTable.scheduledAt, now)))
+    return result.rowsAffected
+  }
+
+  // ---------------- GC ----------------
+
+  /** Delete terminal jobs whose finishedAt is older than the cutoff. */
+  async pruneTerminalOlderThan(cutoffMs: number): Promise<number> {
+    const result = await this.getDb()
+      .delete(jobTable)
+      .where(and(inArray(jobTable.status, TERMINAL_JOB_STATUSES), lte(jobTable.finishedAt, cutoffMs)))
+    return result.rowsAffected
+  }
+
+  /**
+   * Keep only the latest `keepPerType` terminal jobs per type; delete the rest.
+   * At Phase 1 scale (thousands of terminal rows total) this in-memory pass is
+   * cheaper than a window-function SQL and portable across SQLite versions.
+   */
+  async pruneTerminalKeepLatestPerType(keepPerType: number): Promise<number> {
+    const allTerminal = await this.getDb()
+      .select({ id: jobTable.id, type: jobTable.type })
+      .from(jobTable)
+      .where(inArray(jobTable.status, TERMINAL_JOB_STATUSES))
+      .orderBy(desc(jobTable.finishedAt))
+
+    const perType = new Map<string, number>()
+    const toDelete: string[] = []
+    for (const row of allTerminal) {
+      const c = (perType.get(row.type) ?? 0) + 1
+      perType.set(row.type, c)
+      if (c > keepPerType) toDelete.push(row.id)
+    }
+    if (toDelete.length === 0) return 0
+    const result = await this.getDb().delete(jobTable).where(inArray(jobTable.id, toDelete))
+    return result.rowsAffected
+  }
+
+  // ---------------- Row → Entity ----------------
+
+  /**
+   * Row → entity mapping. Intentionally explicit rather than the
+   * `{...nullsToUndefined(row), ...}` skeleton from data-api-in-main.md.
+   *
+   * JobSnapshot's nullable fields are declared as `.nullable()` (not
+   * `.optional()`) so DB NULL → snapshot null cleanly crosses the IPC
+   * boundary. `nullsToUndefined` would actively break that — it turns
+   * `string | null` into `string | undefined`, forcing every renderer reader
+   * to handle a third state. The explicit mapping below preserves the
+   * T|null shape directly. notNull columns (id / type / status / queue /
+   * scheduledAt / attempt / maxAttempts / cancelRequested / metadata /
+   * createdAt / updatedAt) cannot hold NULL at the DB level, so there is
+   * nothing for `nullsToUndefined` to translate anyway.
+   */
+  rowToSnapshot(row: JobRow): JobSnapshot {
+    return {
+      id: row.id,
+      type: row.type,
+      status: row.status as JobStatus,
+      priority: row.priority,
+      queue: row.queue,
+      idempotencyKey: row.idempotencyKey,
+      scheduleId: row.scheduleId,
+      scheduledAt: timestampToISO(row.scheduledAt),
+      startedAt: row.startedAt != null ? timestampToISO(row.startedAt) : null,
+      finishedAt: row.finishedAt != null ? timestampToISO(row.finishedAt) : null,
+      attempt: row.attempt,
+      maxAttempts: row.maxAttempts,
+      input: this.parseJson(row.input, undefined),
+      output: row.output != null ? this.parseJson(row.output, null) : null,
+      error: row.error != null ? (this.parseJson(row.error, null) as JobError | null) : null,
+      parentId: row.parentId,
+      cancelRequested: row.cancelRequested,
+      metadata: (this.parseJson(row.metadata, {}) as Record<string, unknown>) ?? {},
+      timeoutMs: row.timeoutMs,
+      createdAt: timestampToISO(row.createdAt),
+      updatedAt: timestampToISO(row.updatedAt)
+    }
+  }
+
+  private parseJson(raw: string | null, fallback: unknown): unknown {
+    if (raw == null) return fallback
+    try {
+      return JSON.parse(raw)
+    } catch (err) {
+      logger.warn('Failed to parse JSON field', { rawHead: raw.slice(0, 100), error: (err as Error).message })
+      return fallback
+    }
+  }
+}
+
+export const jobService = new JobService()
--- a/src/renderer/src/hooks/useJob.ts
+++ b/src/renderer/src/hooks/useJob.ts
@@ -0,0 +1,36 @@
+import { useSharedCache } from '@renderer/data/hooks/useCache'
+import type { JobSnapshot } from '@shared/data/api/schemas/jobs'
+
+/**
+ * Subscribe to a job's live state in the renderer.
+ *
+ * Reads `jobs.state.${jobId}` from the shared cache — JobManager publishes a
+ * fresh snapshot on every state transition (pending → running → completed /
+ * failed / cancelled) and on progress reports. Cross-window sync is provided
+ * by CacheService.
+ *
+ * Phase 1 behavior:
+ *   - First render: `data` is null until JobManager pushes the initial snapshot.
+ *   - During execution: `data` updates on each transition / progress report.
+ *   - After terminal: snapshot persists for the cache TTL (60s), then null.
+ *     Renderer treats post-TTL null as "job finished, no more updates needed".
+ *
+ * Phase 2 will add a DataApi fallback (GET /jobs/:id) so post-TTL recalls
+ * still resolve the terminal snapshot. Until then, callers that need to
+ * recover a job state after TTL should refetch via their own mechanism.
+ */
+export interface UseJobResult {
+  data: JobSnapshot | null
+  isTerminal: boolean
+  isLoading: boolean
+  error: Error | undefined
+}
+
+const TERMINAL_STATUSES: ReadonlySet<JobSnapshot['status']> = new Set(['completed', 'failed', 'cancelled'])
+
+export function useJob(jobId: string): UseJobResult {
+  const [snapshot] = useSharedCache(`jobs.state.${jobId}` as const)
+  const data = snapshot ?? null
+  const isTerminal = data ? TERMINAL_STATUSES.has(data.status) : false
+  return { data, isTerminal, isLoading: false, error: undefined }
+}