build-flow.api/docs/superpowers/plans/2026-07-23-buildflow-database-batch-performance.md

# BuildFlow Database And Batch Performance Implementation Plan

> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.

**Goal:** Make workflow aggregate refresh, import persistence, and monitoring synchronization use constant-query or bounded-batch database access without changing HTTP contracts.

**Architecture:** Workflow aggregate refresh builds one calculation plan, bulk-loads definitions and rows, and updates the workflow variable mirror once. Import parses rows lazily, writes bounded transactional batches through a module-owned use case, and advances the committed cursor in the same transaction. Monitoring persists at most 500 records per parameterized PostgreSQL `ON CONFLICT DO UPDATE` statement and commits records, job success, and external-system sync metadata atomically.

**Tech Stack:** .NET 10, EF Core 10, Npgsql/PostgreSQL 17, MediatR 12, ClosedXML, xUnit, FluentAssertions.

---

## Scope And Compatibility

- Keep all existing routes, request DTOs, response DTOs, permissions, status values, and SSE contracts unchanged.
- Do not add a cross-module service bridge. New workflow services remain module-internal.
- Keep the second-batch lease contract: `SuccessRows + FailedRows` is the committed import cursor.
- Third batch makes Excel rows lazy over the existing ClosedXML workbook. Fourth batch replaces the workbook implementation with OpenXML streaming and adds upload/export byte boundaries.
- Workflow aggregate SQL count must stay constant as target-row and aggregate-field counts grow.
- Import row batch size defaults to 200 and must be between 1 and 1000.
- Monitoring UPSERT statement batches contain at most 500 records and every external value is a database parameter.

### Task 1: Record The Third-Batch Plan

**Files:**
- Create: `docs/superpowers/plans/2026-07-23-buildflow-database-batch-performance.md`

- [ ] **Step 1: Review the plan against design sections 6.1, 6.2, and 6.3**

Confirm this plan covers constant aggregate queries, lazy import enumeration, bounded transactional import batches, full deterministic instance IDs, PostgreSQL UPSERT, concurrency, invalid JSON warnings, and verification gates.

- [ ] **Step 2: Commit the plan**

```bash
git add docs/superpowers/plans/2026-07-23-buildflow-database-batch-performance.md
git commit -m "docs: plan database and batch performance"
```

### Task 2: Make Aggregate Refresh Query Count Constant

**Files:**
- Create: `tests/BuildFlow.Workflow.Tests/Performance/CountingCommandInterceptor.cs`
- Create: `tests/BuildFlow.Workflow.Tests/Performance/FormAggregateQueryCountTests.cs`
- Modify: `tests/BuildFlow.Workflow.Tests/Reliability/PostgresWorkflowFixture.cs`
- Modify: `src/modules/workflow/Application/Form/Aggregates/FormAggregateField.cs`
- Modify: `src/modules/workflow/Application/Form/Aggregates/IFormAggregateRefreshService.cs`
- Modify: `src/modules/workflow/Application/Form/Aggregates/FormAggregateRefreshService.cs`

- [ ] **Step 1: Add a SELECT-counting interceptor**

Create a `DbCommandInterceptor` whose `ReaderExecuting` and `ReaderExecutingAsync` methods increment `SelectCount` only when trimmed command text starts with `SELECT`. Expose `Reset()` so seed commands are excluded from the assertion.

```csharp
public sealed class CountingCommandInterceptor : DbCommandInterceptor
{
    public int SelectCount { get; private set; }

    public void Reset() => SelectCount = 0;

    public override InterceptionResult<DbDataReader> ReaderExecuting(
        DbCommand command,
        CommandEventData eventData,
        InterceptionResult<DbDataReader> result)
    {
        Count(command);
        return result;
    }

    public override ValueTask<InterceptionResult<DbDataReader>> ReaderExecutingAsync(
        DbCommand command,
        CommandEventData eventData,
        InterceptionResult<DbDataReader> result,
        CancellationToken ct = default)
    {
        Count(command);
        return ValueTask.FromResult(result);
    }

    private void Count(DbCommand command)
    {
        if (command.CommandText.TrimStart().StartsWith("SELECT", StringComparison.OrdinalIgnoreCase))
            SelectCount++;
    }
}
```

- [ ] **Step 2: Allow the PostgreSQL fixture to attach interceptors**

Change the factory signature to:

```csharp
public WorkflowDbContext CreateDbContext(params IInterceptor[] interceptors)
```

