diff --git a/STRESS_TEST_OBSERVABILITY.md b/STRESS_TEST_OBSERVABILITY.md
new file mode 100644
index 000000000..2c7325675
--- /dev/null
+++ b/STRESS_TEST_OBSERVABILITY.md
@@ -0,0 +1,160 @@
+# Stress Testing `reporting-pipeline-service`: What to Watch
+
+Context: load-testing the pipeline (Debezium → Kafka → `reporting-pipeline-service` →
+Kafka Connect JDBC sink → SQL Server) to check for data loss from insufficient
+concurrency guard rails. No standalone Prometheus is deployed; Rancher is the
+container platform. This documents what's already available in the existing
+toolset, grounded in the current code/config.
+
+## 1. The service already exports Prometheus metrics — no Prometheus server required to use them
+
+`reporting-pipeline-service/src/main/resources/application.yaml` sets:
+
+```yaml
+management:
+  endpoints:
+    web:
+      exposure:
+        include: health,metrics,prometheus,liquibase,lag
+  prometheus:
+    metrics:
+      export:
+        enabled: true
+```
+
+So the running service (port `8095`) already serves `GET /actuator/prometheus`
+in plain-text exposition format. You can scrape it by hand during the load run:
+
+```bash
+watch -n2 'curl -s localhost:8095/actuator/prometheus | grep -E "hikaricp_connections|jvm_memory_used|jvm_threads_live"'
+```
+
+Or stand up a throwaway Prometheus + Grafana pointed at it and reuse the
+dashboards **already checked into the repo**:
+`reporting-pipeline-service/src/main/resources/grafana-dashboard/*.json`
+(`person`, `organization`, `investigation`, `observation`, `ldfdata`,
+`postprocessing`). They're wired to an `aws-prometheus` datasource UID for
+prod — swap that UID for a local one and the panels render as-is.
+
+### Metrics that matter most for the concurrency hypothesis
+
+- **`hikaricp_connections_active` / `hikaricp_connections_idle` /
+  `hikaricp_connections_pending` / `hikaricp_connections_timeout_total`** —
+  the likely smoking gun. No `spring.datasource.hikari.maximum-pool-size` is
+  set anywhere, so it runs the Spring Boot default (10 connections). Every
+  domain service spins up its **own** `Executors.newFixedThreadPool` doing
+  synchronous JDBC/stored-proc calls against that one shared pool:
+
+  | Service | Executors |
+  |---|---|
+  | `PersonService` | `rtrExecutor` (`nproc*2`), `prsExecutor` (`featureFlag.thread-pool-size`) |
+  | `OrganizationService` | `rtrExecutor` (`nproc*2`), `org` pool (`thread-pool-size`) |
+  | `InvestigationService` | `phc` pool (`nProc*2`), `inv` pool (`thread-pool-size`) |
+  | `ObservationService` | `obs` pool (`thread-pool-size`) |
+  | `LdfDataService` | `ldf` pool (`thread-pool-size`) |
+  | `ProcessDatamartData` | `dynDmExecutor` (`nProc`) |
+
+  If `hikaricp_connections_pending > 0` and `hikaricp_connections_timeout_total`
+  climbs during the load test, that confirms the connection pool as the
+  concurrency bottleneck.
+
+- **`jvm_memory_used_bytes` / `jvm_gc_pause_seconds`** — `Executors.newFixedThreadPool`
+  backs onto an **unbounded** `LinkedBlockingQueue`. There's no bound tying
+  queued work to consumer throughput, so under sustained high volume the queue
+  can grow without limit. A memory climb followed by a crash, with Kafka lag
+  looking fine right up until the crash, is the signature of this.
+
+- **Per-domain counters/timers** — e.g. `person_msg_processed`,
+  `person_msg_success`, `person_msg_failure`, `inv_msg_failure_total`,
+  `post_dm_success`, `post_dm_failure`, `*_msg_processing_seconds`. Diff
+  processed vs. success vs. failure to spot silent drops.
+
+## 2. The custom `/actuator/lag` endpoint
+
+`gov.cdc.nbs.report.pipeline.lag.LagEndpoint` reports real-time backlog for
+both the pipeline consumer group and the Kafka Connect sink group, with a
+per-topic breakdown:
+
+```bash
+curl -s localhost:8095/actuator/lag | jq
+```
+
+Status is `READY` when both groups are caught up, `PROCESSING` otherwise.
+Poll this on an interval during the load test:
+
+- Sink group lag growing while the pipeline group stays caught up → the JDBC
+  sink connector (writing to SQL Server) is the bottleneck.
+- Pipeline group lag growing → the service's own thread pools/DB contention
+  are the bottleneck.
+
+## 3. Kafka / Kafka Connect layer (no extra tooling needed)
+
+- From inside the `kafka` container:
+  ```bash
+  kafka-consumer-groups.sh --bootstrap-server kafka:29092 --describe --group pipeline-consumer-app
+  kafka-consumer-groups.sh --bootstrap-server kafka:29092 --describe --group connect-Kafka-Connect-SqlServer-Sink
+  ```
+  Same data as `/actuator/lag` but per-partition, straight from the source.
+- Kafka Connect REST API for connector/task health:
+  ```bash
+  curl -s localhost:8083/connectors/<name>/status | jq   # JDBC sink
+  curl -s localhost:8085/connectors/<name>/status | jq   # Debezium source
+  ```
+  A task going `FAILED` (not just lagging) is a distinct data-loss mode from
+  consumer-side loss — worth ruling out separately.
+- Retry/DLT topics (from `spring.kafka.dlq.retry-suffix` / `dlq-suffix` in
+  `application.yaml`): consume `*_retry` / `*_dlt` topics directly with
+  `kafka-console-consumer` to see exactly which payloads are dying, correlated
+  with `nrt_dead_letter_log`.
+
+## 4. SQL Server side — the three tables plus DMVs
+
+- **`nrt_dead_letter_log`** — columns: `origin_topic`, `payload_key`,
+  `payload`, `original_consumer_group`, `exception_stack_trace`,
+  `exception_message`, `exception_fqcn`, `exception_cause_fqcn`,
+  `received_at`. Query the rate of growth during the load window
+  (`COUNT(*) GROUP BY DATEPART(minute, received_at)`), and group by
+  `exception_fqcn` — failures clustering around one exception type (timeout
+  vs. deadlock vs. constraint violation) points to very different root causes.
+- **`nrt_backfill`** — columns: `entity`, `record_uid_list`, `batch_id`,
+  `err_description`, `status_cd`, `retry_count`. Watch for rows stuck at high
+  `retry_count` with a non-terminal `status_cd` — that's the backfill-retry
+  path failing to keep up, a second concurrency-guardrail symptom distinct
+  from the DLQ.
+- **`job_flow_log`** — correlate stored-proc start/end/failure timestamps
+  against the HikariCP pending/timeout spikes above. If proc executions start
+  queuing or timing out at the same moment `hikaricp_connections_pending`
+  spikes, that directly implicates the shared connection pool.
+- **SQL Server DMVs** — none of the above show lock contention directly:
+  ```sql
+  SELECT * FROM sys.dm_exec_requests;
+  SELECT * FROM sys.dm_os_waiting_tasks;
+  SELECT * FROM sys.dm_exec_session_wait_stats
+    WHERE wait_type IN ('THREADPOOL','LCK_M_X','RESOURCE_SEMAPHORE');
+  ```
+  These distinguish connection-pool starvation, row/table lock contention
+  (plausible since multiple threads across multiple executors may hit the
+  same stored procs concurrently for related entities), and SQL Server
+  worker-thread starvation.
+
+## 5. Rancher (in place of Prometheus/Grafana-as-infra)
+
+Rancher's built-in cluster monitoring UI (project/workload metrics view) gives
+per-pod CPU, memory, and **restart count** without a dedicated Prometheus.
+Watch `reporting-pipeline-service`'s memory graph and restart count during the
+load test:
+
+- Restart correlated with a memory climb → corroborates the unbounded-queue
+  hypothesis above.
+- Restart with flat memory but Kafka rebalance activity → points elsewhere,
+  e.g. a health/liveness probe failing due to blocked DB calls.
+
+## Suggested sequence
+
+1. Watch `/actuator/lag` and Rancher pod memory/restarts live — cheapest signals.
+2. If memory climbs unbounded, pull `/actuator/prometheus` for
+   `hikaricp_connections_pending` / `hikaricp_connections_timeout_total` to
+   confirm pool exhaustion is the trigger.
+3. Cross-reference timestamps against `nrt_dead_letter_log` and
+   `job_flow_log` to identify which entity/stored-proc path is actually
+   dropping or failing data.
diff --git a/docker-compose.yaml b/docker-compose.yaml
index b25f8eab3..75216fb30 100644
--- a/docker-compose.yaml
+++ b/docker-compose.yaml
@@ -1,6 +1,6 @@
 services:
   nbs-mssql:
-    image: ghcr.io/cdcent/nedssdb:6.0.17.0-replacement-beta
+    image: ghcr.io/cdcent/nedssdb:latest
     platform: linux/amd64
     environment:
       - DATABASE_VERSION=6.0.18.1
diff --git a/reporting-pipeline-service/README.md b/reporting-pipeline-service/README.md
index 03dcc2f7b..82ee5e932 100644
--- a/reporting-pipeline-service/README.md
+++ b/reporting-pipeline-service/README.md
@@ -7,18 +7,33 @@ If you don't already have a local application config create one using the existi
 cp src/main/resources/application.yaml src/main/resources/application-local.yaml
 ```
 
-Update `src/main/resources/application-local.yaml` with the following values:
+Create a `src/main/resources/application-local.yaml` file. Sample below:
 
 ```yaml
 spring:
   datasource:
-    password: reporting_pipeline_service
-    username: reporting_pipeline_service_rdb
+    password: PizzaIsGood33!
+    username: sa
     url: jdbc:sqlserver://localhost:3433;databaseName=RDB_MODERN;encrypt=true;trustServerCertificate=true;
   kafka:
     bootstrap-servers: ${KAFKA_BOOTSTRAP_SERVER:localhost:9092}
+  kafka-connect:
+    url: http://localhost:8083
+
+  liquibase:
+    enabled: true
+    change-log: classpath:db/changelog/db.changelog-master.yaml
+    user: sa
+    password: PizzaIsGood33!
+
+  featureFlag:
+    person-service-direct-write: true
 ```
 
+Stop the container if it is running:
+```shell
+docker stop nedss-datareporting-pipeline-service-1
+```
 Run `reporting-pipeline-service` using the following Gradle command:
 
 ```shell
