diff --git a/src/modules/workflow/Application/BuildFlow/Imports/Commands/ProcessAvailableImportJobsCommand.cs b/src/modules/workflow/Application/BuildFlow/Imports/Commands/ProcessAvailableImportJobsCommand.cs
new file mode 100644
index 0000000..e511c2b
--- /dev/null
+++ b/src/modules/workflow/Application/BuildFlow/Imports/Commands/ProcessAvailableImportJobsCommand.cs
@@ -0,0 +1,12 @@
+using MediatR;
+
+namespace Workflow.Application.BuildFlow.Imports.Commands;
+
+/// <summary>
+/// 原子领取并处理可执行导入任务的命令。
+/// </summary>
+public sealed record ProcessAvailableImportJobsCommand(
+    string LeaseOwner,
+    int BatchSize,
+    TimeSpan LeaseDuration,
+    int MaxAttempts) : IRequest<int>;
diff --git a/src/modules/workflow/Application/BuildFlow/Imports/Handlers/ProcessAvailableImportJobsCommandHandler.cs b/src/modules/workflow/Application/BuildFlow/Imports/Handlers/ProcessAvailableImportJobsCommandHandler.cs
new file mode 100644
index 0000000..d80bb10
--- /dev/null
+++ b/src/modules/workflow/Application/BuildFlow/Imports/Handlers/ProcessAvailableImportJobsCommandHandler.cs
@@ -0,0 +1,130 @@
+using System.Data.Common;
+using MediatR;
+using Microsoft.EntityFrameworkCore.Storage;
+using Microsoft.Extensions.Logging;
+using Workflow.Application.BuildFlow.Imports.Commands;
+using Workflow.Application.BuildFlow.Imports.Services;
+using Workflow.Domain.Enums;
+using Workflow.Infrastructure.Persistence;
+
+namespace Workflow.Application.BuildFlow.Imports.Handlers;
+
+/// <summary>
+/// 使用 PostgreSQL 租约原子领取导入任务，并逐个交给模块内处理器执行。
+/// </summary>
+public sealed class ProcessAvailableImportJobsCommandHandler(
+    WorkflowDbContext db,
+    IImportJobProcessor processor,
+    ILogger<ProcessAvailableImportJobsCommandHandler> logger)
+    : IRequestHandler<ProcessAvailableImportJobsCommand, int>
+{
+    public async Task<int> Handle(
+        ProcessAvailableImportJobsCommand request,
+        CancellationToken ct)
+    {
+        var claimedIds = await ClaimAsync(request, ct);
+        foreach (var jobId in claimedIds)
+        {
+            ct.ThrowIfCancellationRequested();
+            try
+            {
+                await processor.ProcessAsync(
+                    jobId,
+                    request.LeaseOwner,
+                    request.LeaseDuration,
+                    ct);
+            }
+            catch (OperationCanceledException) when (ct.IsCancellationRequested)
+            {
+                throw;
+            }
+            catch (Exception ex)
+            {
+                logger.LogError(ex, "导入任务 {JobId} 处理失败，等待租约恢复", jobId);
+            }
+        }
+
+        return claimedIds.Count;
+    }
+
+    private async Task<List<Guid>> ClaimAsync(
+        ProcessAvailableImportJobsCommand request,
+        CancellationToken ct)
+    {
+        var now = DateTime.UtcNow;
+        var leaseExpiresAt = now.Add(request.LeaseDuration);
+        await using var transaction = await db.Database.BeginTransactionAsync(ct);
+
+        await db.Database.ExecuteSqlInterpolatedAsync($"""
+            UPDATE b_import_jobs
+            SET status = {(int)BuildFlowImportJobStatus.Failed},
+                lease_owner = NULL,
+                lease_expires_at = NULL,
+                finished_at = {now},
+                error_message = {"任务租约过期且已达到最大重试次数"},
+                updated_by = {"system"},
+                updated_at = {now}
+            WHERE status = {(int)BuildFlowImportJobStatus.Running}
+              AND lease_expires_at <= {now}
+              AND attempt_count >= {request.MaxAttempts}
+            """, ct);
+
+        await using var command = db.Database.GetDbConnection().CreateCommand();
+        command.Transaction = transaction.GetDbTransaction();
+        command.CommandText = """
+            WITH candidates AS (
+                SELECT id
+                FROM b_import_jobs
+                WHERE (
+                    status = @pending
+                    OR (status = @running AND lease_expires_at <= @now)
+                )
+                  AND attempt_count < @max_attempts
+                ORDER BY created_at
+                FOR UPDATE SKIP LOCKED
+                LIMIT @batch_size
+            )
+            UPDATE b_import_jobs AS job
+            SET status = @running,
+                lease_owner = @lease_owner,
+                lease_expires_at = @lease_expires_at,
+                attempt_count = job.attempt_count + 1,
+                last_heartbeat_at = @now,
+                started_at = COALESCE(job.started_at, @now),
+                finished_at = NULL,
+                error_message = NULL,
+                updated_by = 'system',
+                updated_at = @now
+            FROM candidates
+            WHERE job.id = candidates.id
+            RETURNING job.id
+            """;
+        AddParameter(command, "pending", (int)BuildFlowImportJobStatus.Pending);
+        AddParameter(command, "running", (int)BuildFlowImportJobStatus.Running);
+        AddParameter(command, "now", now);
+        AddParameter(command, "max_attempts", request.MaxAttempts);
+        AddParameter(command, "batch_size", request.BatchSize);
+        AddParameter(command, "lease_owner", request.LeaseOwner);
+        AddParameter(command, "lease_expires_at", leaseExpiresAt);
+
+        var claimedIds = new List<Guid>();
+        await using (var reader = await command.ExecuteReaderAsync(ct))
+        {
+            while (await reader.ReadAsync(ct))
+            {
+                claimedIds.Add(reader.GetGuid(0));
+            }
+        }
+
+        await transaction.CommitAsync(ct);
+        return claimedIds;
+    }
+
+    private static void AddParameter(DbCommand command, string name, object value)
+    {
+        var parameter = command.CreateParameter();
+        parameter.ParameterName = name;
+        parameter.Value = value;
+        command.Parameters.Add(parameter);
+    }
+}
diff --git a/src/modules/workflow/Application/BuildFlow/Imports/Services/IImportJobProcessor.cs b/src/modules/workflow/Application/BuildFlow/Imports/Services/IImportJobProcessor.cs
index 086c2e6..f745e47 100644
--- a/src/modules/workflow/Application/BuildFlow/Imports/Services/IImportJobProcessor.cs
+++ b/src/modules/workflow/Application/BuildFlow/Imports/Services/IImportJobProcessor.cs
@@ -1,8 +1,16 @@
 namespace Workflow.Application.BuildFlow.Imports.Services;
 