Add them to the options only when the array is non-empty, while preserving snake-case naming and the audit stub.

- [ ] **Step 3: Write the failing constant-query test**

Seed one target definition containing two aggregate fields that both reference `expense-line`, ten target rows, twenty source rows, and one workflow instance. Reset the interceptor, call `RefreshAsync`, and assert:

```csharp
interceptor.SelectCount.Should().BeLessThanOrEqualTo(4);
updatedTargetRows.Should().OnlyContain(x => x.DataJson.Contains("\"lineTotal\":30"));
updatedInstance.Variables.Should().Contain("\"lineTotal\":30");
```

Use relation values so all ten target rows calculate independently. Current code must fail because it queries source definitions, source rows, and the workflow instance inside loops.

- [ ] **Step 4: Run and verify RED**

```bash
dotnet test tests/BuildFlow.Workflow.Tests/BuildFlow.Workflow.Tests.csproj --no-restore --filter FullyQualifiedName~FormAggregateQueryCountTests
```

Expected: the value assertions pass but `SelectCount <= 4` fails with a count that grows with rows and aggregate fields.

- [ ] **Step 5: Parse aggregate metadata into a calculation plan**

Replace `FormAggregateField` with:

```csharp
internal sealed record FormAggregateField(
    string Path,
    string SourceFormCode,
    string? SourceRelationPath,
    string? TargetRelationValuePath,
    string Function,
    string? ValuePath);
```

`CollectAggregateFields` must discard incomplete metadata and return fully parsed fields. This makes schema parsing happen once per refresh.

- [ ] **Step 6: Add batch refresh methods to the interface**

```csharp
Task RefreshBatchAsync(
    Guid formDefinitionId,
    IReadOnlyCollection<Guid> instanceIds,
    CancellationToken ct);

Task RefreshAffectedBatchAsync(
    Guid sourceFormDefinitionId,
    IReadOnlyCollection<Guid> instanceIds,
    CancellationToken ct);
```

Keep existing single-instance methods and make them delegate to the batch methods.

- [ ] **Step 7: Implement constant-query refresh**

For `RefreshBatchAsync`:

1. Return for an empty instance set.
2. Load the target definition once and throw `NotFoundException(ErrorCode.NotFound, ...)` when absent.
3. Parse its schema once and group fields by `SourceFormCode`.
4. Load all source definitions with one `Code IN (...)` query.
5. Load target and source `FormData` for all requested instances with one query.
6. Load all matching `WorkflowInstance` rows with one query inside the existing best-effort `DbException` boundary.
7. Parse each `DataJson` once, index source rows by `(FormDefinitionId, InstanceId)`, calculate all target values in memory, and assign each target row once.
8. Merge all variable updates in memory and call `SaveChangesAsync` once.

For `RefreshAffectedBatchAsync`, bulk-load the source definition, target form IDs, and target definitions, filter definitions in memory with `ReferencesSourceForm`, then call `RefreshBatchAsync` once per affected target definition, never once per source row.

- [ ] **Step 8: Run focused and existing aggregate tests**

```bash
dotnet test tests/BuildFlow.Workflow.Tests/BuildFlow.Workflow.Tests.csproj --no-restore --filter "FullyQualifiedName~FormAggregateQueryCountTests|FullyQualifiedName~FormAggregateFieldTests|FullyQualifiedName~FormDataTests"
dotnet build src/modules/workflow/Application/Workflow.Application.csproj --no-restore
```

Expected: all selected tests pass; build has 0 warnings and 0 errors.

- [ ] **Step 9: Commit aggregate optimization**

```bash
git add tests/BuildFlow.Workflow.Tests/Performance tests/BuildFlow.Workflow.Tests/Reliability/PostgresWorkflowFixture.cs src/modules/workflow/Application/Form/Aggregates
git commit -m "perf: batch workflow aggregate refresh"
```

### Task 3: Prepare Schemas Once And Enumerate Excel Rows Lazily