diff --git a/reporting-pipeline-service/src/main/java/gov/cdc/nbs/report/pipeline/person/ARCHITECTURE.md b/reporting-pipeline-service/src/main/java/gov/cdc/nbs/report/pipeline/person/ARCHITECTURE.md
new file mode 100644
index 000000000..a695a41ec
--- /dev/null
+++ b/reporting-pipeline-service/src/main/java/gov/cdc/nbs/report/pipeline/person/ARCHITECTURE.md
@@ -0,0 +1,102 @@
+# Person Event Processing — Conditional Data Flow
+
+Describes how `PersonService` routes Patient/Provider/Auth User CDC events, including the
+`person-service-direct-write` branch and how both paths converge on the shared postprocessing
+pipeline.
+
+```mermaid
+%%{init: {'theme': 'base', 'themeVariables': {
+  'primaryColor': '#16273d',
+  'primaryTextColor': '#d7e3f0',
+  'primaryBorderColor': '#3d5a7a',
+  'lineColor': '#7d93ad',
+  'secondaryColor': '#16273d',
+  'tertiaryColor': '#16273d',
+  'fontFamily': 'ui-monospace, SFMono-Regular, Consolas, monospace',
+  'fontSize': '15px'
+}}}%%
+flowchart TD
+    subgraph SRC["CDC Source"]
+        ODSEPerson[(ODSE.Person)]
+        ODSEAuthUser[(ODSE.Auth_user)]
+    end
+
+    ODSEPerson -->|Debezium CDC| TopicPerson[["nbs_Person"]]
+    ODSEAuthUser -->|Debezium CDC| TopicAuthUser[["nbs_Auth_user"]]
+
+    TopicPerson --> PS["PersonService.processMessage()"]
+    TopicAuthUser --> PS
+
+    PS --> TopicSwitch{"which topic?"}
+    TopicSwitch -->|nbs_Person| CdSwitch{"cd field"}
+    TopicSwitch -->|nbs_Auth_user| ComputeAuthUser["sp_auth_user_event"]
+
+    CdSwitch -->|PAT| ComputePatient["sp_Patient_Event"]
+    CdSwitch -->|PRV| ComputeProvider["sp_provider_event"]
+
+    subgraph DIRECTWRITE["Conditional: person-service-direct-write"]
+        DW{"directWrite flag"}
+        DW -->|true| JpaSave["JPA save<br/>→ nrt_patient / nrt_provider / nrt_auth_user"]
+        DW -->|false| KafkaPublish["Publish enriched JSON<br/>→ nrt.patient / nrt.provider / nrt.auth-user"]
+        KafkaPublish --> KafkaConnect["Kafka-Connect JDBC Sink<br/>upserts nrt_* table"]
+        KafkaPublish --> PPListener["PostProcessingService<br/>@KafkaListener on nrt.* topic"]
+    end
+
+    ComputePatient --> DW
+    ComputeProvider --> DW
+    ComputeAuthUser --> DW
+
+    JpaSave --> Enqueue["PostProcessingService.enqueue(topic, uid)<br/>(direct in-process call — no Kafka round-trip)"]
+    PPListener --> ProcessNrt["processNrtMessage()<br/>parses Kafka payload"]
+    ProcessNrt --> Enqueue
+
+    subgraph POSTPROC["Shared postprocessing pipeline (priority-ordered batch)"]
+        Enqueue --> IdCache[("idCache<br/>keyed by topic<br/>producer adds + drain snapshot/clear both under cacheLock")]
+        IdCache -->|"@Scheduled processCachedIds()<br/>sorted by Entity.priority"| Drain["Priority-ordered SP execution"]
+        Drain --> PatientSP["sp_nrt_patient_postprocessing<br/>→ D_PATIENT"]
+        Drain --> ProviderSP["sp_nrt_provider_postprocessing<br/>→ D_PROVIDER"]
+        Drain --> AuthUserSP["sp_user_profile_postprocessing<br/>→ USER_PROFILE"]
+        Drain -.->|"same batch,<br/>later priority"| InvSP["Investigation / case_management<br/>postprocessing (unrelated to Person, unchanged)"]
+    end
+
+    PatientSP -->|"returns Covid_*_Datamart rows<br/>(empty for Provider/AuthUser happy path)"| DmData["DatamartData"]
+
+    subgraph DATAMART["Datamart routing"]
+        DmData --> DatamartTopic[["nbs_Datamart"]]
+        DatamartTopic --> DmCache[("dmCache")]
+        DmCache -->|"@Scheduled processDatamartIds()"| DatamartSP["Datamart-specific stored procs<br/>e.g. sp_covid_case_datamart_postprocessing"]
+    end
+
+    InvSP -.->|"LEFT JOIN D_PATIENT<br/>(ordering-dependent)"| DatamartSP2["sp_std_hiv_datamart_postprocessing"]
+
+    classDef highlight fill:#3a2a12,stroke:#e7a53d,color:#f4dcae,stroke-width:2px;
+    classDef muted fill:#16273d,stroke:#48607e,color:#6c839c,stroke-dasharray: 3 3;
+    class DW,JpaSave highlight;
+    class KafkaPublish,KafkaConnect,PPListener,ProcessNrt muted;
+```
+
+🟧 amber = `person-service-direct-write` (active path) · ⬜ dashed = Kafka-Connect (legacy, feature-flagged off)
+
+## Key conditionals
+
+- **`cd` field** routes `nbs_Person` events to patient vs. provider stored procs.
+- **`person-service-direct-write` flag** is the main fork: JPA save (direct-write) vs. Kafka
+  publish → Kafka-Connect (legacy). Both paths converge on the same
+  `PostProcessingService.enqueue()` call into the shared `idCache` — direct-write calls it
+  in-process, the legacy path via its existing `@KafkaListener` → `processNrtMessage()`, which
+  itself now delegates to `enqueue()` for id-caching.
+- **Priority-ordered batch drain** is what keeps `D_PATIENT` hydration ahead of
+  investigation/case_management processing within the same cycle. Direct-write must go through
+  this same shared pipeline rather than calling postprocessing stored procedures itself —
+  bypassing it breaks that ordering guarantee (see APP-787).
+- **`cacheLock`** guards every producer-side cache write (`enqueue()` and the legacy path's
+  payload-enrichment writes) against the scheduled drain's snapshot-then-clear. Without it, an
+  add landing between the drain's snapshot and its `clear()` was silently and permanently
+  dropped rather than merely delayed — this affected every entity type processed by this shared
+  service, not just Patient/Provider/AuthUser (see APP-787 concurrency follow-up).
+- **Datamart routing** only actually carries rows in the happy path for Patient (Covid
+  datamarts); Provider/AuthUser postprocessing stored procedures return empty result sets there.
+
+Not shown (orthogonal, non-blocking flags that don't affect this core routing):
+`elasticSearchEnable` (publishes to `elastic_search_patient`/`elastic_search_provider` regardless
+of direct-write) and `phcDatamartEnable` (async PHC fact datamart update).
diff --git a/reporting-pipeline-service/src/main/java/gov/cdc/nbs/report/pipeline/person/model/entity/NrtAuthUser.java b/reporting-pipeline-service/src/main/java/gov/cdc/nbs/report/pipeline/person/model/entity/NrtAuthUser.java
new file mode 100644
index 000000000..738892ef5
--- /dev/null
+++ b/reporting-pipeline-service/src/main/java/gov/cdc/nbs/report/pipeline/person/model/entity/NrtAuthUser.java
@@ -0,0 +1,77 @@
+package gov.cdc.nbs.report.pipeline.person.model.entity;
+
+import static gov.cdc.nbs.report.pipeline.util.UtilHelper.parseDateTime;
+
+import gov.cdc.nbs.report.pipeline.person.model.dto.user.AuthUser;
+import jakarta.persistence.Column;
+import jakarta.persistence.Entity;
+import jakarta.persistence.Id;
+import jakarta.persistence.Table;
+import java.time.LocalDateTime;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+@Entity
+@Table(name = "nrt_auth_user")
+@Data
+@NoArgsConstructor
+public class NrtAuthUser {
+  @Id
+  @Column(name = "auth_user_uid")
+  private Long authUserId;
+
+  @Column(name = "user_id")
+  private String userId;
+
+  @Column(name = "first_nm")
+  private String firstNm;
+
+  @Column(name = "last_nm")
+  private String lastNm;
+
+  @Column(name = "nedss_entry_id")
+  private Long nedssEntryId;
+
+  @Column(name = "provider_uid")
+  private Long providerUid;
+
+  @Column(name = "add_time")
+  private LocalDateTime addTime;
+
+  @Column(name = "add_user_id")
+  private Long addUserId;
+
+  @Column(name = "last_chg_time")
+  private LocalDateTime lastChgTime;
+
+  @Column(name = "last_chg_user_id")
+  private Long lastChgUserId;
+
+  @Column(name = "record_status_cd")
+  private String recordStatusCd;
+
+  @Column(name = "record_status_time")
+  private LocalDateTime recordStatusTime;
+
+  // GENERATED ALWAYS AS ROW START: SQL Server populates this column; it must never be
+  // included in the INSERT/UPDATE column list.
+  @Column(name = "refresh_datetime", insertable = false, updatable = false)
+  private LocalDateTime refreshDatetime;
+
+  public static NrtAuthUser from(AuthUser authUser) {
+    NrtAuthUser nrtAuthUser = new NrtAuthUser();
+    nrtAuthUser.setAuthUserId(authUser.getAuthUserUid());
+    nrtAuthUser.setUserId(authUser.getUserId());
+    nrtAuthUser.setFirstNm(authUser.getFirstNm());
+    nrtAuthUser.setLastNm(authUser.getLastNm());
+    nrtAuthUser.setNedssEntryId(authUser.getNedssEntryId());
+    nrtAuthUser.setProviderUid(authUser.getProviderUid());
+    nrtAuthUser.setAddTime(parseDateTime(authUser.getAddTime()));
+    nrtAuthUser.setAddUserId(authUser.getAddUserId());
+    nrtAuthUser.setLastChgTime(parseDateTime(authUser.getLastChgTime()));
+    nrtAuthUser.setLastChgUserId(authUser.getLastChgUserId());
+    nrtAuthUser.setRecordStatusCd(authUser.getRecordStatusCd());
+    nrtAuthUser.setRecordStatusTime(parseDateTime(authUser.getRecordStatusTime()));
+    return nrtAuthUser;
+  }
+}
diff --git a/reporting-pipeline-service/src/main/java/gov/cdc/nbs/report/pipeline/person/model/entity/NrtPatient.java b/reporting-pipeline-service/src/main/java/gov/cdc/nbs/report/pipeline/person/model/entity/NrtPatient.java
new file mode 100644
index 000000000..990279c53
--- /dev/null
+++ b/reporting-pipeline-service/src/main/java/gov/cdc/nbs/report/pipeline/person/model/entity/NrtPatient.java
@@ -0,0 +1,392 @@
+package gov.cdc.nbs.report.pipeline.person.model.entity;
+
+import static gov.cdc.nbs.report.pipeline.util.UtilHelper.parseBigDecimal;
+import static gov.cdc.nbs.report.pipeline.util.UtilHelper.parseDateTime;
+
+import gov.cdc.nbs.report.pipeline.person.model.dto.patient.PatientReporting;
+import jakarta.persistence.Column;
+import jakarta.persistence.Entity;
+import jakarta.persistence.Id;
+import jakarta.persistence.Table;
+import java.math.BigDecimal;
+import java.time.LocalDateTime;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+/** JPA entity mirroring the nrt_patient table for the person-service direct-write path. */
+@Entity
+@Table(name = "nrt_patient")
+@Data
+@NoArgsConstructor
+public class NrtPatient {
+  @Id
+  @Column(name = "patient_uid")
+  private Long patientUid;
+
+  @Column(name = "patient_mpr_uid")
+  private Long patientMprUid;
+
+  @Column(name = "record_status")
+  private String recordStatus;
+
+  @Column(name = "local_id")
+  private String localId;
+
+  @Column(name = "general_comments")
+  private String generalComments;
+
+  @Column(name = "first_name")
+  private String firstNm;
+
+  @Column(name = "middle_name")
+  private String middleNm;
+
+  @Column(name = "last_name")
+  private String lastNm;
+
+  @Column(name = "name_suffix")
+  private String nmSuffix;
+
+  @Column(name = "nm_use_cd")
+  private String nmUseCd;
+
+  @Column(name = "status_name_cd")
+  private String statusNameCd;
+
+  @Column(name = "alias_nickname")
+  private String aliasNickname;
+
+  @Column(name = "street_address_1")
+  private String streetAddress1;
+
+  @Column(name = "street_address_2")
+  private String streetAddress2;
+
+  @Column(name = "city")
+  private String city;
+
+  @Column(name = "state")
+  private String state;
+
+  @Column(name = "state_code")
+  private String stateCode;
+
+  @Column(name = "zip")
+  private String zip;
+
+  @Column(name = "county")
+  private String county;
+
+  @Column(name = "county_code")
+  private String countyCode;
+
+  @Column(name = "country")
+  private String country;
+
+  @Column(name = "country_code")
+  private String countryCode;
+
+  @Column(name = "within_city_limits")
+  private String withinCityLimits;
+
+  @Column(name = "phone_home")
+  private String phoneHome;
+
+  @Column(name = "phone_ext_home")
+  private String phoneExtHome;
+
+  @Column(name = "phone_work")
+  private String phoneWork;
+
+  @Column(name = "phone_ext_work")
+  private String phoneExtWork;
+
+  @Column(name = "phone_cell")
+  private String phoneCell;
+
+  @Column(name = "email")
+  private String email;
+
+  @Column(name = "dob")
+  private LocalDateTime dob;
+
+  @Column(name = "age_reported")
+  private BigDecimal ageReported;
+
+  @Column(name = "age_reported_unit")
+  private String ageReportedUnit;
+
+  @Column(name = "age_reported_unit_cd")
+  private String ageReportedUnitCd;
+
+  @Column(name = "birth_sex")
+  private String birthSex;
+
+  @Column(name = "current_sex")
+  private String currentSex;
+
+  @Column(name = "curr_sex_cd")
+  private String currSexCd;
+
+  @Column(name = "deceased_indicator")
+  private String deceasedIndicator;
+
+  @Column(name = "deceased_ind_cd")
+  private String deceasedIndCd;
+
+  @Column(name = "deceased_date")
+  private LocalDateTime deceasedDate;
+
+  @Column(name = "marital_status")
+  private String maritalStatus;
+
+  @Column(name = "marital_status_cd")
+  private String maritalStatusCd;
+
+  @Column(name = "ssn")
+  private String ssn;
+
+  @Column(name = "ethnic_group_ind")
+  private String ethnicGroupInd;
+
+  @Column(name = "ethnicity")
+  private String ethnicity;
+
+  @Column(name = "race_calculated")
+  private String raceCalculated;
+
+  @Column(name = "race_calc_details")
+  private String raceCalcDetails;
+
+  @Column(name = "race_amer_ind_1")
+  private String raceAmerInd1;
+
+  @Column(name = "race_amer_ind_2")
+  private String raceAmerInd2;
+
+  @Column(name = "race_amer_ind_3")
+  private String raceAmerInd3;
+
+  @Column(name = "race_amer_ind_gt3_ind")
+  private String raceAmerIndGt3Ind;
+
+  @Column(name = "race_amer_ind_all")
+  private String raceAmerIndAll;
+
+  @Column(name = "race_asian_1")
+  private String raceAsian1;
+
+  @Column(name = "race_asian_2")
+  private String raceAsian2;
+
+  @Column(name = "race_asian_3")
+  private String raceAsian3;
+
+  @Column(name = "race_asian_gt3_ind")
+  private String raceAsianGt3Ind;
+
+  @Column(name = "race_asian_all")
+  private String raceAsianAll;
+
+  @Column(name = "race_black_1")
+  private String raceBlack1;
+
+  @Column(name = "race_black_2")
+  private String raceBlack2;
+
+  @Column(name = "race_black_3")
+  private String raceBlack3;
+
+  @Column(name = "race_black_gt3_ind")
+  private String raceBlackGt3Ind;
+
+  @Column(name = "race_black_all")
+  private String raceBlackAll;
+
+  @Column(name = "race_nat_hi_1")
+  private String raceNatHi1;
+
+  @Column(name = "race_nat_hi_2")
+  private String raceNatHi2;
+
+  @Column(name = "race_nat_hi_3")
+  private String raceNatHi3;
+
+  @Column(name = "race_nat_hi_gt3_ind")
+  private String raceNatHiGt3Ind;
+
+  @Column(name = "race_nat_hi_all")
+  private String raceNatHiAll;
+
+  @Column(name = "race_white_1")
+  private String raceWhite1;
+
+  @Column(name = "race_white_2")
+  private String raceWhite2;
+
+  @Column(name = "race_white_3")
+  private String raceWhite3;
+
+  @Column(name = "race_white_gt3_ind")
+  private String raceWhiteGt3Ind;
+
+  @Column(name = "race_white_all")
+  private String raceWhiteAll;
+
+  @Column(name = "patient_number")
+  private String patientNumber;
+
+  @Column(name = "patient_number_auth")
+  private String patientNumberAuth;
+
+  @Column(name = "entry_method")
+  private String entryMethod;
+
+  @Column(name = "speaks_english")
+  private String speaksEnglish;
+
+  @Column(name = "unk_ethnic_rsn")
+  private String unkEthnicRsn;
+
+  @Column(name = "curr_sex_unk_rsn")
+  private String currSexUnkRsn;
+
+  @Column(name = "preferred_gender")
+  private String preferredGender;
+
+  @Column(name = "addl_gender_info")
+  private String addlGenderInfo;
+
+  @Column(name = "census_tract")
+  private String censusTract;
+
+  @Column(name = "race_all")
+  private String raceAll;
+
+  @Column(name = "birth_country")
+  private String birthCountry;
+
+  @Column(name = "primary_occupation")
+  private String primaryOccupation;
+
+  @Column(name = "primary_language")
+  private String primaryLanguage;
+
+  @Column(name = "add_user_id")
+  private Long addUserId;
+
+  @Column(name = "add_user_name")
+  private String addUserName;
+
+  @Column(name = "add_time")
+  private LocalDateTime addTime;
+
+  @Column(name = "last_chg_user_id")
+  private Long lastChgUserId;
+
+  @Column(name = "last_chg_user_name")
+  private String lastChgUserName;
+
+  @Column(name = "last_chg_time")
+  private LocalDateTime lastChgTime;
+
+  // GENERATED ALWAYS AS ROW START: SQL Server populates this column; it must never be
+  // included in the INSERT/UPDATE column list.
+  @Column(name = "refresh_datetime", insertable = false, updatable = false)
+  private LocalDateTime refreshDatetime;
+
+  public static NrtPatient from(PatientReporting r) {
+    NrtPatient p = new NrtPatient();
+    p.setPatientUid(r.getPatientUid());
+    p.setPatientMprUid(r.getPatientMprUid());
+    p.setRecordStatus(r.getRecordStatus());
+    p.setLocalId(r.getLocalId());
+    p.setGeneralComments(r.getGeneralComments());
+    p.setFirstNm(r.getFirstNm());
+    p.setMiddleNm(r.getMiddleNm());
+    p.setLastNm(r.getLastNm());
+    p.setNmSuffix(r.getNmSuffix());
+    p.setNmUseCd(r.getNmUseCd());
+    p.setStatusNameCd(r.getStatusNameCd());
+    p.setAliasNickname(r.getAliasNickname());
+    p.setStreetAddress1(r.getStreetAddress1());
+    p.setStreetAddress2(r.getStreetAddress2());
+    p.setCity(r.getCity());
+    p.setState(r.getState());
+    p.setStateCode(r.getStateCode());
+    p.setZip(r.getZip());
+    p.setCounty(r.getCounty());
+    p.setCountyCode(r.getCountyCode());
+    p.setCountry(r.getHomeCountry());
+    p.setCountryCode(r.getCountryCode());
+    p.setWithinCityLimits(r.getWithinCityLimits());
+    p.setPhoneHome(r.getPhoneHome());
+    p.setPhoneExtHome(r.getPhoneExtHome());
+    p.setPhoneWork(r.getPhoneWork());
+    p.setPhoneExtWork(r.getPhoneExtWork());
+    p.setPhoneCell(r.getPhoneCell());
+    p.setEmail(r.getEmail());
+    p.setDob(parseDateTime(r.getDob()));
+    p.setAgeReported(parseBigDecimal(r.getAgeReported()));
+    p.setAgeReportedUnit(r.getAgeReportedUnit());
+    p.setAgeReportedUnitCd(r.getAgeReportedUnitCd());
+    p.setBirthSex(r.getBirthSex());
+    p.setCurrentSex(r.getCurrentSex());
+    p.setCurrSexCd(r.getCurrSexCd());
+    p.setDeceasedIndicator(r.getDeceasedIndicator());
+    p.setDeceasedIndCd(r.getDeceasedIndCd());
+    p.setDeceasedDate(parseDateTime(r.getDeceasedDate()));
+    p.setMaritalStatus(r.getMaritalStatus());
+    p.setMaritalStatusCd(r.getMaritalStatusCd());
+    p.setSsn(r.getSsn());
+    p.setEthnicGroupInd(r.getEthnicGroupInd());
+    p.setEthnicity(r.getEthnicity());
+    p.setRaceCalculated(r.getRaceCalculated());
+    p.setRaceCalcDetails(r.getRaceCalcDetails());
+    p.setRaceAmerInd1(r.getRaceAmerInd1());
+    p.setRaceAmerInd2(r.getRaceAmerInd2());
+    p.setRaceAmerInd3(r.getRaceAmerInd3());
+    p.setRaceAmerIndGt3Ind(r.getRaceAmerIndGt3Ind());
+    p.setRaceAmerIndAll(r.getRaceAmerIndAll());
+    p.setRaceAsian1(r.getRaceAsian1());
+    p.setRaceAsian2(r.getRaceAsian2());
+    p.setRaceAsian3(r.getRaceAsian3());
+    p.setRaceAsianGt3Ind(r.getRaceAsianGt3Ind());
+    p.setRaceAsianAll(r.getRaceAsianAll());
+    p.setRaceBlack1(r.getRaceBlack1());
+    p.setRaceBlack2(r.getRaceBlack2());
+    p.setRaceBlack3(r.getRaceBlack3());
+    p.setRaceBlackGt3Ind(r.getRaceBlackGt3Ind());
+    p.setRaceBlackAll(r.getRaceBlackAll());
+    p.setRaceNatHi1(r.getRaceNatHi1());
+    p.setRaceNatHi2(r.getRaceNatHi2());
+    p.setRaceNatHi3(r.getRaceNatHi3());
+    p.setRaceNatHiGt3Ind(r.getRaceNatHiGt3Ind());
+    p.setRaceNatHiAll(r.getRaceNatHiAll());
+    p.setRaceWhite1(r.getRaceWhite1());
+    p.setRaceWhite2(r.getRaceWhite2());
+    p.setRaceWhite3(r.getRaceWhite3());
+    p.setRaceWhiteGt3Ind(r.getRaceWhiteGt3Ind());
+    p.setRaceWhiteAll(r.getRaceWhiteAll());
+    p.setPatientNumber(r.getPatientNumber());
+    p.setPatientNumberAuth(r.getPatientNumberAuth());
+    p.setEntryMethod(r.getEntryMethod());
+    p.setSpeaksEnglish(r.getSpeaksEnglish());
+    p.setUnkEthnicRsn(r.getUnkEthnicRsn());
+    p.setCurrSexUnkRsn(r.getCurrSexUnkRsn());
+    p.setPreferredGender(r.getPreferredGender());
+    p.setAddlGenderInfo(r.getAddlGenderInfo());
+    p.setCensusTract(r.getCensusTract());
+    p.setRaceAll(r.getRaceAll());
+    p.setBirthCountry(r.getBirthCountry());
+    p.setPrimaryOccupation(r.getPrimaryOccupation());
+    p.setPrimaryLanguage(r.getPrimaryLanguage());
+    p.setAddUserId(r.getAddUserId());
+    p.setAddUserName(r.getAddUserName());
+    p.setAddTime(parseDateTime(r.getAddTime()));
+    p.setLastChgUserId(r.getLastChgUserId());
+    p.setLastChgUserName(r.getLastChgUserName());
+    p.setLastChgTime(parseDateTime(r.getLastChgTime()));
+    return p;
+  }
+}
diff --git a/reporting-pipeline-service/src/main/java/gov/cdc/nbs/report/pipeline/person/model/entity/NrtProvider.java b/reporting-pipeline-service/src/main/java/gov/cdc/nbs/report/pipeline/person/model/entity/NrtProvider.java
new file mode 100644
index 000000000..eb2e4ebac
--- /dev/null
+++ b/reporting-pipeline-service/src/main/java/gov/cdc/nbs/report/pipeline/person/model/entity/NrtProvider.java
@@ -0,0 +1,186 @@
+package gov.cdc.nbs.report.pipeline.person.model.entity;
+
+import static gov.cdc.nbs.report.pipeline.util.UtilHelper.parseDateTime;
+
+import gov.cdc.nbs.report.pipeline.person.model.dto.provider.ProviderReporting;
+import jakarta.persistence.Column;
+import jakarta.persistence.Entity;
+import jakarta.persistence.Id;
+import jakarta.persistence.Table;
+import java.time.LocalDateTime;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+/** JPA entity mirroring the nrt_provider table for the person-service direct-write path. */
+@Entity
+@Table(name = "nrt_provider")
+@Data
+@NoArgsConstructor
+public class NrtProvider {
+  @Id
+  @Column(name = "provider_uid")
+  private Long providerUid;
+
+  @Column(name = "local_id")
+  private String localId;
+
+  @Column(name = "record_status")
+  private String recordStatus;
+
+  @Column(name = "name_prefix")
+  private String nmPrefix;
+
+  @Column(name = "first_name")
+  private String firstNm;
+
+  @Column(name = "middle_name")
+  private String middleNm;
+
+  @Column(name = "last_name")
+  private String lastNm;
+
+  @Column(name = "name_suffix")
+  private String nmSuffix;
+
+  @Column(name = "name_degree")
+  private String nmDegree;
+
+  @Column(name = "general_comments")
+  private String generalComments;
+
+  @Column(name = "quick_code")
+  private String providerQuickCode;
+
+  @Column(name = "provider_registration_num")
+  private String providerRegistrationNum;
+
+  @Column(name = "provider_registration_num_auth")
+  private String providerRegistrationNumAuth;
+
+  @Column(name = "provider_npi")
+  private String providerNpi;
+
+  @Column(name = "street_address_1")
+  private String streetAddress1;
+
+  @Column(name = "street_address_2")
+  private String streetAddress2;
+
+  @Column(name = "city")
+  private String city;
+
+  @Column(name = "state")
+  private String state;
+
+  @Column(name = "state_code")
+  private String stateCode;
+
+  @Column(name = "zip")
+  private String zip;
+
+  @Column(name = "county")
+  private String county;
+
+  @Column(name = "county_code")
+  private String countyCode;
+
+  @Column(name = "country")
+  private String country;
+
+  @Column(name = "country_code")
+  private String countryCode;
+
+  @Column(name = "address_comments")
+  private String addressComments;
+
+  @Column(name = "phone_work")
+  private String phoneWork;
+
+  @Column(name = "phone_ext_work")
+  private String phoneExtWork;
+
+  @Column(name = "phone_comments")
+  private String phoneComments;
+
+  @Column(name = "phone_work_phone")
+  private String phoneWorkPhone;
+
+  @Column(name = "phone_ext_work_phone")
+  private String phoneExtWorkPhone;
+
+  @Column(name = "email_work")
+  private String email;
+
+  @Column(name = "phone_cell")
+  private String phoneCell;
+
+  @Column(name = "entry_method")
+  private String entryMethod;
+
+  @Column(name = "add_user_id")
+  private Long addUserId;
+
+  @Column(name = "add_user_name")
+  private String addUserName;
+
+  @Column(name = "add_time")
+  private LocalDateTime addTime;
+
+  @Column(name = "last_chg_user_id")
+  private Long lastChgUserId;
+
+  @Column(name = "last_chg_user_name")
+  private String lastChgUserName;
+
+  @Column(name = "last_chg_time")
+  private LocalDateTime lastChgTime;
+
+  // GENERATED ALWAYS AS ROW START: SQL Server populates this column; it must never be
+  // included in the INSERT/UPDATE column list.
+  @Column(name = "refresh_datetime", insertable = false, updatable = false)
+  private LocalDateTime refreshDatetime;
+
+  public static NrtProvider from(ProviderReporting r) {
+    NrtProvider p = new NrtProvider();
+    p.setProviderUid(r.getProviderUid());
+    p.setLocalId(r.getLocalId());
+    p.setRecordStatus(r.getRecordStatus());
+    p.setNmPrefix(r.getNmPrefix());
+    p.setFirstNm(r.getFirstNm());
+    p.setMiddleNm(r.getMiddleNm());
+    p.setLastNm(r.getLastNm());
+    p.setNmSuffix(r.getNmSuffix());
+    p.setNmDegree(r.getNmDegree());
+    p.setGeneralComments(r.getGeneralComments());
+    p.setProviderQuickCode(r.getProviderQuickCode());
+    p.setProviderRegistrationNum(r.getProviderRegistrationNum());
+    p.setProviderRegistrationNumAuth(r.getProviderRegistrationNumAuth());
+    p.setProviderNpi(r.getProviderNpi());
+    p.setStreetAddress1(r.getStreetAddress1());
+    p.setStreetAddress2(r.getStreetAddress2());
+    p.setCity(r.getCity());
+    p.setState(r.getState());
+    p.setStateCode(r.getStateCode());
+    p.setZip(r.getZip());
+    p.setCounty(r.getCounty());
+    p.setCountyCode(r.getCountyCode());
+    p.setCountry(r.getCountry());
+    p.setCountryCode(r.getCountryCode());
+    p.setAddressComments(r.getAddressComments());
+    p.setPhoneWork(r.getPhoneWork());
+    p.setPhoneExtWork(r.getPhoneExtWork());
+    p.setPhoneComments(r.getPhoneComments());
+    p.setPhoneWorkPhone(r.getPhoneWorkPhone());
+    p.setPhoneExtWorkPhone(r.getPhoneExtWorkPhone());
+    p.setEmail(r.getEmail());
+    p.setPhoneCell(r.getPhoneCell());
+    p.setEntryMethod(r.getEntryMethod());
+    p.setAddUserId(r.getAddUserId());
+    p.setAddUserName(r.getAddUserName());
+    p.setAddTime(parseDateTime(r.getAddTime()));
+    p.setLastChgUserId(r.getLastChgUserId());
+    p.setLastChgUserName(r.getLastChgUserName());
+    p.setLastChgTime(parseDateTime(r.getLastChgTime()));
+    return p;
+  }
+}
diff --git a/reporting-pipeline-service/src/main/java/gov/cdc/nbs/report/pipeline/person/repository/NrtAuthUserRepository.java b/reporting-pipeline-service/src/main/java/gov/cdc/nbs/report/pipeline/person/repository/NrtAuthUserRepository.java
new file mode 100644
index 000000000..81e0ea067
--- /dev/null
+++ b/reporting-pipeline-service/src/main/java/gov/cdc/nbs/report/pipeline/person/repository/NrtAuthUserRepository.java
@@ -0,0 +1,8 @@
+package gov.cdc.nbs.report.pipeline.person.repository;
+
+import gov.cdc.nbs.report.pipeline.person.model.entity.NrtAuthUser;
+import org.springframework.data.jpa.repository.JpaRepository;
+import org.springframework.stereotype.Repository;
+
+@Repository
+public interface NrtAuthUserRepository extends JpaRepository<NrtAuthUser, Long> {}
diff --git a/reporting-pipeline-service/src/main/java/gov/cdc/nbs/report/pipeline/person/repository/NrtPatientRepository.java b/reporting-pipeline-service/src/main/java/gov/cdc/nbs/report/pipeline/person/repository/NrtPatientRepository.java
new file mode 100644
index 000000000..332ab7cee
--- /dev/null
+++ b/reporting-pipeline-service/src/main/java/gov/cdc/nbs/report/pipeline/person/repository/NrtPatientRepository.java
@@ -0,0 +1,8 @@
+package gov.cdc.nbs.report.pipeline.person.repository;
+
+import gov.cdc.nbs.report.pipeline.person.model.entity.NrtPatient;
+import org.springframework.data.jpa.repository.JpaRepository;
+import org.springframework.stereotype.Repository;
+
+@Repository
+public interface NrtPatientRepository extends JpaRepository<NrtPatient, Long> {}
diff --git a/reporting-pipeline-service/src/main/java/gov/cdc/nbs/report/pipeline/person/repository/NrtProviderRepository.java b/reporting-pipeline-service/src/main/java/gov/cdc/nbs/report/pipeline/person/repository/NrtProviderRepository.java
new file mode 100644
index 000000000..4ca4aa737
--- /dev/null
+++ b/reporting-pipeline-service/src/main/java/gov/cdc/nbs/report/pipeline/person/repository/NrtProviderRepository.java
@@ -0,0 +1,8 @@
+package gov.cdc.nbs.report.pipeline.person.repository;
+
+import gov.cdc.nbs.report.pipeline.person.model.entity.NrtProvider;
+import org.springframework.data.jpa.repository.JpaRepository;
+import org.springframework.stereotype.Repository;
+
+@Repository
+public interface NrtProviderRepository extends JpaRepository<NrtProvider, Long> {}
diff --git a/reporting-pipeline-service/src/main/java/gov/cdc/nbs/report/pipeline/person/service/PersonService.java b/reporting-pipeline-service/src/main/java/gov/cdc/nbs/report/pipeline/person/service/PersonService.java
index cfb7f55a5..8cfe89f90 100644
--- a/reporting-pipeline-service/src/main/java/gov/cdc/nbs/report/pipeline/person/service/PersonService.java
+++ b/reporting-pipeline-service/src/main/java/gov/cdc/nbs/report/pipeline/person/service/PersonService.java
@@ -6,14 +6,23 @@
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
+import gov.cdc.nbs.report.pipeline.person.model.dto.patient.PatientReporting;
 import gov.cdc.nbs.report.pipeline.person.model.dto.patient.PatientSp;
+import gov.cdc.nbs.report.pipeline.person.model.dto.provider.ProviderReporting;
 import gov.cdc.nbs.report.pipeline.person.model.dto.provider.ProviderSp;
 import gov.cdc.nbs.report.pipeline.person.model.dto.user.AuthUser;
+import gov.cdc.nbs.report.pipeline.person.model.entity.NrtAuthUser;
+import gov.cdc.nbs.report.pipeline.person.model.entity.NrtPatient;
+import gov.cdc.nbs.report.pipeline.person.model.entity.NrtProvider;
+import gov.cdc.nbs.report.pipeline.person.repository.NrtAuthUserRepository;
+import gov.cdc.nbs.report.pipeline.person.repository.NrtPatientRepository;
+import gov.cdc.nbs.report.pipeline.person.repository.NrtProviderRepository;
 import gov.cdc.nbs.report.pipeline.person.repository.PatientRepository;
 import gov.cdc.nbs.report.pipeline.person.repository.ProviderRepository;
 import gov.cdc.nbs.report.pipeline.person.repository.UserRepository;
 import gov.cdc.nbs.report.pipeline.person.transformer.PersonTransformers;
 import gov.cdc.nbs.report.pipeline.person.transformer.PersonType;
+import gov.cdc.nbs.report.pipeline.postprocessing.service.PostProcessingService;
 import gov.cdc.nbs.report.pipeline.util.DataProcessingException;
 import gov.cdc.nbs.report.pipeline.util.NoDataException;
 import gov.cdc.nbs.report.pipeline.util.metrics.CustomMetrics;
@@ -45,6 +54,7 @@
 import org.springframework.retry.annotation.Backoff;
 import org.springframework.scheduling.concurrent.CustomizableThreadFactory;
 import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
 
 /**
  * Service class for processing Person-related change events in the Real Time Reporting (RTR)
@@ -71,6 +81,16 @@ public class PersonService {
   private final PatientRepository patientRepository;
   private final ProviderRepository providerRepository;
   private final UserRepository userRepository;
+  private final NrtPatientRepository nrtPatientRepository;
+  private final NrtProviderRepository nrtProviderRepository;
+  private final NrtAuthUserRepository nrtAuthUserRepository;
+
+  // Feeds the same shared, priority-ordered postprocessing pipeline that Kafka-Connect-sourced
+  // nrt_* messages already go through, instead of calling the postprocessing stored procedures
+  // directly. This keeps direct-write patients/providers/auth-users on the same batching/ordering
+  // guarantees relative to investigation/case_management processing (see APP-787).
+  private final PostProcessingService postProcessingService;
+
   private final PersonTransformers transformer;
 
   @Qualifier("personKafkaTemplate")
@@ -106,6 +126,9 @@ public class PersonService {
   @Value("${featureFlag.thread-pool-size:1}")
   private int threadPoolSize;
 
+  @Value("${featureFlag.person-service-direct-write}")
+  private boolean directWrite;
+
   private ExecutorService rtrExecutor;
   private ExecutorService prsExecutor;
 
@@ -229,14 +252,24 @@ private void processProviderData(List<ProviderSp> providerData) {
 
     providerData.forEach(
         provider -> {
-          String reportingKey = transformer.buildProviderKey(provider);
-          String reportingData = transformer.processData(provider, PersonType.PROVIDER_REPORTING);
-          kafkaTemplate.send(providerReportingOutputTopic, reportingKey, reportingData);
-          log.info(
-              "Provider data (uid={}) sent to {}",
-              provider.getPersonUid(),
-              providerReportingOutputTopic);
-          log.debug("Provider Reporting: {}", reportingData);
+          if (directWrite) {
+            ProviderReporting reporting =
+                (ProviderReporting)
+                    transformer.processData(null, provider, PersonType.PROVIDER_REPORTING);
+            nrtProviderRepository.save(NrtProvider.from(reporting));
+            log.info(
+                "Provider data (uid={}) directly written to nrt_provider", provider.getPersonUid());
+            postProcessingService.enqueue(providerReportingOutputTopic, provider.getPersonUid());
+          } else {
+            String reportingKey = transformer.buildProviderKey(provider);
+            String reportingData = transformer.processData(provider, PersonType.PROVIDER_REPORTING);
+            kafkaTemplate.send(providerReportingOutputTopic, reportingKey, reportingData);
+            log.info(
+                "Provider data (uid={}) sent to {}",
+                provider.getPersonUid(),
+                providerReportingOutputTopic);
+            log.debug("Provider Reporting: {}", reportingData);
+          }
 
           if (elasticSearchEnable) {
             String elasticKey = transformer.buildProviderKey(provider);
@@ -266,14 +299,25 @@ private void processPatientData(List<PatientSp> patientData) {
 
     patientData.forEach(
         personData -> {
-          String reportingKey = transformer.buildPatientKey(personData);
-          String reportingData = transformer.processData(personData, PersonType.PATIENT_REPORTING);
-          kafkaTemplate.send(patientReportingOutputTopic, reportingKey, reportingData);
-          log.info(
-              "Patient data (uid={}) sent to {}",
-              personData.getPersonUid(),
-              patientReportingOutputTopic);
-          log.debug("Patient Reporting: {}", reportingData != null ? reportingData : "");
+          if (directWrite) {
+            PatientReporting reporting =
+                (PatientReporting)
+                    transformer.processData(personData, null, PersonType.PATIENT_REPORTING);
+            nrtPatientRepository.save(NrtPatient.from(reporting));
+            log.info(
+                "Patient data (uid={}) directly written to nrt_patient", personData.getPersonUid());
+            postProcessingService.enqueue(patientReportingOutputTopic, personData.getPersonUid());
+          } else {
+            String reportingKey = transformer.buildPatientKey(personData);
+            String reportingData =
+                transformer.processData(personData, PersonType.PATIENT_REPORTING);
+            kafkaTemplate.send(patientReportingOutputTopic, reportingKey, reportingData);
+            log.info(
+                "Patient data (uid={}) sent to {}",
+                personData.getPersonUid(),
+                patientReportingOutputTopic);
+            log.debug("Patient Reporting: {}", reportingData != null ? reportingData : "");
+          }
 
           if (elasticSearchEnable) {
             String elasticKey = transformer.buildPatientKey(personData);
@@ -289,6 +333,7 @@ private void processPatientData(List<PatientSp> patientData) {
         });
   }
 
+  @Transactional
   private void processUser(String message, String topic) {
     String userUid = "";
     try {
@@ -296,22 +341,34 @@ private void processUser(String message, String topic) {
       log.info(topicDebugLog, "User", userUid, topic);
       Optional<List<AuthUser>> userData = userRepository.computeAuthUsers(userUid);
 
+      List<AuthUser> authUsers = new ArrayList<>();
+
       if (userData.isPresent() && !userData.get().isEmpty()) {
-        userData
-            .get()
-            .forEach(
-                authUser -> {
-                  String jsonKey = transformer.buildUserKey(authUser);
-                  String jsonValue = transformer.processData(authUser);
-                  kafkaTemplate.send(userReportingOutputTopic, jsonKey, jsonValue);
-                  log.info(
-                      "User data (uid={}) sent to {}",
-                      authUser.getAuthUserUid(),
-                      userReportingOutputTopic);
-                });
+        authUsers = userData.get();
       } else {
         throw new EntityNotFoundException("Unable to find AuthUser data for id(s): " + userUid);
       }
+
+      if (directWrite) {
+        // write directly to the database
+        authUsers.forEach(
+            authUser -> {
+              nrtAuthUserRepository.save(NrtAuthUser.from(authUser));
+              postProcessingService.enqueue(userReportingOutputTopic, authUser.getAuthUserUid());
+            });
+      } else {
+        // publish events in kafka
+        authUsers.forEach(
+            authUser -> {
+              String jsonKey = transformer.buildUserKey(authUser);
+              String jsonValue = transformer.processData(authUser);
+              kafkaTemplate.send(userReportingOutputTopic, jsonKey, jsonValue);
+              log.info(
+                  "User data (uid={}) sent to {}",
+                  authUser.getAuthUserUid(),
+                  userReportingOutputTopic);
+            });
+      }
     } catch (EntityNotFoundException ex) {
       throw new NoDataException(ex.getMessage(), ex);
     } catch (Exception e) {
diff --git a/reporting-pipeline-service/src/main/java/gov/cdc/nbs/report/pipeline/postprocessing/service/PostProcessingService.java b/reporting-pipeline-service/src/main/java/gov/cdc/nbs/report/pipeline/postprocessing/service/PostProcessingService.java
index c590198a8..4f32d942a 100644
--- a/reporting-pipeline-service/src/main/java/gov/cdc/nbs/report/pipeline/postprocessing/service/PostProcessingService.java
+++ b/reporting-pipeline-service/src/main/java/gov/cdc/nbs/report/pipeline/postprocessing/service/PostProcessingService.java
@@ -323,6 +323,22 @@ private void extractIdFromMessage(String topic, String messageKey, String payloa
     }
   }
 
+  /**
+   * Adds a single id directly to the postprocessing batch cache for the given topic, without
+   * requiring a Kafka round-trip. Intended for direct-write callers (e.g. {@code PersonService},
+   * for patient/provider/auth-user records written straight to their {@code nrt_*} table instead of
+   * via Kafka-Connect) so they still go through the same priority-ordered batch drain (see {@link
+   * #processCachedIds()}) and its ordering guarantees relative to other entity types.
+   *
+   * <p>This bypasses the payload-based enrichment in {@link #extractValFromMessage}, which only
+   * applies to investigation/notification/observation topics — do not use this for those.
+   */
+  public void enqueue(String topic, Long uid) {
+    synchronized (cacheLock) {
+      idCache.computeIfAbsent(topic, k -> new ConcurrentLinkedQueue<>()).add(uid);
+    }
+  }
+
   /**
    * Extracts and categorizes additional values from a Kafka message payload based on the entity
    * type.
diff --git a/reporting-pipeline-service/src/main/java/gov/cdc/nbs/report/pipeline/util/UtilHelper.java b/reporting-pipeline-service/src/main/java/gov/cdc/nbs/report/pipeline/util/UtilHelper.java
index f6ce8e601..741879a09 100644
--- a/reporting-pipeline-service/src/main/java/gov/cdc/nbs/report/pipeline/util/UtilHelper.java
+++ b/reporting-pipeline-service/src/main/java/gov/cdc/nbs/report/pipeline/util/UtilHelper.java
@@ -4,6 +4,9 @@
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
+import java.math.BigDecimal;
+import java.sql.Timestamp;
+import java.time.LocalDateTime;
 import java.util.NoSuchElementException;
 import lombok.extern.slf4j.Slf4j;
 
@@ -65,4 +68,27 @@ public static String errorMessage(String entityName, String ids, Exception e) {
     }
     return base + ": " + e.getMessage();
   }
+
+  /**
+   * Parses a datetime value pulled from a native SQL Server query result (rendered as a
+   * java.sql.Timestamp-formatted string) back into a LocalDateTime for direct-write entities.
+   */
+  public static LocalDateTime parseDateTime(String value) {
+    if (value == null || value.isBlank()) {
+      return null;
+    }
+    return Timestamp.valueOf(value).toLocalDateTime();
+  }
+
+  public static BigDecimal parseBigDecimal(String value) {
+    if (value == null || value.isBlank()) {
+      return null;
+    }
+    try {
+      return new BigDecimal(value.trim());
+    } catch (NumberFormatException e) {
+      log.warn("Unable to parse numeric value: '{}'", value);
+      return null;
+    }
+  }
 }