-/// <summary>导入任务核心处理器。由后台服务（ImportJobProcessorService）调用。</summary>
+/// <summary>导入任务核心处理器。仅处理已由 Application 命令原子领取的任务。</summary>
 public interface IImportJobProcessor
 {
-    /// <summary>处理一个 Pending 导入任务。处理器内部负责状态流转（Running → Success/Failed/PartialSuccess）。</summary>
-    Task ProcessAsync(Guid jobId, CancellationToken ct);
+    /// <summary>处理已持有指定租约的 Running 导入任务。</summary>
+    Task ProcessAsync(
+        Guid jobId,
+        string leaseOwner,
+        TimeSpan leaseDuration,
+        CancellationToken ct);
+
+    /// <summary>旧 Host 扫描器的临时兼容入口，Host 改为 MediatR 后删除。</summary>
+    Task ProcessAsync(Guid jobId, CancellationToken ct)
+        => ProcessAsync(jobId, "legacy-import-host", TimeSpan.FromMinutes(5), ct);
 }
diff --git a/src/modules/workflow/Application/BuildFlow/Imports/Services/ImportJobProcessor.cs b/src/modules/workflow/Application/BuildFlow/Imports/Services/ImportJobProcessor.cs
index 5501e63..2f2783c 100644
--- a/src/modules/workflow/Application/BuildFlow/Imports/Services/ImportJobProcessor.cs
+++ b/src/modules/workflow/Application/BuildFlow/Imports/Services/ImportJobProcessor.cs
@@ -24,7 +24,11 @@ public class ImportJobProcessor(
         WriteIndented = false,
     };
 