**Files:**
- Create: `src/modules/workflow/Application/Form/Schema/PreparedFormSchema.cs`
- Create: `src/modules/workflow/Application/BuildFlow/Common/Services/ExcelParsedRow.cs`
- Modify: `src/modules/workflow/Application/Form/Schema/FormDataValidator.cs`
- Modify: `src/modules/workflow/Application/Form/Derivation/IFormValueDerivationService.cs`
- Modify: `src/modules/workflow/Application/Form/Derivation/FormValueDerivationService.cs`
- Modify: `src/modules/workflow/Application/BuildFlow/Common/Services/IFormSchemaExcelService.cs`
- Modify: `src/modules/workflow/Application/BuildFlow/Common/Services/FormSchemaExcelService.cs`
- Create: `tests/BuildFlow.Workflow.Tests/Performance/PreparedFormSchemaTests.cs`
- Create: `tests/BuildFlow.Workflow.Tests/Performance/LazyExcelRowTests.cs`

- [ ] **Step 1: Write failing prepared-schema tests**

The desired API is:

```csharp
var prepared = PreparedFormSchema.Create(schemaJson);
var first = FormDataValidator.Validate(prepared, """{"name":"A"}""");
var second = await derivation.DeriveAsync(
    prepared,
    """{"name":"B"}""",
    new FormDerivationContext(Guid.NewGuid(), Guid.NewGuid(), null),
    CancellationToken.None);
```

Assert the same prepared object supports repeated validation and derivation and rejects an invalid schema during `Create`.

- [ ] **Step 2: Write the failing lazy-enumeration test**

Change the desired parser contract to `IEnumerable<ExcelParsedRow>`. Build an in-memory workbook with three data rows, call `Parse`, assert the return type is not `IReadOnlyList<ExcelParsedRow>`, enumerate only `Take(1)`, and assert the first row is returned correctly.

- [ ] **Step 3: Run and verify RED**

```bash
dotnet test tests/BuildFlow.Workflow.Tests/BuildFlow.Workflow.Tests.csproj --no-restore --filter "FullyQualifiedName~PreparedFormSchemaTests|FullyQualifiedName~LazyExcelRowTests"
```

Expected: compilation fails because `PreparedFormSchema` and the new parser contract do not exist.

- [ ] **Step 4: Add `PreparedFormSchema`**

The type owns the original JSON, parsed `JsonObject`, and `SchemaValidationResult`:

```csharp
public sealed class PreparedFormSchema
{
    private PreparedFormSchema(
        string schemaJson,
        JsonObject schema,
        SchemaValidationResult validation)
    {
        SchemaJson = schemaJson;
        Schema = schema;
        Validation = validation;
    }

    public string SchemaJson { get; }
    public JsonObject Schema { get; }
    public SchemaValidationResult Validation { get; }

    public static PreparedFormSchema Create(string schemaJson)
    {
        var validation = SchemaValidator.Validate(schemaJson);
        if (!validation.IsValid)
            throw new BusinessException(
                ErrorCode.InvalidParam,
                $"表单 Schema 无效: {string.Join("; ", validation.Errors)}");
        var schema = JsonNode.Parse(schemaJson) as JsonObject ?? new JsonObject();
        return new PreparedFormSchema(schemaJson, schema, validation);
    }
}
```

- [ ] **Step 5: Add prepared overloads**

Keep existing string overloads for compatibility. `FormDataValidator.Validate(string, ...)` still returns schema-validation errors instead of throwing: it runs `SchemaValidator.Validate` and delegates to a shared private validator. `Validate(PreparedFormSchema, ...)` delegates with `prepared.Validation` and skips reparsing. `IFormValueDerivationService` and `FormValueDerivationService` gain a `PreparedFormSchema` overload; the string overload delegates to it and the prepared overload clones `prepared.Schema` before mutation.

- [ ] **Step 6: Split and lazily yield parsed rows**

Move `ExcelParsedRow` to its own file. Change:

```csharp
IEnumerable<ExcelParsedRow> Parse(string schemaJson, byte[] xlsxBytes);
```

Implement `Parse` as an iterator with `yield return` inside the workbook lifetime. Do not allocate a result list and do not call `ToList()`.

- [ ] **Step 7: Verify and commit**

```bash
dotnet test tests/BuildFlow.Workflow.Tests/BuildFlow.Workflow.Tests.csproj --no-restore --filter "FullyQualifiedName~PreparedFormSchemaTests|FullyQualifiedName~LazyExcelRowTests|FullyQualifiedName~FormDataValidatorTests"
git add src/modules/workflow/Application/Form/Schema src/modules/workflow/Application/Form/Derivation src/modules/workflow/Application/BuildFlow/Common/Services tests/BuildFlow.Workflow.Tests/Performance
git commit -m "perf: reuse prepared form schemas"
```

### Task 4: Add The Module-Owned Import Batch Use Case