diff --git a/reporting-pipeline-service/src/main/resources/application.yaml b/reporting-pipeline-service/src/main/resources/application.yaml
index ee98645ee..805295ac8 100644
--- a/reporting-pipeline-service/src/main/resources/application.yaml
+++ b/reporting-pipeline-service/src/main/resources/application.yaml
@@ -124,6 +124,7 @@ featureFlag:
   thread-pool-size: ${FF_THREAD_POOL_SIZE:1}
   post-processing-enable: ${FF_POST_PROCESSING_ENABLE:true}
   phc-datamart-enable: false
+  person-service-direct-write: ${PERSON_SERVICE_DIRECT_WRITE:true}
 
 management:
   health:
diff --git a/reporting-pipeline-service/src/test/java/gov/cdc/nbs/report/pipeline/person/service/PersonServiceTest.java b/reporting-pipeline-service/src/test/java/gov/cdc/nbs/report/pipeline/person/service/PersonServiceTest.java
index ca16b566a..b6d6df254 100644
--- a/reporting-pipeline-service/src/test/java/gov/cdc/nbs/report/pipeline/person/service/PersonServiceTest.java
+++ b/reporting-pipeline-service/src/test/java/gov/cdc/nbs/report/pipeline/person/service/PersonServiceTest.java
@@ -16,10 +16,14 @@
 import gov.cdc.nbs.report.pipeline.person.model.dto.provider.ProviderSp;
 import gov.cdc.nbs.report.pipeline.person.model.dto.user.AuthUser;
 import gov.cdc.nbs.report.pipeline.person.model.dto.user.AuthUserKey;