-    public async Task ProcessAsync(Guid jobId, CancellationToken ct)
+    public async Task ProcessAsync(
+        Guid jobId,
+        string leaseOwner,
+        TimeSpan leaseDuration,
+        CancellationToken ct)
     {
         var job = await db.BuildFlowImportJobs.FindAsync([jobId], ct);
         if (job is null)
diff --git a/tests/BuildFlow.Workflow.Tests/Reliability/ImportJobClaimTests.cs b/tests/BuildFlow.Workflow.Tests/Reliability/ImportJobClaimTests.cs
new file mode 100644
index 0000000..b36cf0a
--- /dev/null
+++ b/tests/BuildFlow.Workflow.Tests/Reliability/ImportJobClaimTests.cs
@@ -0,0 +1,158 @@
+using System.Collections.Concurrent;
+using FluentAssertions;
+using Microsoft.EntityFrameworkCore;
+using Microsoft.Extensions.Logging.Abstractions;
+using Workflow.Application.BuildFlow.Imports.Commands;
+using Workflow.Application.BuildFlow.Imports.Handlers;
+using Workflow.Application.BuildFlow.Imports.Services;
+using Workflow.Domain.Entities;
+using Workflow.Domain.Enums;
+using Xunit;
+
+namespace BuildFlow.Workflow.Tests.Reliability;
+
+/// <summary>
+/// 导入任务的 PostgreSQL 原子领取和过期租约恢复测试。
+/// </summary>
+[Collection(PostgresWorkflowCollection.Name)]
+public class ImportJobClaimTests(PostgresWorkflowFixture fixture)
+{
+    [Fact]
+    public async Task TwoWorkersCanClaimPendingImportOnlyOnce()
+    {
+        await ResetAsync();
+        var jobId = await SeedImportAsync(BuildFlowImportJobStatus.Pending);
+        var processor = new RecordingImportJobProcessor();
+        await using var dbA = fixture.CreateDbContext();
+        await using var dbB = fixture.CreateDbContext();
+        var handlerA = CreateHandler(dbA, processor);
+        var handlerB = CreateHandler(dbB, processor);
+
+        var results = await Task.WhenAll(
+            handlerA.Handle(CreateCommand("worker-a"), CancellationToken.None),
+            handlerB.Handle(CreateCommand("worker-b"), CancellationToken.None));
+
+        results.Sum().Should().Be(1);
+        processor.Calls.Should().ContainSingle(x => x.JobId == jobId);
+        await using var verify = fixture.CreateDbContext();
+        var job = await verify.BuildFlowImportJobs.SingleAsync(x => x.Id == jobId);
+        job.Status.Should().Be(BuildFlowImportJobStatus.Running);
+        job.AttemptCount.Should().Be(1);
+        job.LeaseOwner.Should().BeOneOf("worker-a", "worker-b");
+        job.LastHeartbeatAt.Should().NotBeNull();
+    }
+
+    [Fact]
+    public async Task ExpiredRunningImportCanBeReclaimed()
+    {
+        await ResetAsync();
+        var jobId = await SeedImportAsync(
+            BuildFlowImportJobStatus.Running,
+            "dead-worker",
+            DateTime.UtcNow.AddMinutes(-1),
+            attemptCount: 1);
+        var processor = new RecordingImportJobProcessor();
+        await using var db = fixture.CreateDbContext();
+        var handler = CreateHandler(db, processor);
+
+        var claimed = await handler.Handle(CreateCommand("recovery-worker"), CancellationToken.None);
+
+        claimed.Should().Be(1);
+        processor.Calls.Should().ContainSingle(x =>
+            x.JobId == jobId && x.LeaseOwner == "recovery-worker");
+        await using var verify = fixture.CreateDbContext();
+        var job = await verify.BuildFlowImportJobs.SingleAsync(x => x.Id == jobId);
+        job.AttemptCount.Should().Be(2);
+        job.LeaseOwner.Should().Be("recovery-worker");
+        job.LastHeartbeatAt.Should().NotBeNull();
+    }
+
+    [Fact]
+    public async Task ExpiredImportAtMaxAttemptsBecomesFailed()
+    {
+        await ResetAsync();
+        var jobId = await SeedImportAsync(
+            BuildFlowImportJobStatus.Running,
+            "dead-worker",
+            DateTime.UtcNow.AddMinutes(-1),
+            attemptCount: 5);
+        var processor = new RecordingImportJobProcessor();
+        await using var db = fixture.CreateDbContext();
+        var handler = CreateHandler(db, processor);
+
+        var claimed = await handler.Handle(CreateCommand("worker-a"), CancellationToken.None);
+
+        claimed.Should().Be(0);
+        processor.Calls.Should().BeEmpty();
+        await using var verify = fixture.CreateDbContext();
+        var job = await verify.BuildFlowImportJobs.SingleAsync(x => x.Id == jobId);
+        job.Status.Should().Be(BuildFlowImportJobStatus.Failed);
+        job.LeaseOwner.Should().BeNull();
+        job.LeaseExpiresAt.Should().BeNull();
+        job.FinishedAt.Should().NotBeNull();
+        job.ErrorMessage.Should().Contain("最大重试次数");
+    }
+
+    private static ProcessAvailableImportJobsCommand CreateCommand(string owner) => new(
+        owner,
+        BatchSize: 10,
+        LeaseDuration: TimeSpan.FromMinutes(5),
+        MaxAttempts: 5);
+
+    private static ProcessAvailableImportJobsCommandHandler CreateHandler(
+        WorkflowDbContext db,
+        IImportJobProcessor processor)
+        => new(
+            db,
+            processor,
+            NullLogger<ProcessAvailableImportJobsCommandHandler>.Instance);
+
+    private async Task ResetAsync()
+    {
+        await using var db = fixture.CreateDbContext();
+        await db.BuildFlowImportJobs.ExecuteDeleteAsync();
+    }
+
+    private async Task<Guid> SeedImportAsync(
+        BuildFlowImportJobStatus status,
+        string? leaseOwner = null,
+        DateTime? leaseExpiresAt = null,
+        int attemptCount = 0)
+    {
+        await using var db = fixture.CreateDbContext();
+        var job = new BuildFlowImportJob
+        {
+            Id = Guid.NewGuid(),
+            FormDefinitionId = Guid.NewGuid(),
+            Status = status,
+            FileUrl = "unused.xlsx",
+            OriginalFileName = "unused.xlsx",
+            LeaseOwner = leaseOwner,
+            LeaseExpiresAt = leaseExpiresAt,
+            AttemptCount = attemptCount,
+        };
+        db.BuildFlowImportJobs.Add(job);
+        await db.SaveChangesAsync();
+        return job.Id;
+    }
+
+    private sealed class RecordingImportJobProcessor : IImportJobProcessor
+    {
+        public ConcurrentQueue<ImportJobCall> Calls { get; } = new();
+
+        public Task ProcessAsync(
+            Guid jobId,
+            string leaseOwner,
+            TimeSpan leaseDuration,
+            CancellationToken ct)
+        {
+            Calls.Enqueue(new ImportJobCall(jobId, leaseOwner, leaseDuration));
+            return Task.CompletedTask;
+        }
+    }
+
+    private sealed record ImportJobCall(
+        Guid JobId,
+        string LeaseOwner,
+        TimeSpan LeaseDuration);
+}