**Files:**
- Create: `src/modules/workflow/Application/BuildFlow/Imports/Services/ImportBatchItem.cs`
- Create: `src/modules/workflow/Application/BuildFlow/Imports/Services/ImportBatchResult.cs`
- Create: `src/modules/workflow/Application/BuildFlow/Imports/Services/ImportBatchWriter.cs`
- Modify: `src/modules/workflow/Application/WorkflowApplicationModule.cs`
- Create: `tests/BuildFlow.Workflow.Tests/Performance/ImportBatchWriterTests.cs`

- [ ] **Step 1: Write failing batch-writer tests**

Cover:

```csharp
[Fact]
public async Task WritesValidRowsAndReturnsInvalidRowsWithoutPerRowAggregateCalls()

[Fact]
public async Task UsesFullJobAndRowIndexForDeterministicInstanceIds()
```

The first test supplies two valid rows and one parser-invalid row. Assert two `FormData` rows are added, one error is returned, and the recording aggregate service receives one `RefreshBatchAsync` plus one `RefreshAffectedBatchAsync` call. The second test uses row indexes `1` and `65537` and asserts their instance IDs differ, preventing the existing 16-bit collision.

- [ ] **Step 2: Run and verify RED**

```bash
dotnet test tests/BuildFlow.Workflow.Tests/BuildFlow.Workflow.Tests.csproj --no-restore --filter FullyQualifiedName~ImportBatchWriterTests
```

Expected: compilation fails because the batch use case does not exist.

- [ ] **Step 3: Add batch input and result records**

```csharp
public sealed record ImportBatchItem(int RowIndex, ExcelParsedRow Row);

public sealed record ImportBatchResult(
    int SuccessCount,
    IReadOnlyList<ImportRowError> Errors);
```

- [ ] **Step 4: Implement `ImportBatchWriter`**

Inject `WorkflowDbContext`, `IFormValueDerivationService`, and `IFormAggregateRefreshService`. For each item:

1. Convert parser errors directly to `ImportRowError`.
2. Derive with the prepared schema.
3. Validate with the prepared schema.
4. Create `FormData` with a deterministic instance ID.
5. Convert row-level exceptions to `ImportRowError` while propagating cancellation.

Generate the ID with SHA-256 over all 16 job-ID bytes plus a four-byte little-endian row index:

```csharp
Span<byte> input = stackalloc byte[20];
jobId.TryWriteBytes(input[..16]);
BinaryPrimitives.WriteInt32LittleEndian(input[16..], rowIndex);
Span<byte> hash = stackalloc byte[32];
SHA256.HashData(input, hash);
return new Guid(hash[..16]);
```

Add all successful rows, call `SaveChangesAsync` once, then call each aggregate batch method once with distinct instance IDs. Return success and error counts. Register the concrete writer as scoped.

- [ ] **Step 5: Verify and commit**

```bash
dotnet test tests/BuildFlow.Workflow.Tests/BuildFlow.Workflow.Tests.csproj --no-restore --filter FullyQualifiedName~ImportBatchWriterTests
dotnet build src/modules/workflow/Application/Workflow.Application.csproj --no-restore
git add src/modules/workflow/Application/BuildFlow/Imports/Services src/modules/workflow/Application/WorkflowApplicationModule.cs tests/BuildFlow.Workflow.Tests/Performance/ImportBatchWriterTests.cs
git commit -m "feat: add transactional import batch writer"
```

### Task 5: Process Imports In Bounded Transactional Batches

**Files:**
- Modify: `src/modules/workflow/Application/Reliability/WorkflowJobOptions.cs`
- Modify: `src/modules/workflow/Application/BuildFlow/Imports/Commands/ProcessAvailableImportJobsCommand.cs`
- Modify: `src/modules/workflow/Application/BuildFlow/Imports/Handlers/ProcessAvailableImportJobsCommandHandler.cs`
- Modify: `src/modules/workflow/Application/BuildFlow/Imports/Services/IImportJobProcessor.cs`
- Modify: `src/modules/workflow/Application/BuildFlow/Imports/Services/ImportJobProcessor.cs`
- Modify: `src/BuildFlow.Host/Services/BackgroundServices/ImportJobProcessorService.cs`
- Modify: `src/BuildFlow.Host/BuildFlowHostModule.cs`
- Modify: `src/BuildFlow.Host/appsettings.json`
- Modify: `tests/BuildFlow.Workflow.Tests/Reliability/ImportJobClaimTests.cs`
- Modify: `tests/BuildFlow.Workflow.Tests/Reliability/ImportJobRecoveryTests.cs`
- Create: `tests/BuildFlow.Workflow.Tests/Performance/CountingTransactionInterceptor.cs`
- Create: `tests/BuildFlow.Workflow.Tests/Performance/ImportJobBatchingTests.cs`