+import gov.cdc.nbs.report.pipeline.person.repository.NrtAuthUserRepository;
+import gov.cdc.nbs.report.pipeline.person.repository.NrtPatientRepository;
+import gov.cdc.nbs.report.pipeline.person.repository.NrtProviderRepository;
 import gov.cdc.nbs.report.pipeline.person.repository.PatientRepository;
 import gov.cdc.nbs.report.pipeline.person.repository.ProviderRepository;
 import gov.cdc.nbs.report.pipeline.person.repository.UserRepository;
 import gov.cdc.nbs.report.pipeline.person.transformer.PersonTransformers;
+import gov.cdc.nbs.report.pipeline.postprocessing.service.PostProcessingService;
 import gov.cdc.nbs.report.pipeline.util.DataProcessingException;
 import gov.cdc.nbs.report.pipeline.util.NoDataException;
 import gov.cdc.nbs.report.pipeline.util.metrics.CustomMetrics;
@@ -49,6 +53,14 @@ class PersonServiceTest {
 
   @Mock UserRepository userRepository;
 
+  @Mock NrtPatientRepository nrtPatientRepository;
+
+  @Mock NrtProviderRepository nrtProviderRepository;
+
+  @Mock NrtAuthUserRepository nrtAuthUserRepository;
+
+  @Mock PostProcessingService postProcessingService;
+
   @Mock private KafkaTemplate<String, String> kafkaTemplate;
 
   @Captor private ArgumentCaptor<String> topicCaptor;
@@ -81,6 +93,10 @@ void setUp() {
             patientRepository,
             providerRepository,
             userRepository,
+            nrtPatientRepository,
+            nrtProviderRepository,
+            nrtAuthUserRepository,
+            postProcessingService,
             transformer,
             kafkaTemplate,
             new CustomMetrics(new SimpleMeterRegistry()));
diff --git a/reporting-pipeline-service/src/test/java/gov/cdc/nbs/report/pipeline/postprocessing/service/PostProcessingServiceCacheConcurrencyTest.java b/reporting-pipeline-service/src/test/java/gov/cdc/nbs/report/pipeline/postprocessing/service/PostProcessingServiceCacheConcurrencyTest.java
new file mode 100644
index 000000000..0bb9bb4d3
--- /dev/null
+++ b/reporting-pipeline-service/src/test/java/gov/cdc/nbs/report/pipeline/postprocessing/service/PostProcessingServiceCacheConcurrencyTest.java
@@ -0,0 +1,154 @@
+package gov.cdc.nbs.report.pipeline.postprocessing.service;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.doAnswer;
+
+import gov.cdc.nbs.report.pipeline.postprocessing.repository.InvestigationRepository;
+import gov.cdc.nbs.report.pipeline.postprocessing.repository.PostProcRepository;
+import gov.cdc.nbs.report.pipeline.util.metrics.CustomMetrics;
+import io.micrometer.core.instrument.simple.SimpleMeterRegistry;
+import java.util.Collections;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mock;
+import org.mockito.MockitoAnnotations;
+import org.springframework.kafka.core.KafkaTemplate;
+
+/**
+ * Concurrency regression test for the id-cache snapshot/clear race in {@link
+ * PostProcessingService}.
+ *
+ * <p>{@code idCache} (and its siblings {@code cdCache}/{@code pbCache}/{@code obsCache}/{@code
+ * sumCache}/{@code dmCache}) is drained on a schedule by copying each queue's contents and then
+ * clearing the whole map, both under {@code cacheLock}. Before this fix, producers (a Kafka
+ * listener thread, or a direct-write caller like {@code PersonService} via {@link
+ * PostProcessingService#enqueue(String, Long)}) added to those queues without taking {@code
+ * cacheLock}. If an add landed between the drain's snapshot copy and its {@code clear()}, the map
+ * entry backing that queue was dropped from under it — the id was not delayed to the next cycle, it
+ * was lost permanently, since a later {@code computeIfAbsent} call creates a brand-new queue rather
+ * than reusing the orphaned one.
+ *
+ * <p>This fires many concurrent producers against a tight drain loop and asserts every enqueued id
+ * is eventually seen by the postprocessing stored procedure call exactly once. Against the
+ * unsynchronized producer side this test is flaky-to-failing (ids go missing); with producer adds
+ * synchronized on {@code cacheLock} it passes deterministically.
+ */
+class PostProcessingServiceCacheConcurrencyTest {
+
+  @Mock private PostProcRepository postProcRepositoryMock;
+  @Mock private InvestigationRepository investigationRepositoryMock;
+  @Mock private KafkaTemplate<String, String> kafkaTemplate;
+
+  private PostProcessingService postProcessingService;
+  private AutoCloseable closeable;
+
+  @BeforeEach
+  void setUp() {
+    closeable = MockitoAnnotations.openMocks(this);
+    ProcessDatamartData datamartProcessor =
+        new ProcessDatamartData(
+            kafkaTemplate,
+            postProcRepositoryMock,
+            investigationRepositoryMock,
+            new CustomMetrics(new SimpleMeterRegistry()));
+    datamartProcessor.initMetrics();
+
+    postProcessingService =
+        new PostProcessingService(
+            postProcRepositoryMock,
+            investigationRepositoryMock,
+            datamartProcessor,
+            new CustomMetrics(new SimpleMeterRegistry()));
+    postProcessingService.setMaxRetries(0);
+    postProcessingService.initMetrics();
+    postProcessingService.setServiceEnable(true);
+  }
+
+  @AfterEach
+  void tearDown() throws Exception {
+    closeable.close();
+  }
+
+  @Test
+  void concurrentEnqueueDuringDrainLosesNoIds() throws InterruptedException {
+    final String topic = "dummy_patient";
+    final int producerThreads = 12;
+    final int idsPerThread = 200;
+    final int totalIds = producerThreads * idsPerThread;
+
+    Set<Long> observedIds = ConcurrentHashMap.newKeySet();
+    doAnswer(
+            invocation -> {
+              String idsCsv = invocation.getArgument(0);
+              for (String id : idsCsv.split(",")) {
+                if (!id.isBlank()) {
+                  observedIds.add(Long.valueOf(id));
+                }
+              }
+              return Collections.emptyList();
+            })
+        .when(postProcRepositoryMock)
+        .executeStoredProcForPatientIds(anyString());
+
+    AtomicBoolean producing = new AtomicBoolean(true);
+
+    // Drain thread: hammers processCachedIds() continuously -- far more often than the real
+    // @Scheduled cycle would -- to maximize the odds of hitting the snapshot/clear race window.
+    Thread drainThread =
+        new Thread(
+            () -> {
+              while (producing.get()) {
+                postProcessingService.processCachedIds();
+              }
+            });
+    drainThread.start();
+
+    ExecutorService producers = Executors.newFixedThreadPool(producerThreads);
+    CountDownLatch done = new CountDownLatch(producerThreads);
+    try {
+      for (int t = 0; t < producerThreads; t++) {
+        final int threadIndex = t;
+        producers.submit(
+            () -> {
+              try {
+                for (int i = 0; i < idsPerThread; i++) {
+                  long id = (long) threadIndex * idsPerThread + i;
+                  postProcessingService.enqueue(topic, id);
+                }
+              } finally {
+                done.countDown();
+              }
+            });
+      }
+      assertTrue(done.await(30, TimeUnit.SECONDS), "producer threads did not finish in time");
+    } finally {
+      producers.shutdownNow();
+    }
+
+    producing.set(false);
+    drainThread.join(TimeUnit.SECONDS.toMillis(10));
+
+    // Final drain to catch anything enqueued right at the end of the producer run.
+    postProcessingService.processCachedIds();
+
+    assertEquals(
+        totalIds,
+        observedIds.size(),
+        () ->
+            "expected all "
+                + totalIds
+                + " enqueued ids to be observed by the postprocessing call, but only saw "
+                + observedIds.size()
+                + " -- ids are being silently dropped by a producer/drain race");
+  }
+}
diff --git a/reporting-pipeline-service/src/test/java/gov/cdc/nbs/report/pipeline/postprocessing/service/PostProcessingServiceEnqueueTest.java b/reporting-pipeline-service/src/test/java/gov/cdc/nbs/report/pipeline/postprocessing/service/PostProcessingServiceEnqueueTest.java
new file mode 100644
index 000000000..c289ffec4
--- /dev/null
+++ b/reporting-pipeline-service/src/test/java/gov/cdc/nbs/report/pipeline/postprocessing/service/PostProcessingServiceEnqueueTest.java
@@ -0,0 +1,176 @@
+package gov.cdc.nbs.report.pipeline.postprocessing.service;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoInteractions;
+import static org.mockito.Mockito.when;
+
+import ch.qos.logback.classic.Logger;
+import ch.qos.logback.classic.spi.ILoggingEvent;
+import ch.qos.logback.core.read.ListAppender;
+import gov.cdc.nbs.report.pipeline.postprocessing.repository.InvestigationRepository;
+import gov.cdc.nbs.report.pipeline.postprocessing.repository.PostProcRepository;
+import gov.cdc.nbs.report.pipeline.util.metrics.CustomMetrics;
+import io.micrometer.core.instrument.simple.SimpleMeterRegistry;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mock;
+import org.mockito.MockitoAnnotations;
+import org.slf4j.LoggerFactory;
+import org.springframework.kafka.core.KafkaTemplate;
+
+/**
+ * Single-threaded correctness/contract tests for {@link PostProcessingService#enqueue(String,
+ * Long)}. Concurrency behavior (producer/drain race safety) is covered separately in {@link
+ * PostProcessingServiceCacheConcurrencyTest}.
+ */
+class PostProcessingServiceEnqueueTest {
+
+  @Mock private PostProcRepository postProcRepositoryMock;
+  @Mock private InvestigationRepository investigationRepositoryMock;
+  @Mock private KafkaTemplate<String, String> kafkaTemplate;
+
+  private PostProcessingService postProcessingService;
+  private final ListAppender<ILoggingEvent> listAppender = new ListAppender<>();
+  private AutoCloseable closeable;
+
+  @BeforeEach
+  void setUp() {
+    closeable = MockitoAnnotations.openMocks(this);
+    ProcessDatamartData datamartProcessor =
+        new ProcessDatamartData(
+            kafkaTemplate,
+            postProcRepositoryMock,
+            investigationRepositoryMock,
+            new CustomMetrics(new SimpleMeterRegistry()));
+    datamartProcessor.initMetrics();
+
+    postProcessingService =
+        new PostProcessingService(
+            postProcRepositoryMock,
+            investigationRepositoryMock,
+            datamartProcessor,
+            new CustomMetrics(new SimpleMeterRegistry()));
+    postProcessingService.initMetrics();
+    postProcessingService.setServiceEnable(true);
+
+    Logger logger = (Logger) LoggerFactory.getLogger(PostProcessingService.class);
+    listAppender.start();
+    logger.addAppender(listAppender);
+  }
+
+  @AfterEach
+  void tearDown() throws Exception {
+    Logger logger = (Logger) LoggerFactory.getLogger(PostProcessingService.class);
+    logger.detachAppender(listAppender);
+    closeable.close();
+  }
+
+  @Test
+  void enqueueAddsIdToCache() {
+    String topic = "dummy_patient";
+
+    postProcessingService.enqueue(topic, 123L);
+
+    assertTrue(postProcessingService.idCache.containsKey(topic));
+    assertEquals(123L, postProcessingService.idCache.get(topic).element());
+  }
+
+  @Test
+  void enqueueAccumulatesMultipleIdsUnderSameTopic() {
+    String topic = "dummy_provider";
+
+    postProcessingService.enqueue(topic, 1L);
+    postProcessingService.enqueue(topic, 2L);
+    postProcessingService.enqueue(topic, 3L);
+
+    assertEquals(List.of(1L, 2L, 3L), new ArrayList<>(postProcessingService.idCache.get(topic)));
+  }
+
+  @Test
+  void enqueueKeepsDifferentTopicsIsolated() {
+    postProcessingService.enqueue("dummy_patient", 100L);
+    postProcessingService.enqueue("dummy_provider", 200L);
+
+    assertEquals(2, postProcessingService.idCache.size());
+    assertEquals(100L, postProcessingService.idCache.get("dummy_patient").element());
+    assertEquals(200L, postProcessingService.idCache.get("dummy_provider").element());
+  }
+
+  @Test
+  void enqueuedPatientIdIsDrainedToPatientStoredProc() {
+    when(postProcRepositoryMock.executeStoredProcForPatientIds(anyString()))
+        .thenReturn(Collections.emptyList());
+
+    postProcessingService.enqueue("dummy_patient", 555L);
+    postProcessingService.processCachedIds();
+
+    verify(postProcRepositoryMock).executeStoredProcForPatientIds("555");
+    assertTrue(postProcessingService.idCache.isEmpty());
+  }
+
+  @Test
+  void enqueuedProviderIdIsDrainedToProviderStoredProc() {
+    when(postProcRepositoryMock.executeStoredProcForProviderIds(anyString()))
+        .thenReturn(Collections.emptyList());
+
+    postProcessingService.enqueue("dummy_provider", 777L);
+    postProcessingService.processCachedIds();
+
+    verify(postProcRepositoryMock).executeStoredProcForProviderIds("777");
+    assertTrue(postProcessingService.idCache.isEmpty());
+  }
+
+  @Test
+  void enqueuedAuthUserIdIsDrainedToUserProfileStoredProc() {
+    when(postProcRepositoryMock.executeStoredProcForUserProfile(anyString()))
+        .thenReturn(Collections.emptyList());
+
+    postProcessingService.enqueue("dummy_auth_user", 999L);
+    postProcessingService.processCachedIds();
+
+    verify(postProcRepositoryMock).executeStoredProcForUserProfile("999");
+    assertTrue(postProcessingService.idCache.isEmpty());
+  }
+
+  @Test
+  void enqueueIgnoresServiceEnableFlag() {
+    postProcessingService.setServiceEnable(false);
+
+    postProcessingService.enqueue("dummy_patient", 42L);
+
+    assertTrue(postProcessingService.idCache.containsKey("dummy_patient"));
+  }
+
+  @Test
+  void enqueueRejectsNullUid() {
+    assertThrows(
+        NullPointerException.class, () -> postProcessingService.enqueue("dummy_patient", null));
+  }
+
+  @Test
+  void enqueueRejectsNullTopic() {
+    assertThrows(NullPointerException.class, () -> postProcessingService.enqueue(null, 1L));
+  }
+
+  @Test
+  void enqueueOnUnrecognizedTopicIsSilentlyDroppedByDrainWithWarning() {
+    String topic = "dummy_totally_unknown_entity";
+
+    postProcessingService.enqueue(topic, 1234L);
+    postProcessingService.processCachedIds();
+
+    verifyNoInteractions(postProcRepositoryMock);
+    assertTrue(postProcessingService.idCache.isEmpty());
+    assertTrue(
+        listAppender.list.stream().anyMatch(event -> event.getFormattedMessage().contains(topic)),
+        "expected a warning log naming the unrecognized topic");
+  }
+}
diff --git a/reporting-pipeline-service/src/test/resources/testData/functional/README.md b/reporting-pipeline-service/src/test/resources/testData/functional/README.md
index 8b0a85d3c..b1d6dc29b 100644
--- a/reporting-pipeline-service/src/test/resources/testData/functional/README.md
+++ b/reporting-pipeline-service/src/test/resources/testData/functional/README.md
@@ -22,6 +22,8 @@ When adding a new functional test, pick the next available range, and add your t
 | stdContactTracingPartTwo | 1000009000  | 1000009028 |
 | hivNotificationPlan      | 1000010000  | 1000010004 |
 | d_tb_pam                 | 1000011000  | 1000011004 |
+| authUserDirectWrite      | 1000012000  | 1000012000 |
+| providerDirectWrite      | 1000013000  | 1000013000 |
 
 ## Helper tools
 
diff --git a/reporting-pipeline-service/src/test/resources/testData/functional/authUserDirectWrite/010-createAuthUser/expected.json b/reporting-pipeline-service/src/test/resources/testData/functional/authUserDirectWrite/010-createAuthUser/expected.json
new file mode 100644
index 000000000..cc3455902
--- /dev/null
+++ b/reporting-pipeline-service/src/test/resources/testData/functional/authUserDirectWrite/010-createAuthUser/expected.json
@@ -0,0 +1,21 @@
+{
+  "0": [
+    {
+      "auth_user_uid": 1000012000,
+      "user_id": "ada.lovelace",
+      "first_nm": "Ada",
+      "last_nm": "Lovelace",
+      "nedss_entry_id": 1000012000,
+      "record_status_cd": "ACTIVE"
+    }
+  ],
+  "1": [
+    {
+      "FIRST_NM": "Ada",
+      "LAST_NM": "Lovelace",
+      "NEDSS_ENTRY_ID": 1000012000,
+      "PROVIDER_KEY": 1,
+      "USER_NM": "Lovelace, Ada"
+    }
+  ]
+}
diff --git a/reporting-pipeline-service/src/test/resources/testData/functional/authUserDirectWrite/010-createAuthUser/query.sql b/reporting-pipeline-service/src/test/resources/testData/functional/authUserDirectWrite/010-createAuthUser/query.sql
new file mode 100644
index 000000000..0096e1828
--- /dev/null
+++ b/reporting-pipeline-service/src/test/resources/testData/functional/authUserDirectWrite/010-createAuthUser/query.sql
@@ -0,0 +1,20 @@
+-- Query 0: Verify nrt_auth_user was directly written by PersonService (direct-write path)
+SELECT
+    [auth_user_uid],
+    [user_id],
+    [first_nm],
+    [last_nm],
+    [nedss_entry_id],
+    [record_status_cd]
+FROM [RDB_MODERN].[dbo].[nrt_auth_user]
+WHERE [auth_user_uid] = 1000012000;
+
+-- Query 1: Verify sp_user_profile_postprocessing hydrated USER_PROFILE
+SELECT
+    [FIRST_NM],
+    [LAST_NM],
+    [NEDSS_ENTRY_ID],
+    [PROVIDER_KEY],
+    [USER_NM]
+FROM [RDB_MODERN].[dbo].[USER_PROFILE]
+WHERE [NEDSS_ENTRY_ID] = 1000012000;
diff --git a/reporting-pipeline-service/src/test/resources/testData/functional/authUserDirectWrite/010-createAuthUser/setup.sql b/reporting-pipeline-service/src/test/resources/testData/functional/authUserDirectWrite/010-createAuthUser/setup.sql
new file mode 100644
index 000000000..faa33f6d4
--- /dev/null
+++ b/reporting-pipeline-service/src/test/resources/testData/functional/authUserDirectWrite/010-createAuthUser/setup.sql
@@ -0,0 +1,20 @@
+USE [NBS_ODSE];
+
+-- Adjust the UID declarations below manually so they remain unique across other tests.
+DECLARE @superuser_id bigint = 10009282;
+DECLARE @dbo_Auth_user_auth_user_uid bigint = 1000012000;
+DECLARE @dbo_Auth_user_nedss_entry_id bigint = 1000012000;
+
+-- dbo.Auth_user
+SET IDENTITY_INSERT [dbo].[Auth_user] ON;
+INSERT INTO [dbo].[Auth_user] (
+    [auth_user_uid], [user_id], [user_first_nm], [user_last_nm], [nedss_entry_id],
+    [provider_uid], [add_user_id], [last_chg_user_id], [add_time], [last_chg_time],
+    [record_status_cd], [record_status_time]
+)
+VALUES (
+    @dbo_Auth_user_auth_user_uid, N'ada.lovelace', N'Ada', N'Lovelace', @dbo_Auth_user_nedss_entry_id,
+    NULL, @superuser_id, @superuser_id, N'2026-07-01T00:00:00.000', N'2026-07-01T00:00:00.000',
+    N'ACTIVE', N'2026-07-01T00:00:00.000'
+);
+SET IDENTITY_INSERT [dbo].[Auth_user] OFF;
diff --git a/reporting-pipeline-service/src/test/resources/testData/functional/providerDirectWrite/010-createProvider/expected.json b/reporting-pipeline-service/src/test/resources/testData/functional/providerDirectWrite/010-createProvider/expected.json
new file mode 100644
index 000000000..23eacce82
--- /dev/null
+++ b/reporting-pipeline-service/src/test/resources/testData/functional/providerDirectWrite/010-createProvider/expected.json
@@ -0,0 +1,20 @@
+{
+  "0": [
+    {
+      "provider_uid": 1000013000,
+      "local_id": "PRV1000013000GA01",
+      "first_name": "Grace",
+      "last_name": "Hopper",
+      "record_status": "ACTIVE"
+    }
+  ],
+  "1": [
+    {
+      "PROVIDER_UID": 1000013000,
+      "PROVIDER_LOCAL_ID": "PRV1000013000GA01",
+      "PROVIDER_FIRST_NAME": "Grace",
+      "PROVIDER_LAST_NAME": "Hopper",
+      "PROVIDER_RECORD_STATUS": "ACTIVE"
+    }
+  ]
+}
diff --git a/reporting-pipeline-service/src/test/resources/testData/functional/providerDirectWrite/010-createProvider/query.sql b/reporting-pipeline-service/src/test/resources/testData/functional/providerDirectWrite/010-createProvider/query.sql
new file mode 100644
index 000000000..ba59c1f18
--- /dev/null
+++ b/reporting-pipeline-service/src/test/resources/testData/functional/providerDirectWrite/010-createProvider/query.sql
@@ -0,0 +1,19 @@
+-- Query 0: Verify nrt_provider was directly written by PersonService (direct-write path)
+SELECT
+    [provider_uid],
+    [local_id],
+    [first_name],
+    [last_name],
+    [record_status]
+FROM [RDB_MODERN].[dbo].[nrt_provider]
+WHERE [provider_uid] = 1000013000;
+
+-- Query 1: Verify sp_nrt_provider_postprocessing hydrated D_PROVIDER
+SELECT
+    [PROVIDER_UID],
+    [PROVIDER_LOCAL_ID],
+    [PROVIDER_FIRST_NAME],
+    [PROVIDER_LAST_NAME],
+    [PROVIDER_RECORD_STATUS]
+FROM [RDB_MODERN].[dbo].[D_PROVIDER]
+WHERE [PROVIDER_UID] = 1000013000;
diff --git a/reporting-pipeline-service/src/test/resources/testData/functional/providerDirectWrite/010-createProvider/setup.sql b/reporting-pipeline-service/src/test/resources/testData/functional/providerDirectWrite/010-createProvider/setup.sql
new file mode 100644
index 000000000..984d69016
--- /dev/null
+++ b/reporting-pipeline-service/src/test/resources/testData/functional/providerDirectWrite/010-createProvider/setup.sql
@@ -0,0 +1,35 @@
+USE [NBS_ODSE];
+
+-- Adjust the UID declarations below manually so they remain unique across other tests.
+DECLARE @superuser_id bigint = 10009282;
+DECLARE @dbo_Entity_entity_uid bigint = 1000013000;
+DECLARE @dbo_Person_local_id varchar(50) = N'PRV1000013000GA01';
+
+-- dbo.Entity
+INSERT INTO [dbo].[Entity] ([entity_uid], [class_cd])
+VALUES (@dbo_Entity_entity_uid, N'PSN');
+
+-- dbo.Person (cd = 'PRV' marks this Person row as a Provider)
+INSERT INTO [dbo].[Person] (
+    [person_uid], [add_time], [add_user_id], [last_chg_time], [last_chg_user_id],
+    [cd], [cd_desc_txt], [first_nm], [last_nm], [middle_nm],
+    [record_status_cd], [record_status_time], [status_cd], [status_time],
+    [local_id], [version_ctrl_nbr], [electronic_ind], [person_parent_uid]
+)
+VALUES (
+    @dbo_Entity_entity_uid, N'2026-07-01T00:00:00.000', @superuser_id, N'2026-07-01T00:00:00.000', @superuser_id,
+    N'PRV', N'Provider', N'Grace', N'Hopper', NULL,
+    N'ACTIVE', N'2026-07-01T00:00:00.000', N'A', N'2026-07-01T00:00:00.000',
+    @dbo_Person_local_id, 1, N'Y', @dbo_Entity_entity_uid
+);
+
+-- dbo.Person_name (sp_provider_event sources first/last name from here, not from dbo.Person)
+-- No middle name yet, and no phone/email/address on the entity yet -- step 020 adds those.
+INSERT INTO [dbo].[Person_name] (
+    [person_uid], [person_name_seq], [first_nm], [last_nm], [middle_nm],
+    [nm_use_cd], [record_status_cd], [record_status_time], [status_cd], [status_time], [last_chg_time]
+)
+VALUES (
+    @dbo_Entity_entity_uid, 1, N'Grace', N'Hopper', NULL,
+    N'L', N'ACTIVE', N'2026-07-01T00:00:00.000', N'A', N'2026-07-01T00:00:00.000', N'2026-07-01T00:00:00.000'
+);
diff --git a/reporting-pipeline-service/src/test/resources/testData/functional/providerDirectWrite/020-updateProvider/expected.json b/reporting-pipeline-service/src/test/resources/testData/functional/providerDirectWrite/020-updateProvider/expected.json
new file mode 100644
index 000000000..7b5341ef4
--- /dev/null
+++ b/reporting-pipeline-service/src/test/resources/testData/functional/providerDirectWrite/020-updateProvider/expected.json
@@ -0,0 +1,20 @@
+{
+  "0": [
+    {
+      "provider_uid": 1000013000,
+      "first_name": "Grace",
+      "middle_name": "Brewster",
+      "last_name": "Hopper",
+      "email_work": "grace.hopper@navy.mil"
+    }
+  ],
+  "1": [
+    {
+      "PROVIDER_UID": 1000013000,
+      "PROVIDER_FIRST_NAME": "Grace",
+      "PROVIDER_MIDDLE_NAME": "Brewster",
+      "PROVIDER_LAST_NAME": "Hopper",
+      "PROVIDER_EMAIL_WORK": "grace.hopper@navy.mil"
+    }
+  ]
+}
diff --git a/reporting-pipeline-service/src/test/resources/testData/functional/providerDirectWrite/020-updateProvider/query.sql b/reporting-pipeline-service/src/test/resources/testData/functional/providerDirectWrite/020-updateProvider/query.sql
new file mode 100644
index 000000000..4bef12dd0
--- /dev/null
+++ b/reporting-pipeline-service/src/test/resources/testData/functional/providerDirectWrite/020-updateProvider/query.sql
@@ -0,0 +1,19 @@
+-- Query 0: Verify nrt_provider was re-written with the updated middle name and email
+SELECT
+    [provider_uid],
+    [first_name],
+    [middle_name],
+    [last_name],
+    [email_work]
+FROM [RDB_MODERN].[dbo].[nrt_provider]
+WHERE [provider_uid] = 1000013000;
+
+-- Query 1: Verify sp_nrt_provider_postprocessing propagated the update to D_PROVIDER
+SELECT
+    [PROVIDER_UID],
+    [PROVIDER_FIRST_NAME],
+    [PROVIDER_MIDDLE_NAME],
+    [PROVIDER_LAST_NAME],
+    [PROVIDER_EMAIL_WORK]
+FROM [RDB_MODERN].[dbo].[D_PROVIDER]
+WHERE [PROVIDER_UID] = 1000013000;
diff --git a/reporting-pipeline-service/src/test/resources/testData/functional/providerDirectWrite/020-updateProvider/setup.sql b/reporting-pipeline-service/src/test/resources/testData/functional/providerDirectWrite/020-updateProvider/setup.sql
new file mode 100644
index 000000000..2b483dce6
--- /dev/null
+++ b/reporting-pipeline-service/src/test/resources/testData/functional/providerDirectWrite/020-updateProvider/setup.sql
@@ -0,0 +1,29 @@
+USE [NBS_ODSE];
+
+-- Adjust the UID declarations below manually so they remain unique across other tests.
+DECLARE @dbo_Entity_entity_uid bigint = 1000013000;
+DECLARE @dbo_Tele_locator_tele_locator_uid_email bigint = 1000013001;
+
+-- Add a middle name to the provider created in 010-createProvider
+UPDATE [dbo].[Person_name]
+SET [middle_nm] = N'Brewster', [last_chg_time] = N'2026-07-01T01:00:00.000'
+WHERE [person_uid] = @dbo_Entity_entity_uid AND [person_name_seq] = 1;
+
+-- Add a work email (contact info)
+INSERT INTO [dbo].[Tele_locator] ([tele_locator_uid], [email_address], [record_status_cd], [record_status_time])
+VALUES (@dbo_Tele_locator_tele_locator_uid_email, N'grace.hopper@navy.mil', N'ACTIVE', N'2026-07-01T01:00:00.000');
+
+INSERT INTO [dbo].[Entity_locator_participation] (
+    [entity_uid], [locator_uid], [class_cd], [cd], [use_cd],
+    [record_status_cd], [record_status_time], [status_cd], [version_ctrl_nbr]
+)
+VALUES (
+    @dbo_Entity_entity_uid, @dbo_Tele_locator_tele_locator_uid_email, N'TELE', N'O', N'WP',
+    N'ACTIVE', N'2026-07-01T01:00:00.000', N'A', 1
+);
+
+-- dbo.Person_name/Tele_locator/Entity_locator_participation are not captured by Debezium CDC directly;
+-- touch dbo.Person (which is captured) to trigger PersonService to reprocess this provider with the new data.
+UPDATE [dbo].[Person]
+SET [last_chg_time] = N'2026-07-01T01:00:00.000'
+WHERE [person_uid] = @dbo_Entity_entity_uid;