- [ ] **Step 1: Write failing batch transaction tests**

Use real PostgreSQL and a counting `DbTransactionInterceptor`. Cover 401 valid rows with `rowsPerBatch = 200`; assert 401 `FormData`, `SuccessRows = 401`, final success, and exactly three batch commits plus one completion commit. Add a cancellation test that cancels before row 201 and asserts exactly 200 rows/cursor values remain committed and the job returns to `Pending`.

- [ ] **Step 2: Update existing tests to the desired signatures**

The command becomes:

```csharp
public sealed record ProcessAvailableImportJobsCommand(
    string LeaseOwner,
    int BatchSize,
    TimeSpan LeaseDuration,
    int MaxAttempts,
    int RowsPerBatch) : IRequest<int>;
```

The processor becomes:

```csharp
Task ProcessAsync(
    Guid jobId,
    string leaseOwner,
    TimeSpan leaseDuration,
    int rowsPerBatch,
    CancellationToken ct);
```

- [ ] **Step 3: Run and verify RED**

```bash
dotnet test tests/BuildFlow.Workflow.Tests/BuildFlow.Workflow.Tests.csproj --no-restore --filter "FullyQualifiedName~ImportJobBatchingTests|FullyQualifiedName~ImportJobRecoveryTests|FullyQualifiedName~ImportJobClaimTests"
```

Expected: compilation fails for the new batch-size signatures and missing batch implementation.

- [ ] **Step 4: Add bounded configuration**

Add `RowsPerBatch = 200` to `WorkflowJobOptions.ImportJobSection` and `Workflow:Jobs:Import:RowsPerBatch` to appsettings. Extend startup validation with:

```csharp
options.Import.RowsPerBatch is > 0 and <= 1000
```

Host sends the configured value in `ProcessAvailableImportJobsCommand`; the handler passes it to the processor.

- [ ] **Step 5: Refactor the processor around a lazy batch buffer**

Inject `ImportBatchWriter`. Load the claimed job and target form once, create one `PreparedFormSchema`, and enumerate `excelService.Parse(...)` once. Skip the already committed cursor without materializing rows. Accumulate at most `rowsPerBatch` `ImportBatchItem` values.

For each full or final batch:

1. Begin a transaction.
2. Lock and revalidate the owned Running job.
3. Call `ImportBatchWriter.WriteAsync` inside that transaction.
4. Increment `SuccessRows` and `FailedRows` from the batch result.
5. Append batch errors, set `TotalRows` to at least the enumerated row count, renew lease/heartbeat, save, and commit once.
6. Clear the buffer.

After enumeration, reject `committedCursor > enumeratedRowCount`, then complete the job in its final transaction and set exact `TotalRows`. Preserve the second-batch cancellation and task-failure cleanup behavior. Delete manual construction of `SubmitFormDataCommandHandler`, `FormValueDerivationService`, and `FormAggregateRefreshService`.

- [ ] **Step 6: Run import suites and architecture scan**

```bash
dotnet test tests/BuildFlow.Workflow.Tests/BuildFlow.Workflow.Tests.csproj --no-restore --filter "FullyQualifiedName~ImportJobBatchingTests|FullyQualifiedName~ImportJobRecoveryTests|FullyQualifiedName~ImportJobClaimTests|FullyQualifiedName~ImportBatchWriterTests"
rg -n "new SubmitFormDataCommandHandler|new FormValueDerivationService|new FormAggregateRefreshService|\.ToList\(\)" src/modules/workflow/Application/BuildFlow/Imports/Services/ImportJobProcessor.cs
```

Expected: tests pass and the scan has no output.

- [ ] **Step 7: Commit import batching**

```bash
git add src/modules/workflow/Application/Reliability/WorkflowJobOptions.cs src/modules/workflow/Application/BuildFlow/Imports src/BuildFlow.Host/Services/BackgroundServices/ImportJobProcessorService.cs src/BuildFlow.Host/BuildFlowHostModule.cs src/BuildFlow.Host/appsettings.json tests/BuildFlow.Workflow.Tests/Reliability tests/BuildFlow.Workflow.Tests/Performance
git commit -m "perf: persist imports in bounded batches"
```

### Task 6: Prove Monitoring N-Query And Concurrency Failures

**Files:**
- Create: `tests/BuildFlow.Monitoring.Tests/Sync/PostgresMonitoringCollection.cs`
- Create: `tests/BuildFlow.Monitoring.Tests/Sync/PostgresMonitoringFixture.cs`
- Create: `tests/BuildFlow.Monitoring.Tests/Sync/CountingUpsertInterceptor.cs`
- Create: `tests/BuildFlow.Monitoring.Tests/Sync/RecordingLogger.cs`
- Create: `tests/BuildFlow.Monitoring.Tests/Sync/MonitoringUpsertTests.cs`

- [ ] **Step 1: Add an isolated PostgreSQL fixture**

Mirror the Workflow reliability fixture using `BUILDFLOW_MONITORING_TEST_CONNECTION` or `Host=127.0.0.1;Port=55432;Username=buildflow;Password=buildflow;Database=postgres`. Create a unique database, call `EnsureCreatedAsync`, attach optional interceptors and an audit stub, disable collection parallelism, terminate connections, and drop the database in disposal.

- [ ] **Step 2: Write the N-query test**

Run a connector returning 1,001 records through `SyncProcessor`. Count commands that target `b_synced_records`. Assert the desired behavior is three parameterized `INSERT ... ON CONFLICT DO UPDATE` statements, final row count 1,001, and no per-record SELECT.

- [ ] **Step 3: Write the concurrent UPSERT test**

Create two contexts/processors that synchronize the same `(ExternalSystemId, ExternalId, RecordType)` concurrently. Assert both jobs succeed, only one record remains, and the row contains one completed job's `SyncJobId`, `OccurredAt`, and payload without unique-key exceptions.

- [ ] **Step 4: Write sequential update and invalid JSON tests**

First sync an existing record, then sync a newer version and assert `SyncJobId`, `OccurredAt`, and payload are replaced. Send invalid JSON and assert payload contains `_raw` while `RecordingLogger` captured one warning that contains identifiers but not raw payload text.

- [ ] **Step 5: Run and verify RED**

```bash
dotnet test tests/BuildFlow.Monitoring.Tests/BuildFlow.Monitoring.Tests.csproj --no-restore --filter FullyQualifiedName~MonitoringUpsertTests
```

Expected: the N-query assertion fails, and the concurrent test fails with a unique-key race or failed sync job.

### Task 7: Implement Parameterized Monitoring UPSERT Batches

**Files:**
- Modify: `src/modules/monitoring/Application/Sync/SyncProcessor.cs`
- Test: `tests/BuildFlow.Monitoring.Tests/Sync/MonitoringUpsertTests.cs`

- [ ] **Step 1: Build normalized record values once**

Convert each `SyncedRecordData` to a `SyncedRecordEntity`-shaped value with a client-generated ID, parsed `RecordData`, UTC audit timestamps, `CreatedBy/UpdatedBy = "system"`, and `OperatorIP = "system"`. On `JsonException`, log:

```csharp
logger.LogWarning(
    "同步记录 JSON 无效，保留原始内容。系统 {SystemId}，外部记录 {ExternalId}，类型 {RecordType}",
    systemId,
    data.ExternalId,
    data.RecordType);
```

Do not log `RawJson`.

- [ ] **Step 2: Implement batches of at most 500**

For each `records.Chunk(500)`, build one multi-row command whose values are all named parameters. The SQL shape is:

```sql
INSERT INTO b_synced_records (
    id, external_system_id, sync_job_id, external_id, record_type,
    occurred_at, json_data, created_by, created_at, updated_by,
    updated_at, is_deleted, operator_ip)
VALUES (...), (...)
ON CONFLICT (external_system_id, external_id, record_type)
DO UPDATE SET
    sync_job_id = EXCLUDED.sync_job_id,
    occurred_at = EXCLUDED.occurred_at,
    json_data = EXCLUDED.json_data,
    updated_by = EXCLUDED.updated_by,
    updated_at = EXCLUDED.updated_at,
    operator_ip = EXCLUDED.operator_ip
```

Create every parameter through `DbCommand.CreateParameter`; never interpolate `ExternalId`, `RecordType`, JSON, or timestamps into SQL.

- [ ] **Step 3: Make the successful persistence boundary atomic**

After the connector fetch succeeds, begin a database transaction. Execute all UPSERT batches, update the tracked `SyncJobEntity` to `Succeeded`, update `ExternalSystemEntity.Data.LastSyncedAt/LastSyncJobId`, call `SaveChangesAsync` once, and commit. Evaluate alerts only after the transaction commits. On failure, roll back, reload the tracked job and external system so rolled-back success metadata cannot leak through a later save, then record the job as `Failed` with the existing bounded error message.

- [ ] **Step 4: Run monitoring tests and build**

```bash
dotnet test tests/BuildFlow.Monitoring.Tests/BuildFlow.Monitoring.Tests.csproj --no-restore --filter FullyQualifiedName~MonitoringUpsertTests
dotnet test tests/BuildFlow.Monitoring.Tests/BuildFlow.Monitoring.Tests.csproj --no-restore
dotnet build src/modules/monitoring/Application/Monitoring.Application.csproj --no-restore
```

Expected: all tests pass and build has 0 warnings and 0 errors.

- [ ] **Step 5: Commit Monitoring UPSERT**

```bash
git add src/modules/monitoring/Application/Sync/SyncProcessor.cs tests/BuildFlow.Monitoring.Tests/Sync
git commit -m "perf: batch monitoring record upserts"
```

### Task 8: Run The Third-Batch Verification Gate

**Files:**
- Verify only.

- [ ] **Step 1: Run focused performance suites**

```bash
dotnet test tests/BuildFlow.Workflow.Tests/BuildFlow.Workflow.Tests.csproj --no-restore --filter "FullyQualifiedName~Performance|FullyQualifiedName~ImportJob|FullyQualifiedName~FormAggregate"
dotnet test tests/BuildFlow.Monitoring.Tests/BuildFlow.Monitoring.Tests.csproj --no-restore --filter FullyQualifiedName~MonitoringUpsertTests
```

- [ ] **Step 2: Run affected projects**

```bash
dotnet test tests/BuildFlow.Workflow.Tests/BuildFlow.Workflow.Tests.csproj --no-restore
dotnet test tests/BuildFlow.Monitoring.Tests/BuildFlow.Monitoring.Tests.csproj --no-restore
dotnet test tests/BuildFlow.Architecture.Tests/BuildFlow.Architecture.Tests.csproj --no-restore
```

- [ ] **Step 3: Run the full solution gate**

```bash
dotnet test BuildFlow.slnx --no-restore
dotnet build BuildFlow.slnx --no-restore
```

Expected: every project passes, zero skipped, and build has 0 warnings and 0 errors.

- [ ] **Step 4: Run mandatory static scans**

```bash
rg -n "CalculateAsync\(|new SubmitFormDataCommandHandler|new FormValueDerivationService|new FormAggregateRefreshService" src/modules/workflow/Application/Form/Aggregates src/modules/workflow/Application/BuildFlow/Imports
rg -n "FirstOrDefaultAsync\(.*ExternalId|foreach \(var data in records\)" src/modules/monitoring/Application/Sync/SyncProcessor.cs
rg -n "INSERT INTO b_synced_records" src/modules/monitoring/Application/Sync/SyncProcessor.cs
git diff --check
git status --short
```

Expected: the first two scans have no output, the UPSERT scan finds exactly the bounded command, diff check has no output, and the worktree is clean.

- [ ] **Step 5: Record batch evidence**

Report aggregate SELECT count for the fixed large input, import transaction count for 401 rows, committed cursor after cancellation, deterministic ID collision test, Monitoring UPSERT command count for 1,001 rows, concurrent duplicate result, affected-project totals, full-solution totals, build warnings/errors, static scan results, and all third-batch commit hashes.

Do not start the fourth-batch chat/file/Excel I/O plan until every command above is green.

## Self-Review Results

- Spec coverage: sections 6.1, 6.2, and 6.3 each have RED tests, implementation tasks, deterministic performance gates, and commits.
- Batch boundary: ClosedXML still owns workbook memory in this batch; fourth batch replaces it with streaming OpenXML and adds byte/row/cell limits.
- Recovery contract: the import cursor remains `SuccessRows + FailedRows`, and each batch's form rows plus cursor update share one PostgreSQL transaction.
- Type consistency: `RowsPerBatch` flows from options to Host command, command handler, processor interface, and processor implementation.
- Placeholder scan: no deferred implementation markers or unnamed error-handling steps remain.