diff --git a/docs/superpowers/plans/2026-06-23-gb32960-service-split.md b/docs/superpowers/plans/2026-06-23-gb32960-service-split.md
new file mode 100644
index 00000000..0cbd3ae3
--- /dev/null
+++ b/docs/superpowers/plans/2026-06-23-gb32960-service-split.md
@@ -0,0 +1,1639 @@
+# GB32960 Service Split 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:** Split the current all-in-one GB32960 production path into separate ingest, history, and analytics runnable apps with Kafka as the durability boundary and ACK after Kafka success.
+
+**Architecture:** Add three Spring Boot app modules while keeping `bootstrap-all` available for development and rollback. First isolate runtime composition, then introduce GB32960-specific Kafka topics, then change the GB32960 ACK path so realtime/report ACKs are written after required Kafka production succeeds. History and analytics become Kafka consumers in separate apps and no longer share a JVM with TCP ingest.
+
+**Tech Stack:** Java 25, Spring Boot 3.5, Maven multi-module build, Netty, Kafka clients, Protobuf envelope, DuckDB/Parquet event store, local archive sink.
+
+---
+
+## Current-State Notes
+
+- Existing all-in-one entrypoint: `modules/apps/bootstrap-all/src/main/java/com/lingniu/ingest/bootstrap/IngestApplication.java`.
+- Existing all-in-one module: `modules/apps/bootstrap-all/pom.xml` imports protocols, sinks, history, state, stat, command gateway, MQTT, Xinda, and JT modules.
+- Existing GB32960 handler: `modules/protocols/protocol-gb32960/src/main/java/com/lingniu/ingest/protocol/gb32960/inbound/Gb32960ChannelHandler.java`.
+- Existing handler currently writes report ACK before `dispatcher.dispatch(rf)`.
+- Existing `DisruptorEventBus` publishes to sinks asynchronously and does not wait for sink completion.
+- Existing `KafkaEventSink.accepts()` rejects `VehicleEvent.RawArchive`.
+- Existing versioned topic names are not yet present; current topic names live under `lingniu.ingest.sink.mq.topics`.
+- Existing history consumer logic: `modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/EventHistoryEnvelopeIngestor.java`.
+- Existing analytics consumers: `VehicleStateEnvelopeIngestor` and `VehicleStatEnvelopeIngestor`.
+- There is an unrelated working-tree change in `modules/apps/bootstrap-all/src/main/resources/application.yml` for GB32960 diagnostics capacity. Do not revert it. Do not include it in service-split commits unless the user explicitly asks.
+
+## File Structure
+
+Create app modules:
+
+- `modules/apps/gb32960-ingest-app/pom.xml`
+- `modules/apps/gb32960-ingest-app/src/main/java/com/lingniu/ingest/gb32960app/Gb32960IngestApplication.java`
+- `modules/apps/gb32960-ingest-app/src/main/resources/application.yml`
+- `modules/apps/gb32960-ingest-app/src/test/java/com/lingniu/ingest/gb32960app/Gb32960IngestAppCompositionTest.java`
+- `modules/apps/gb32960-ingest-app/src/test/java/com/lingniu/ingest/gb32960app/Gb32960IngestAppDefaultsTest.java`
+
+- `modules/apps/vehicle-history-app/pom.xml`
+- `modules/apps/vehicle-history-app/src/main/java/com/lingniu/ingest/historyapp/VehicleHistoryApplication.java`
+- `modules/apps/vehicle-history-app/src/main/resources/application.yml`
+- `modules/apps/vehicle-history-app/src/test/java/com/lingniu/ingest/historyapp/VehicleHistoryAppCompositionTest.java`
+- `modules/apps/vehicle-history-app/src/test/java/com/lingniu/ingest/historyapp/VehicleHistoryAppDefaultsTest.java`
+
+- `modules/apps/vehicle-analytics-app/pom.xml`
+- `modules/apps/vehicle-analytics-app/src/main/java/com/lingniu/ingest/analyticsapp/VehicleAnalyticsApplication.java`
+- `modules/apps/vehicle-analytics-app/src/main/resources/application.yml`
+- `modules/apps/vehicle-analytics-app/src/test/java/com/lingniu/ingest/analyticsapp/VehicleAnalyticsAppCompositionTest.java`
+- `modules/apps/vehicle-analytics-app/src/test/java/com/lingniu/ingest/analyticsapp/VehicleAnalyticsAppDefaultsTest.java`
+
+Modify shared build and Kafka contract files:
+
+- `pom.xml`
+- `modules/sinks/sink-mq/src/main/java/com/lingniu/ingest/sink/mq/SinkMqProperties.java`
+- `modules/sinks/sink-mq/src/main/java/com/lingniu/ingest/sink/mq/TopicRouter.java`
+- `modules/sinks/sink-mq/src/main/java/com/lingniu/ingest/sink/mq/KafkaEventSink.java`
+- `modules/sinks/sink-mq/src/test/java/com/lingniu/ingest/sink/mq/TopicRouterTest.java`
+- `modules/sinks/sink-mq/src/test/java/com/lingniu/ingest/sink/mq/KafkaEventSinkTest.java`
+
+Modify ACK/Kafka durability boundary files:
+
+- `modules/core/ingest-api/src/main/java/com/lingniu/ingest/api/sink/EventSink.java`
+- `modules/core/ingest-core/src/main/java/com/lingniu/ingest/core/dispatcher/Dispatcher.java`
+- `modules/core/ingest-core/src/main/java/com/lingniu/ingest/core/concurrency/DisruptorEventBus.java`
+- `modules/protocols/protocol-gb32960/src/main/java/com/lingniu/ingest/protocol/gb32960/inbound/Gb32960ChannelHandler.java`
+- `modules/protocols/protocol-gb32960/src/test/java/com/lingniu/ingest/protocol/gb32960/inbound/Gb32960ChannelHandlerAckBoundaryTest.java`
+
+Optional later follow-up files after the first split is running:
+
+- `docs/operations/gb32960-service-split-runbook.md`
+- Deployment manifests or service manager files used by the user's environment.
+
+---
+
+### Task 1: Register the Three New App Modules
+
+**Files:**
+- Modify: `pom.xml`
+- Create: `modules/apps/gb32960-ingest-app/pom.xml`
+- Create: `modules/apps/vehicle-history-app/pom.xml`
+- Create: `modules/apps/vehicle-analytics-app/pom.xml`
+
+- [ ] **Step 1: Add failing module-presence test by running Maven before modules exist**
+
+Run:
+
+```bash
+export JAVA_HOME=/opt/homebrew/opt/openjdk@25/libexec/openjdk.jdk/Contents/Home
+export PATH="$JAVA_HOME/bin:/opt/homebrew/bin:$PATH"
+mvn -pl :gb32960-ingest-app -am validate
+```
+
+Expected: FAIL with a message that Maven cannot find project `:gb32960-ingest-app`.
+
+- [ ] **Step 2: Add modules to the root reactor**
+
+In `pom.xml`, add these module entries after `modules/apps/command-gateway` and before `modules/apps/bootstrap-all`:
+
+```xml
+ modules/apps/gb32960-ingest-app
+ modules/apps/vehicle-history-app
+ modules/apps/vehicle-analytics-app
+```
+
+Also add dependency management entries near the existing `command-gateway` entry:
+
+```xml
+
+ com.lingniu.ingest
+ gb32960-ingest-app
+ ${project.version}
+
+
+ com.lingniu.ingest
+ vehicle-history-app
+ ${project.version}
+
+
+ com.lingniu.ingest
+ vehicle-analytics-app
+ ${project.version}
+
+```
+
+- [ ] **Step 3: Create `gb32960-ingest-app` POM**
+
+Create `modules/apps/gb32960-ingest-app/pom.xml`:
+
+```xml
+
+
+ 4.0.0
+
+ com.lingniu.ingest
+ lingniu-vehicle-ingest
+ 0.1.0-SNAPSHOT
+ ../../../pom.xml
+
+
+ gb32960-ingest-app
+ gb32960-ingest-app
+ GB32960 protocol ingress runtime: TCP decode, auth, Kafka production, and ACK.
+
+
+
+ com.lingniu.ingest
+ ingest-core
+
+
+ com.lingniu.ingest
+ session-core
+
+
+ com.lingniu.ingest
+ observability
+
+
+ com.lingniu.ingest
+ protocol-gb32960
+
+
+ com.lingniu.ingest
+ sink-mq
+
+
+ org.springframework.boot
+ spring-boot-starter
+
+
+ org.springframework.boot
+ spring-boot-starter-test
+ test
+
+
+
+
+ gb32960-ingest-app
+
+
+ org.springframework.boot
+ spring-boot-maven-plugin
+
+ --sun-misc-unsafe-memory-access=allow
+
+
+
+
+ repackage
+
+
+
+
+
+
+
+```
+
+- [ ] **Step 4: Create `vehicle-history-app` POM**
+
+Create `modules/apps/vehicle-history-app/pom.xml`:
+
+```xml
+
+
+ 4.0.0
+
+ com.lingniu.ingest
+ lingniu-vehicle-ingest
+ 0.1.0-SNAPSHOT
+ ../../../pom.xml
+
+
+ vehicle-history-app
+ vehicle-history-app
+ Vehicle raw archive, event history, and GB32960 frame query runtime.
+
+
+
+ com.lingniu.ingest
+ ingest-api
+
+
+ com.lingniu.ingest
+ observability
+
+
+ com.lingniu.ingest
+ sink-mq
+
+
+ com.lingniu.ingest
+ sink-archive
+
+
+ com.lingniu.ingest
+ event-file-store
+
+
+ com.lingniu.ingest
+ event-history-service
+
+
+ com.lingniu.ingest
+ protocol-gb32960
+
+
+ org.springframework.boot
+ spring-boot-starter-web
+
+
+ org.springframework.boot
+ spring-boot-starter-test
+ test
+
+
+
+
+ vehicle-history-app
+
+
+ org.springframework.boot
+ spring-boot-maven-plugin
+
+ --sun-misc-unsafe-memory-access=allow
+
+
+
+
+ repackage
+
+
+
+
+
+
+
+```
+
+- [ ] **Step 5: Create `vehicle-analytics-app` POM**
+
+Create `modules/apps/vehicle-analytics-app/pom.xml`:
+
+```xml
+
+
+ 4.0.0
+
+ com.lingniu.ingest
+ lingniu-vehicle-ingest
+ 0.1.0-SNAPSHOT
+ ../../../pom.xml
+
+
+ vehicle-analytics-app
+ vehicle-analytics-app
+ Vehicle state, daily statistics, alarms, and analytics runtime.
+
+
+
+ com.lingniu.ingest
+ ingest-api
+
+
+ com.lingniu.ingest
+ observability
+
+
+ com.lingniu.ingest
+ sink-mq
+
+
+ com.lingniu.ingest
+ vehicle-state-service
+
+
+ com.lingniu.ingest
+ vehicle-stat-service
+
+
+ org.springframework.boot
+ spring-boot-starter-web
+
+
+ org.springframework.boot
+ spring-boot-starter-test
+ test
+
+
+
+
+ vehicle-analytics-app
+
+
+ org.springframework.boot
+ spring-boot-maven-plugin
+
+ --sun-misc-unsafe-memory-access=allow
+
+
+
+
+ repackage
+
+
+
+
+
+
+
+```
+
+- [ ] **Step 6: Verify the modules are visible to Maven**
+
+Run:
+
+```bash
+export JAVA_HOME=/opt/homebrew/opt/openjdk@25/libexec/openjdk.jdk/Contents/Home
+export PATH="$JAVA_HOME/bin:/opt/homebrew/bin:$PATH"
+mvn -pl :gb32960-ingest-app,:vehicle-history-app,:vehicle-analytics-app -am validate
+```
+
+Expected: BUILD SUCCESS.
+
+- [ ] **Step 7: Commit**
+
+```bash
+git add pom.xml modules/apps/gb32960-ingest-app/pom.xml modules/apps/vehicle-history-app/pom.xml modules/apps/vehicle-analytics-app/pom.xml
+git commit -m "build: add split service app modules"
+```
+
+---
+
+### Task 2: Add App Entrypoints and Runtime Defaults
+
+**Files:**
+- Create: `modules/apps/gb32960-ingest-app/src/main/java/com/lingniu/ingest/gb32960app/Gb32960IngestApplication.java`
+- Create: `modules/apps/gb32960-ingest-app/src/main/resources/application.yml`
+- Create: `modules/apps/vehicle-history-app/src/main/java/com/lingniu/ingest/historyapp/VehicleHistoryApplication.java`
+- Create: `modules/apps/vehicle-history-app/src/main/resources/application.yml`
+- Create: `modules/apps/vehicle-analytics-app/src/main/java/com/lingniu/ingest/analyticsapp/VehicleAnalyticsApplication.java`
+- Create: `modules/apps/vehicle-analytics-app/src/main/resources/application.yml`
+
+- [ ] **Step 1: Create the GB32960 ingest main class**
+
+Create `modules/apps/gb32960-ingest-app/src/main/java/com/lingniu/ingest/gb32960app/Gb32960IngestApplication.java`:
+
+```java
+package com.lingniu.ingest.gb32960app;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+
+@SpringBootApplication(scanBasePackages = "com.lingniu.ingest")
+public class Gb32960IngestApplication {
+
+ private static final Logger log = LoggerFactory.getLogger(Gb32960IngestApplication.class);
+
+ public static void main(String[] args) {
+ try {
+ SpringApplication.run(Gb32960IngestApplication.class, args);
+ } catch (Throwable t) {
+ log.error("GB32960 ingest application failed to start, forcing JVM exit", t);
+ System.exit(1);
+ }
+ }
+}
+```
+
+- [ ] **Step 2: Create the GB32960 ingest config**
+
+Create `modules/apps/gb32960-ingest-app/src/main/resources/application.yml`:
+
+```yaml
+spring:
+ application:
+ name: gb32960-ingest-app
+ threads:
+ virtual:
+ enabled: true
+
+server:
+ port: ${HTTP_PORT:20100}
+
+lingniu:
+ ingest:
+ gb32960:
+ enabled: true
+ port: ${GB32960_PORT:32960}
+ boss-threads: ${GB32960_BOSS_THREADS:1}
+ worker-threads: ${GB32960_WORKER_THREADS:0}
+ auth:
+ enabled: ${GB32960_AUTH_ENABLED:false}
+ case-sensitive: false
+ whitelist: []
+ platforms:
+ - username: ${GB32960_PLATFORM_USER_HYUNDAI:Hyundai}
+ password: ${GB32960_PLATFORM_PWD_HYUNDAI:f2e3445d7cda409fb4f278f6fb890734}
+ allowed-ips:
+ - ${GB32960_PLATFORM_IP_HYUNDAI:8.134.95.166}
+ description: 外部下级平台 - Hyundai
+ tls:
+ enabled: ${GB32960_TLS_ENABLED:false}
+ cert-path: ${GB32960_TLS_CERT:}
+ key-path: ${GB32960_TLS_KEY:}
+ trust-cert-path: ${GB32960_TLS_CA:}
+ require-client-auth: true
+ vendor-extensions:
+ - name: guangdong-fc
+ match:
+ platform-accounts:
+ - Hyundai
+ vin-prefixes: []
+ vins: []
+ parse:
+ lenient-block-failure: true
+ diagnostics:
+ max-logged-frame-keys-per-channel: ${GB32960_DIAGNOSTICS_MAX_LOGGED_FRAME_KEYS_PER_CHANNEL:4096}
+ pipeline:
+ disruptor:
+ ring-buffer-size: ${PIPELINE_RING_BUFFER_SIZE:131072}
+ wait-strategy: ${PIPELINE_WAIT_STRATEGY:yielding}
+ producer-type: multi
+ dedup:
+ enabled: true
+ cache-size: 200000
+ ttl-seconds: 600
+ rate-limit:
+ per-vin-qps: 50
+ session:
+ store: ${SESSION_STORE:memory}
+ ttl: ${SESSION_TTL:30m}
+ identity:
+ store: ${VEHICLE_IDENTITY_STORE:file}
+ file:
+ path: ${VEHICLE_IDENTITY_FILE:./data/vehicle-identity.jsonl}
+ sink:
+ mq:
+ enabled: ${KAFKA_ENABLED:true}
+ type: kafka
+ bootstrap-servers: ${KAFKA_BROKERS:114.55.58.251:9092}
+ compression-type: zstd
+ linger-ms: 20
+ batch-size: 65536
+ acks: all
+ enable-idempotence: true
+ node-id: ${KAFKA_NODE_ID:gb32960-ingest-local}
+ topics:
+ realtime: ${KAFKA_TOPIC_GB32960_EVENT:vehicle.event.gb32960.v1}
+ location: ${KAFKA_TOPIC_GB32960_EVENT:vehicle.event.gb32960.v1}
+ alarm: ${KAFKA_TOPIC_GB32960_EVENT:vehicle.event.gb32960.v1}
+ session: ${KAFKA_TOPIC_GB32960_EVENT:vehicle.event.gb32960.v1}
+ media-meta: ${KAFKA_TOPIC_MEDIA_META:vehicle.media.meta.v1}
+ raw-archive: ${KAFKA_TOPIC_GB32960_RAW:vehicle.raw.gb32960.v1}
+ dlq: ${KAFKA_TOPIC_GB32960_DLQ:vehicle.dlq.gb32960.v1}
+ consumer:
+ enabled: false
+ event-file-store:
+ enabled: false
+ event-history:
+ enabled: false
+ vehicle-state:
+ enabled: false
+ vehicle-stat:
+ enabled: false
+
+management:
+ endpoints:
+ web:
+ exposure:
+ include: health,info,metrics,prometheus,env
+ health:
+ redis:
+ enabled: ${MANAGEMENT_HEALTH_REDIS_ENABLED:false}
+ metrics:
+ tags:
+ application: gb32960-ingest-app
+
+logging:
+ level:
+ root: INFO
+ com.lingniu.ingest: INFO
+```
+
+- [ ] **Step 3: Create the history app main class**
+
+Create `modules/apps/vehicle-history-app/src/main/java/com/lingniu/ingest/historyapp/VehicleHistoryApplication.java`:
+
+```java
+package com.lingniu.ingest.historyapp;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+
+@SpringBootApplication(scanBasePackages = "com.lingniu.ingest")
+public class VehicleHistoryApplication {
+
+ private static final Logger log = LoggerFactory.getLogger(VehicleHistoryApplication.class);
+
+ public static void main(String[] args) {
+ try {
+ SpringApplication.run(VehicleHistoryApplication.class, args);
+ } catch (Throwable t) {
+ log.error("Vehicle history application failed to start, forcing JVM exit", t);
+ System.exit(1);
+ }
+ }
+}
+```
+
+- [ ] **Step 4: Create the history app config**
+
+Create `modules/apps/vehicle-history-app/src/main/resources/application.yml`:
+
+```yaml
+spring:
+ application:
+ name: vehicle-history-app
+ threads:
+ virtual:
+ enabled: true
+
+server:
+ port: ${HTTP_PORT:20200}
+
+lingniu:
+ ingest:
+ gb32960:
+ enabled: false
+ sink:
+ mq:
+ enabled: ${KAFKA_ENABLED:true}
+ type: kafka
+ bootstrap-servers: ${KAFKA_BROKERS:114.55.58.251:9092}
+ topics:
+ realtime: ${KAFKA_TOPIC_GB32960_EVENT:vehicle.event.gb32960.v1}
+ raw-archive: ${KAFKA_TOPIC_GB32960_RAW:vehicle.raw.gb32960.v1}
+ dlq: ${KAFKA_TOPIC_GB32960_DLQ:vehicle.dlq.gb32960.v1}
+ consumer:
+ enabled: ${KAFKA_CONSUMER_ENABLED:true}
+ client-id-prefix: ${KAFKA_CONSUMER_CLIENT_ID_PREFIX:vehicle-history}
+ auto-offset-reset: ${KAFKA_CONSUMER_AUTO_OFFSET_RESET:earliest}
+ max-poll-records: ${KAFKA_CONSUMER_MAX_POLL_RECORDS:500}
+ bindings:
+ eventHistoryEnvelopeConsumerProcessor:
+ enabled: true
+ group-id: ${KAFKA_GROUP_HISTORY:vehicle-history}
+ topics:
+ - ${KAFKA_TOPIC_GB32960_EVENT:vehicle.event.gb32960.v1}
+ - ${KAFKA_TOPIC_GB32960_RAW:vehicle.raw.gb32960.v1}
+ archive:
+ enabled: ${SINK_ARCHIVE_ENABLED:true}
+ type: local
+ path: ${SINK_ARCHIVE_PATH:./archive/}
+ event-file-store:
+ enabled: ${EVENT_FILE_STORE_ENABLED:true}
+ path: ${EVENT_FILE_STORE_PATH:./target/event-store/}
+ zone-id: ${EVENT_FILE_STORE_ZONE_ID:Asia/Shanghai}
+ batch-size: ${EVENT_FILE_STORE_BATCH_SIZE:500}
+ flush-interval-millis: ${EVENT_FILE_STORE_FLUSH_INTERVAL_MILLIS:1000}
+ event-history:
+ enabled: true
+ vehicle-state:
+ enabled: false
+ vehicle-stat:
+ enabled: false
+
+management:
+ endpoints:
+ web:
+ exposure:
+ include: health,info,metrics,prometheus,env
+ health:
+ redis:
+ enabled: ${MANAGEMENT_HEALTH_REDIS_ENABLED:false}
+ metrics:
+ tags:
+ application: vehicle-history-app
+```
+
+- [ ] **Step 5: Create the analytics app main class**
+
+Create `modules/apps/vehicle-analytics-app/src/main/java/com/lingniu/ingest/analyticsapp/VehicleAnalyticsApplication.java`:
+
+```java
+package com.lingniu.ingest.analyticsapp;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+
+@SpringBootApplication(scanBasePackages = "com.lingniu.ingest")
+public class VehicleAnalyticsApplication {
+
+ private static final Logger log = LoggerFactory.getLogger(VehicleAnalyticsApplication.class);
+
+ public static void main(String[] args) {
+ try {
+ SpringApplication.run(VehicleAnalyticsApplication.class, args);
+ } catch (Throwable t) {
+ log.error("Vehicle analytics application failed to start, forcing JVM exit", t);
+ System.exit(1);
+ }
+ }
+}
+```
+
+- [ ] **Step 6: Create the analytics app config**
+
+Create `modules/apps/vehicle-analytics-app/src/main/resources/application.yml`:
+
+```yaml
+spring:
+ application:
+ name: vehicle-analytics-app
+ threads:
+ virtual:
+ enabled: true
+
+server:
+ port: ${HTTP_PORT:20300}
+
+lingniu:
+ ingest:
+ gb32960:
+ enabled: false
+ sink:
+ mq:
+ enabled: ${KAFKA_ENABLED:true}
+ type: kafka
+ bootstrap-servers: ${KAFKA_BROKERS:114.55.58.251:9092}
+ topics:
+ realtime: ${KAFKA_TOPIC_GB32960_EVENT:vehicle.event.gb32960.v1}
+ dlq: ${KAFKA_TOPIC_GB32960_DLQ:vehicle.dlq.gb32960.v1}
+ consumer:
+ enabled: ${KAFKA_CONSUMER_ENABLED:true}
+ client-id-prefix: ${KAFKA_CONSUMER_CLIENT_ID_PREFIX:vehicle-analytics}
+ auto-offset-reset: ${KAFKA_CONSUMER_AUTO_OFFSET_RESET:earliest}
+ max-poll-records: ${KAFKA_CONSUMER_MAX_POLL_RECORDS:500}
+ bindings:
+ vehicleStateEnvelopeConsumerProcessor:
+ enabled: ${VEHICLE_STATE_ENABLED:false}
+ group-id: ${KAFKA_GROUP_STATE:vehicle-state}
+ topics:
+ - ${KAFKA_TOPIC_GB32960_EVENT:vehicle.event.gb32960.v1}
+ vehicleStatEnvelopeConsumerProcessor:
+ enabled: ${VEHICLE_STAT_ENABLED:true}
+ group-id: ${KAFKA_GROUP_STAT:vehicle-stat}
+ topics:
+ - ${KAFKA_TOPIC_GB32960_EVENT:vehicle.event.gb32960.v1}
+ archive:
+ enabled: false
+ event-file-store:
+ enabled: false
+ event-history:
+ enabled: false
+ vehicle-state:
+ enabled: ${VEHICLE_STATE_ENABLED:false}
+ vehicle-stat:
+ enabled: ${VEHICLE_STAT_ENABLED:true}
+ file-path: ${VEHICLE_STAT_FILE_PATH:./target/vehicle-stat/}
+ zone-id: ${VEHICLE_STAT_ZONE_ID:Asia/Shanghai}
+
+management:
+ endpoints:
+ web:
+ exposure:
+ include: health,info,metrics,prometheus,env
+ health:
+ redis:
+ enabled: ${MANAGEMENT_HEALTH_REDIS_ENABLED:false}
+ metrics:
+ tags:
+ application: vehicle-analytics-app
+```
+
+- [ ] **Step 7: Build all three app jars**
+
+Run:
+
+```bash
+export JAVA_HOME=/opt/homebrew/opt/openjdk@25/libexec/openjdk.jdk/Contents/Home
+export PATH="$JAVA_HOME/bin:/opt/homebrew/bin:$PATH"
+mvn -pl :gb32960-ingest-app,:vehicle-history-app,:vehicle-analytics-app -am package -Dmaven.test.skip=true
+```
+
+Expected: BUILD SUCCESS and jars under each app module's `target/`.
+
+- [ ] **Step 8: Commit**
+
+```bash
+git add modules/apps/gb32960-ingest-app modules/apps/vehicle-history-app modules/apps/vehicle-analytics-app
+git commit -m "feat: add split service entrypoints"
+```
+
+---
+
+### Task 3: Add Composition Tests for Service Boundaries
+
+**Files:**
+- Create: `modules/apps/gb32960-ingest-app/src/test/java/com/lingniu/ingest/gb32960app/Gb32960IngestAppCompositionTest.java`
+- Create: `modules/apps/vehicle-history-app/src/test/java/com/lingniu/ingest/historyapp/VehicleHistoryAppCompositionTest.java`
+- Create: `modules/apps/vehicle-analytics-app/src/test/java/com/lingniu/ingest/analyticsapp/VehicleAnalyticsAppCompositionTest.java`
+
+- [ ] **Step 1: Write GB32960 ingest composition test**
+
+Create `modules/apps/gb32960-ingest-app/src/test/java/com/lingniu/ingest/gb32960app/Gb32960IngestAppCompositionTest.java`:
+
+```java
+package com.lingniu.ingest.gb32960app;
+
+import com.lingniu.ingest.eventfilestore.EventFileStore;
+import com.lingniu.ingest.protocol.gb32960.config.Gb32960AutoConfiguration;
+import com.lingniu.ingest.protocol.gb32960.inbound.Gb32960NettyServer;
+import com.lingniu.ingest.sink.archive.ArchiveStore;
+import com.lingniu.ingest.sink.mq.KafkaEventSink;
+import org.junit.jupiter.api.Test;
+import org.springframework.boot.autoconfigure.AutoConfigurations;
+import org.springframework.boot.test.context.runner.ApplicationContextRunner;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class Gb32960IngestAppCompositionTest {
+
+ @Test
+ void gb32960IngestRuntimeStartsProtocolAndKafkaWithoutHistoryStorage() {
+ new ApplicationContextRunner()
+ .withConfiguration(AutoConfigurations.of(Gb32960AutoConfiguration.class))
+ .withPropertyValues(
+ "lingniu.ingest.gb32960.enabled=true",
+ "lingniu.ingest.gb32960.port=0",
+ "lingniu.ingest.sink.mq.enabled=false",
+ "lingniu.ingest.sink.archive.enabled=false",
+ "lingniu.ingest.event-file-store.enabled=false",
+ "lingniu.ingest.event-history.enabled=false",
+ "lingniu.ingest.vehicle-state.enabled=false",
+ "lingniu.ingest.vehicle-stat.enabled=false")
+ .run(context -> {
+ assertThat(context).hasSingleBean(Gb32960NettyServer.class);
+ assertThat(context).doesNotHaveBean(ArchiveStore.class);
+ assertThat(context).doesNotHaveBean(EventFileStore.class);
+ assertThat(context).doesNotHaveBean(KafkaEventSink.class);
+ });
+ }
+}
+```
+
+If this test does not create required supporting beans, add the same auto-config classes used by existing bootstrap tests: `IngestCoreAutoConfiguration`, `SessionCoreAutoConfiguration`, and `VehicleIdentityAutoConfiguration`.
+
+- [ ] **Step 2: Run GB32960 ingest composition test**
+
+Run:
+
+```bash
+mvn -pl :gb32960-ingest-app -Dtest=Gb32960IngestAppCompositionTest test
+```
+
+Expected after adding required auto-config classes: PASS.
+
+- [ ] **Step 3: Write history composition test**
+
+Create `modules/apps/vehicle-history-app/src/test/java/com/lingniu/ingest/historyapp/VehicleHistoryAppCompositionTest.java`:
+
+```java
+package com.lingniu.ingest.historyapp;
+
+import com.lingniu.ingest.eventfilestore.EventFileStore;
+import com.lingniu.ingest.eventhistory.EventHistoryController;
+import com.lingniu.ingest.protocol.gb32960.inbound.Gb32960NettyServer;
+import com.lingniu.ingest.sink.archive.ArchiveStore;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.springframework.boot.autoconfigure.AutoConfigurations;
+import org.springframework.boot.test.context.runner.ApplicationContextRunner;
+
+import java.nio.file.Path;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class VehicleHistoryAppCompositionTest {
+
+ @TempDir
+ Path tempDir;
+
+ @Test
+ void historyRuntimeCreatesStorageAndQueryBeansWithoutProtocolListener() {
+ new ApplicationContextRunner()
+ .withPropertyValues(
+ "lingniu.ingest.gb32960.enabled=false",
+ "lingniu.ingest.sink.archive.enabled=true",
+ "lingniu.ingest.sink.archive.path=" + tempDir.resolve("archive"),
+ "lingniu.ingest.event-file-store.enabled=true",
+ "lingniu.ingest.event-file-store.path=" + tempDir.resolve("event-store"),
+ "lingniu.ingest.event-history.enabled=true",
+ "lingniu.ingest.vehicle-state.enabled=false",
+ "lingniu.ingest.vehicle-stat.enabled=false",
+ "lingniu.ingest.sink.mq.enabled=false")
+ .withConfiguration(AutoConfigurations.of(
+ com.lingniu.ingest.sink.archive.config.SinkArchiveAutoConfiguration.class,
+ com.lingniu.ingest.eventfilestore.config.EventFileStoreAutoConfiguration.class,
+ com.lingniu.ingest.eventhistory.config.EventHistoryAutoConfiguration.class))
+ .run(context -> {
+ assertThat(context).hasSingleBean(ArchiveStore.class);
+ assertThat(context).hasSingleBean(EventFileStore.class);
+ assertThat(context).hasSingleBean(EventHistoryController.class);
+ assertThat(context).doesNotHaveBean(Gb32960NettyServer.class);
+ });
+ }
+}
+```
+
+- [ ] **Step 4: Run history composition test**
+
+Run:
+
+```bash
+mvn -pl :vehicle-history-app -Dtest=VehicleHistoryAppCompositionTest test
+```
+
+Expected: PASS.
+
+- [ ] **Step 5: Write analytics composition test**
+
+Create `modules/apps/vehicle-analytics-app/src/test/java/com/lingniu/ingest/analyticsapp/VehicleAnalyticsAppCompositionTest.java`:
+
+```java
+package com.lingniu.ingest.analyticsapp;
+
+import com.lingniu.ingest.eventfilestore.EventFileStore;
+import com.lingniu.ingest.protocol.gb32960.inbound.Gb32960NettyServer;
+import com.lingniu.ingest.vehiclestat.DailyVehicleStatService;
+import com.lingniu.ingest.vehiclestate.VehicleStateRepository;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.springframework.boot.autoconfigure.AutoConfigurations;
+import org.springframework.boot.test.context.runner.ApplicationContextRunner;
+
+import java.nio.file.Path;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class VehicleAnalyticsAppCompositionTest {
+
+ @TempDir
+ Path tempDir;
+
+ @Test
+ void analyticsRuntimeCreatesStatisticsWithoutProtocolOrHistoryStorage() {
+ new ApplicationContextRunner()
+ .withPropertyValues(
+ "lingniu.ingest.gb32960.enabled=false",
+ "lingniu.ingest.sink.archive.enabled=false",
+ "lingniu.ingest.event-file-store.enabled=false",
+ "lingniu.ingest.event-history.enabled=false",
+ "lingniu.ingest.vehicle-state.enabled=false",
+ "lingniu.ingest.vehicle-stat.enabled=true",
+ "lingniu.ingest.vehicle-stat.file-path=" + tempDir.resolve("vehicle-stat"),
+ "lingniu.ingest.sink.mq.enabled=false")
+ .withConfiguration(AutoConfigurations.of(
+ com.lingniu.ingest.vehiclestat.config.VehicleStatAutoConfiguration.class,
+ com.lingniu.ingest.vehiclestate.config.VehicleStateAutoConfiguration.class))
+ .run(context -> {
+ assertThat(context).hasSingleBean(DailyVehicleStatService.class);
+ assertThat(context).doesNotHaveBean(VehicleStateRepository.class);
+ assertThat(context).doesNotHaveBean(Gb32960NettyServer.class);
+ assertThat(context).doesNotHaveBean(EventFileStore.class);
+ });
+ }
+}
+```
+
+- [ ] **Step 6: Run analytics composition test**
+
+Run:
+
+```bash
+mvn -pl :vehicle-analytics-app -Dtest=VehicleAnalyticsAppCompositionTest test
+```
+
+Expected: PASS.
+
+- [ ] **Step 7: Commit**
+
+```bash
+git add modules/apps/gb32960-ingest-app/src/test modules/apps/vehicle-history-app/src/test modules/apps/vehicle-analytics-app/src/test
+git commit -m "test: cover split service composition"
+```
+
+---
+
+### Task 4: Version Kafka Topics for GB32960 Raw/Event/DLQ
+
+**Files:**
+- Modify: `modules/sinks/sink-mq/src/main/java/com/lingniu/ingest/sink/mq/SinkMqProperties.java`
+- Modify: `modules/sinks/sink-mq/src/main/java/com/lingniu/ingest/sink/mq/TopicRouter.java`
+- Modify: `modules/sinks/sink-mq/src/main/java/com/lingniu/ingest/sink/mq/KafkaEventSink.java`
+- Create or modify: `modules/sinks/sink-mq/src/test/java/com/lingniu/ingest/sink/mq/TopicRouterTest.java`
+- Create or modify: `modules/sinks/sink-mq/src/test/java/com/lingniu/ingest/sink/mq/KafkaEventSinkTest.java`
+
+- [ ] **Step 1: Write topic routing tests**
+
+Create `modules/sinks/sink-mq/src/test/java/com/lingniu/ingest/sink/mq/TopicRouterTest.java` if it does not exist. Add tests proving GB32960 raw and normalized events can route to versioned topics. Use the existing `VehicleEvent` constructors from `EventFileStoreSinkTest` and `EnvelopeMapperTelemetrySnapshotTest` as examples.
+
+Expected test names:
+
+```java
+@Test
+void rawArchiveRoutesToVersionedGb32960RawTopic() {
+ SinkMqProperties.Topics topics = new SinkMqProperties.Topics();
+ topics.setRawArchive("vehicle.raw.gb32960.v1");
+ TopicRouter router = new TopicRouter(topics);
+
+ VehicleEvent.RawArchive raw = new VehicleEvent.RawArchive(
+ "event-1",
+ "trace-1",
+ ProtocolId.GB32960,
+ "VIN001",
+ Instant.parse("2026-06-23T07:00:00Z"),
+ 0x02,
+ new byte[] {0x23, 0x23},
+ Map.of("platformAccount", "Hyundai"));
+
+ assertThat(router.route(raw)).isEqualTo("vehicle.raw.gb32960.v1");
+}
+```
+
+Also add a normalized realtime event test:
+
+```java
+@Test
+void realtimeRoutesToVersionedGb32960EventTopic() {
+ SinkMqProperties.Topics topics = new SinkMqProperties.Topics();
+ topics.setRealtime("vehicle.event.gb32960.v1");
+ TopicRouter router = new TopicRouter(topics);
+
+ VehicleEvent.Realtime realtime = new VehicleEvent.Realtime(
+ "event-2",
+ "trace-2",
+ ProtocolId.GB32960,
+ "VIN001",
+ Instant.parse("2026-06-23T07:00:01Z"),
+ 0x02,
+ Map.of("soc", "80"),
+ Map.of("platformAccount", "Hyundai"));
+
+ assertThat(router.route(realtime)).isEqualTo("vehicle.event.gb32960.v1");
+}
+```
+
+Adjust constructor arguments to match the current sealed `VehicleEvent` definitions in `modules/core/ingest-api/src/main/java/com/lingniu/ingest/api/event/VehicleEvent.java`.
+
+- [ ] **Step 2: Run topic router test and verify failure if topic defaults are still old**
+
+Run:
+
+```bash
+mvn -pl :sink-mq -Dtest=TopicRouterTest test
+```
+
+Expected before implementation: compile failure if constructors are wrong or assertion failure if defaults are old. Fix constructors first; keep routing assertions.
+
+- [ ] **Step 3: Update `SinkMqProperties.Topics` defaults**
+
+In `SinkMqProperties.Topics`, change defaults:
+
+```java
+private String realtime = "vehicle.event.gb32960.v1";
+private String location = "vehicle.event.gb32960.v1";
+private String alarm = "vehicle.event.gb32960.v1";
+private String session = "vehicle.event.gb32960.v1";
+private String mediaMeta = "vehicle.media.meta.v1";
+private String rawArchive = "vehicle.raw.gb32960.v1";
+private String dlq = "vehicle.dlq.gb32960.v1";
+```
+
+- [ ] **Step 4: Allow raw archive records in KafkaEventSink**
+
+Change `KafkaEventSink.accepts()` to:
+
+```java
+@Override
+public boolean accepts(VehicleEvent event) {
+ return true;
+}
+```
+
+Update the JavaDoc above it to say Kafka carries both normalized events and raw archive records in split-service mode.
+
+- [ ] **Step 5: Run sink-mq tests**
+
+Run:
+
+```bash
+mvn -pl :sink-mq test
+```
+
+Expected: PASS.
+
+- [ ] **Step 6: Commit**
+
+```bash
+git add modules/sinks/sink-mq/src/main/java/com/lingniu/ingest/sink/mq/SinkMqProperties.java \
+ modules/sinks/sink-mq/src/main/java/com/lingniu/ingest/sink/mq/TopicRouter.java \
+ modules/sinks/sink-mq/src/main/java/com/lingniu/ingest/sink/mq/KafkaEventSink.java \
+ modules/sinks/sink-mq/src/test/java/com/lingniu/ingest/sink/mq/TopicRouterTest.java \
+ modules/sinks/sink-mq/src/test/java/com/lingniu/ingest/sink/mq/KafkaEventSinkTest.java
+git commit -m "feat: route gb32960 kafka topics"
+```
+
+---
+
+### Task 5: Introduce an ACK-Aware GB32960 Dispatch Boundary
+
+**Files:**
+- Modify: `modules/core/ingest-core/src/main/java/com/lingniu/ingest/core/dispatcher/Dispatcher.java`
+- Modify: `modules/core/ingest-core/src/main/java/com/lingniu/ingest/core/concurrency/DisruptorEventBus.java`
+- Modify: `modules/protocols/protocol-gb32960/src/main/java/com/lingniu/ingest/protocol/gb32960/inbound/Gb32960ChannelHandler.java`
+- Create: `modules/protocols/protocol-gb32960/src/test/java/com/lingniu/ingest/protocol/gb32960/inbound/Gb32960ChannelHandlerAckBoundaryTest.java`
+
+Design choice for first implementation:
+
+- Keep `Dispatcher.dispatch(RawFrame)` for existing async protocols and `bootstrap-all`.
+- Add `Dispatcher.dispatchAndAwait(RawFrame, Duration)` returning `CompletableFuture` or throwing a checked runtime exception on timeout/failure.
+- Implement `DisruptorEventBus.publishAndAwait(VehicleEvent, Duration)` by synchronously calling accepted sinks' `publish()` futures and waiting for all to complete.
+- Use the synchronous path only from GB32960 report/heartbeat/session commands that require ACK after Kafka success.
+
+- [ ] **Step 1: Write failing ACK boundary test**
+
+Create `Gb32960ChannelHandlerAckBoundaryTest` using `EmbeddedChannel` and a fake dispatcher that records whether ACK happens before the future completes.
+
+Core assertion:
+
+```java
+@Test
+void realtimeReportAckWaitsForKafkaDurabilityBoundary() {
+ CompletableFuture durability = new CompletableFuture<>();
+ RecordingAckService ackService = new RecordingAckService();
+ AwaitingDispatcher dispatcher = new AwaitingDispatcher(durability);
+ Gb32960ChannelHandler handler = handlerWith(dispatcher, ackService);
+
+ EmbeddedChannel channel = new EmbeddedChannel(handler);
+ channel.writeInbound(validRealtimeFrameBytes());
+
+ assertThat(ackService.writtenTags()).isEmpty();
+
+ durability.complete(null);
+ channel.runPendingTasks();
+
+ assertThat(ackService.writtenTags()).contains("report-ack-0x2");
+}
+```
+
+Implementation notes:
+
+- Reuse valid frame fixture builders from existing GB32960 codec tests.
+- If `Gb32960AckService` is hard to spy because methods are concrete, subclass it in test and override `writeResponse`/`writeAndClose` to record tags instead of writing bytes.
+- If `Dispatcher` is not subclassable, extract a small interface such as `FrameDispatcher` in `ingest-core` and adapt `Dispatcher` to implement it.
+
+- [ ] **Step 2: Run the failing test**
+
+Run:
+
+```bash
+mvn -pl :protocol-gb32960 -Dtest=Gb32960ChannelHandlerAckBoundaryTest test
+```
+
+Expected before implementation: FAIL because current handler writes ACK before dispatch completion.
+
+- [ ] **Step 3: Add await API to event bus**
+
+In `DisruptorEventBus`, add:
+
+```java
+public CompletableFuture publishAndAwait(VehicleEvent event) {
+ List> futures = sinks.stream()
+ .filter(sink -> sink.accepts(event))
+ .map(sink -> sink.publish(event))
+ .toList();
+ if (futures.isEmpty()) {
+ return CompletableFuture.completedFuture(null);
+ }
+ return CompletableFuture.allOf(futures.toArray(CompletableFuture[]::new));
+}
+```
+
+This requires promoting `sinks` from constructor-local variable to a field:
+
+```java
+private final List sinks;
+```
+
+Set it in the constructor:
+
+```java
+this.sinks = List.copyOf(sinks);
+```
+
+- [ ] **Step 4: Add await API to Dispatcher**
+
+In `Dispatcher`, add:
+
+```java
+public CompletableFuture dispatchAndAwait(RawFrame frame) {
+ List events = mapFrameToEvents(frame);
+ CompletableFuture[] futures = events.stream()
+ .map(eventBus::publishAndAwait)
+ .toArray(CompletableFuture[]::new);
+ return CompletableFuture.allOf(futures);
+}
+```
+
+Refactor existing `dispatch(RawFrame)` to reuse `mapFrameToEvents(frame)`:
+
+```java
+public void dispatch(RawFrame frame) {
+ for (VehicleEvent e : mapFrameToEvents(frame)) {
+ eventBus.publish(e);
+ }
+}
+```
+
+If `Dispatcher` currently interleaves raw archive event publication and handler mapping, move that logic into a private method:
+
+```java
+private List mapFrameToEvents(RawFrame frame) {
+ List events = new ArrayList<>();
+ events.add(toRawArchive(frame));
+ events.addAll(handlerEvents(frame));
+ return events;
+}
+```
+
+Use exact existing helper names from `Dispatcher.java`; do not duplicate mapping logic.
+
+- [ ] **Step 5: Move GB32960 ACK after dispatch completion**
+
+In `Gb32960ChannelHandler.channelRead0`, for commands that currently ACK before `dispatcher.dispatch(rf)`, build `RawFrame` first, call `dispatcher.dispatchAndAwait(rf)`, and write ACK in the completion callback:
+
+```java
+dispatcher.dispatchAndAwait(rf).whenComplete((ignored, ex) -> {
+ if (ex != null) {
+ log.warn("[gb32960] dispatch failed before ack peer={} vin={} cmd=0x{}",
+ addr(ctx), vin, Integer.toHexString(cmd.code()), ex);
+ ctx.close();
+ return;
+ }
+ writeAckForCommand(ctx, msg, rawVin);
+});
+```
+
+Extract current ACK logic into command-specific method:
+
+```java
+private void writeAckForCommand(ChannelHandlerContext ctx, Gb32960Message msg, byte[] rawVin) {
+ switch (msg.header().command()) {
+ case VEHICLE_LOGIN -> writeVehicleLoginAck(ctx, msg, rawVin);
+ case VEHICLE_LOGOUT -> writeVehicleLogoutAck(ctx, msg, rawVin);
+ case PLATFORM_LOGIN -> writePlatformLoginAck(ctx, msg, rawVin);
+ case PLATFORM_LOGOUT -> writePlatformLogoutAck(ctx, msg, rawVin);
+ case REALTIME_REPORT, RESEND_REPORT, HEARTBEAT -> writeReportOrHeartbeatAck(ctx, msg, rawVin);
+ case TIME_CALIBRATION -> writeTimeCalibrationAck(ctx, msg, rawVin);
+ default -> { }
+ }
+}
+```
+
+Keep authorization failure ACKs immediate because they are not accepted frames and do not cross the Kafka durability boundary.
+
+- [ ] **Step 6: Run ACK boundary test**
+
+Run:
+
+```bash
+mvn -pl :protocol-gb32960 -Dtest=Gb32960ChannelHandlerAckBoundaryTest test
+```
+
+Expected: PASS.
+
+- [ ] **Step 7: Run core dispatcher tests**
+
+Run:
+
+```bash
+mvn -pl :ingest-core test
+```
+
+Expected: PASS.
+
+- [ ] **Step 8: Commit**
+
+```bash
+git add modules/core/ingest-core/src/main/java/com/lingniu/ingest/core/dispatcher/Dispatcher.java \
+ modules/core/ingest-core/src/main/java/com/lingniu/ingest/core/concurrency/DisruptorEventBus.java \
+ modules/protocols/protocol-gb32960/src/main/java/com/lingniu/ingest/protocol/gb32960/inbound/Gb32960ChannelHandler.java \
+ modules/protocols/protocol-gb32960/src/test/java/com/lingniu/ingest/protocol/gb32960/inbound/Gb32960ChannelHandlerAckBoundaryTest.java
+git commit -m "feat: ack gb32960 after kafka boundary"
+```
+
+---
+
+### Task 6: Wire History and Analytics Consumer Bindings
+
+**Files:**
+- Modify: `modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/EventHistoryEnvelopeIngestor.java`
+- Modify: `modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/VehicleStatEnvelopeIngestor.java`
+- Modify: `modules/services/vehicle-state-service/src/main/java/com/lingniu/ingest/vehiclestate/VehicleStateEnvelopeIngestor.java`
+- Modify tests under the same service modules.
+
+- [ ] **Step 1: Verify event-history accepts raw and normalized envelopes**
+
+Run:
+
+```bash
+rg -n "RawArchive|raw_archive|TelemetrySnapshot|Envelope" modules/services/event-history-service/src/main/java modules/services/event-history-service/src/test/java
+```
+
+Expected: identify whether `EventHistoryEnvelopeIngestor` already persists raw archive envelopes. If raw archive envelopes are ignored, continue with Step 2.
+
+- [ ] **Step 2: Add event-history raw archive ingestion test**
+
+Add a test to the existing event-history test suite that creates an envelope with `raw_archive` populated and asserts an `EventFileRecord` with `eventType = RAW_ARCHIVE` is written with `rawArchiveUri`.
+
+Expected assertion shape:
+
+```java
+assertThat(record.eventType()).isEqualTo("RAW_ARCHIVE");
+assertThat(record.rawArchiveUri()).isEqualTo("archive://2026/06/23/GB32960/VIN001/event-1.bin");
+```
+
+- [ ] **Step 3: Implement raw archive ingestion if missing**
+
+In `EventHistoryEnvelopeIngestor`, map raw archive envelopes to `EventFileRecord` using the same conventions as `EventFileStoreSink`.
+
+Implementation rule:
+
+- Do not embed raw bytes into Parquet history records.
+- Store `rawArchiveUri`, event id, protocol, VIN, command, event time, and metadata.
+
+- [ ] **Step 4: Verify analytics consumers ignore raw topic records**
+
+Add or update tests for `VehicleStatEnvelopeIngestor` and `VehicleStateEnvelopeIngestor`:
+
+```java
+@Test
+void ignoresRawArchiveEnvelopeForAnalytics() {
+ ConsumerProcessResult result = ingestor.ingest(rawArchiveEnvelopeRecord());
+ assertThat(result.success()).isTrue();
+ assertThat(repositoryWrites()).isEmpty();
+}
+```
+
+Use the actual result type currently returned by the ingestor.
+
+- [ ] **Step 5: Run service consumer tests**
+
+Run:
+
+```bash
+mvn -pl :event-history-service,:vehicle-state-service,:vehicle-stat-service test
+```
+
+Expected: PASS.
+
+- [ ] **Step 6: Commit**
+
+```bash
+git add modules/services/event-history-service modules/services/vehicle-state-service modules/services/vehicle-stat-service
+git commit -m "feat: split history and analytics consumers"
+```
+
+---
+
+### Task 7: Add App Default Tests
+
+**Files:**
+- Create: `modules/apps/gb32960-ingest-app/src/test/java/com/lingniu/ingest/gb32960app/Gb32960IngestAppDefaultsTest.java`
+- Create: `modules/apps/vehicle-history-app/src/test/java/com/lingniu/ingest/historyapp/VehicleHistoryAppDefaultsTest.java`
+- Create: `modules/apps/vehicle-analytics-app/src/test/java/com/lingniu/ingest/analyticsapp/VehicleAnalyticsAppDefaultsTest.java`
+
+- [ ] **Step 1: Add YAML property loader helper to each test**
+
+Use this helper in each defaults test:
+
+```java
+private static Properties loadApplicationProperties() {
+ YamlPropertiesFactoryBean yaml = new YamlPropertiesFactoryBean();
+ yaml.setResources(new ClassPathResource("application.yml"));
+ Properties props = yaml.getObject();
+ assertThat(props).isNotNull();
+ return props;
+}
+```
+
+Imports:
+
+```java
+import org.springframework.beans.factory.config.YamlPropertiesFactoryBean;
+import org.springframework.core.io.ClassPathResource;
+import java.util.Properties;
+import static org.assertj.core.api.Assertions.assertThat;
+```
+
+- [ ] **Step 2: Assert GB32960 ingest defaults**
+
+Create `Gb32960IngestAppDefaultsTest`:
+
+```java
+@Test
+void gb32960IngestDefaultsOnlyEnableProtocolAndKafkaProducer() {
+ Properties props = loadApplicationProperties();
+
+ assertThat(props.getProperty("spring.application.name")).isEqualTo("gb32960-ingest-app");
+ assertThat(props.getProperty("lingniu.ingest.gb32960.enabled")).isEqualTo("true");
+ assertThat(props.getProperty("lingniu.ingest.gb32960.port")).isEqualTo("${GB32960_PORT:32960}");
+ assertThat(props.getProperty("lingniu.ingest.sink.mq.enabled")).isEqualTo("${KAFKA_ENABLED:true}");
+ assertThat(props.getProperty("lingniu.ingest.sink.mq.consumer.enabled")).isEqualTo("false");
+ assertThat(props.getProperty("lingniu.ingest.event-file-store.enabled")).isEqualTo("false");
+ assertThat(props.getProperty("lingniu.ingest.event-history.enabled")).isEqualTo("false");
+ assertThat(props.getProperty("lingniu.ingest.vehicle-state.enabled")).isEqualTo("false");
+ assertThat(props.getProperty("lingniu.ingest.vehicle-stat.enabled")).isEqualTo("false");
+}
+```
+
+- [ ] **Step 3: Assert history defaults**
+
+Create `VehicleHistoryAppDefaultsTest`:
+
+```java
+@Test
+void historyDefaultsEnableStorageAndHistoryConsumerOnly() {
+ Properties props = loadApplicationProperties();
+
+ assertThat(props.getProperty("spring.application.name")).isEqualTo("vehicle-history-app");
+ assertThat(props.getProperty("lingniu.ingest.gb32960.enabled")).isEqualTo("false");
+ assertThat(props.getProperty("lingniu.ingest.sink.archive.enabled")).isEqualTo("${SINK_ARCHIVE_ENABLED:true}");
+ assertThat(props.getProperty("lingniu.ingest.event-file-store.enabled")).isEqualTo("${EVENT_FILE_STORE_ENABLED:true}");
+ assertThat(props.getProperty("lingniu.ingest.event-history.enabled")).isEqualTo("true");
+ assertThat(props.getProperty("lingniu.ingest.vehicle-state.enabled")).isEqualTo("false");
+ assertThat(props.getProperty("lingniu.ingest.vehicle-stat.enabled")).isEqualTo("false");
+ assertThat(props.getProperty("lingniu.ingest.sink.mq.consumer.bindings.eventHistoryEnvelopeConsumerProcessor.group-id"))
+ .isEqualTo("${KAFKA_GROUP_HISTORY:vehicle-history}");
+}
+```
+
+- [ ] **Step 4: Assert analytics defaults**
+
+Create `VehicleAnalyticsAppDefaultsTest`:
+
+```java
+@Test
+void analyticsDefaultsEnableStatConsumerAndDisableProtocolAndHistoryStorage() {
+ Properties props = loadApplicationProperties();
+
+ assertThat(props.getProperty("spring.application.name")).isEqualTo("vehicle-analytics-app");
+ assertThat(props.getProperty("lingniu.ingest.gb32960.enabled")).isEqualTo("false");
+ assertThat(props.getProperty("lingniu.ingest.sink.archive.enabled")).isEqualTo("false");
+ assertThat(props.getProperty("lingniu.ingest.event-file-store.enabled")).isEqualTo("false");
+ assertThat(props.getProperty("lingniu.ingest.event-history.enabled")).isEqualTo("false");
+ assertThat(props.getProperty("lingniu.ingest.vehicle-stat.enabled")).isEqualTo("${VEHICLE_STAT_ENABLED:true}");
+ assertThat(props.getProperty("lingniu.ingest.sink.mq.consumer.bindings.vehicleStatEnvelopeConsumerProcessor.group-id"))
+ .isEqualTo("${KAFKA_GROUP_STAT:vehicle-stat}");
+}
+```
+
+- [ ] **Step 5: Run app default tests**
+
+Run:
+
+```bash
+mvn -pl :gb32960-ingest-app,:vehicle-history-app,:vehicle-analytics-app test
+```
+
+Expected: PASS.
+
+- [ ] **Step 6: Commit**
+
+```bash
+git add modules/apps/gb32960-ingest-app/src/test/java/com/lingniu/ingest/gb32960app/Gb32960IngestAppDefaultsTest.java \
+ modules/apps/vehicle-history-app/src/test/java/com/lingniu/ingest/historyapp/VehicleHistoryAppDefaultsTest.java \
+ modules/apps/vehicle-analytics-app/src/test/java/com/lingniu/ingest/analyticsapp/VehicleAnalyticsAppDefaultsTest.java
+git commit -m "test: assert split app defaults"
+```
+
+---
+
+### Task 8: End-to-End Verification With Local Kafka
+
+**Files:**
+- Create: `docs/operations/gb32960-service-split-runbook.md`
+
+- [ ] **Step 1: Build all modules**
+
+Run:
+
+```bash
+export JAVA_HOME=/opt/homebrew/opt/openjdk@25/libexec/openjdk.jdk/Contents/Home
+export PATH="$JAVA_HOME/bin:/opt/homebrew/bin:$PATH"
+mvn -pl :gb32960-ingest-app,:vehicle-history-app,:vehicle-analytics-app -am package -Dmaven.test.skip=true
+```
+
+Expected: BUILD SUCCESS.
+
+- [ ] **Step 2: Start Kafka test environment**
+
+Use the repository's existing Kafka setup if present. If there is no repository-local Kafka script, use a local Kafka compatible with `KAFKA_BROKERS=127.0.0.1:9092`.
+
+Create topics:
+
+```bash
+kafka-topics --bootstrap-server 127.0.0.1:9092 --create --if-not-exists --topic vehicle.raw.gb32960.v1 --partitions 12 --replication-factor 1
+kafka-topics --bootstrap-server 127.0.0.1:9092 --create --if-not-exists --topic vehicle.event.gb32960.v1 --partitions 12 --replication-factor 1
+kafka-topics --bootstrap-server 127.0.0.1:9092 --create --if-not-exists --topic vehicle.dlq.gb32960.v1 --partitions 3 --replication-factor 1
+```
+
+- [ ] **Step 3: Start split services on local ports**
+
+Run each command in a separate terminal:
+
+```bash
+KAFKA_BROKERS=127.0.0.1:9092 \
+GB32960_PORT=32960 \
+HTTP_PORT=20100 \
+java --sun-misc-unsafe-memory-access=allow \
+ -jar modules/apps/gb32960-ingest-app/target/gb32960-ingest-app.jar
+```
+
+```bash
+KAFKA_BROKERS=127.0.0.1:9092 \
+HTTP_PORT=20200 \
+EVENT_FILE_STORE_PATH=./target/split-event-store \
+SINK_ARCHIVE_PATH=./target/split-archive \
+java --sun-misc-unsafe-memory-access=allow \
+ -jar modules/apps/vehicle-history-app/target/vehicle-history-app.jar
+```
+
+```bash
+KAFKA_BROKERS=127.0.0.1:9092 \
+HTTP_PORT=20300 \
+VEHICLE_STAT_FILE_PATH=./target/split-vehicle-stat \
+java --sun-misc-unsafe-memory-access=allow \
+ -jar modules/apps/vehicle-analytics-app/target/vehicle-analytics-app.jar
+```
+
+- [ ] **Step 4: Verify health endpoints**
+
+Run:
+
+```bash
+curl -sS http://127.0.0.1:20100/actuator/health
+curl -sS http://127.0.0.1:20200/actuator/health
+curl -sS http://127.0.0.1:20300/actuator/health
+```
+
+Expected: each returns `{"status":"UP"}`.
+
+- [ ] **Step 5: Send a known-valid GB32960 frame**
+
+Use an existing GB32960 test fixture from `modules/protocols/protocol-gb32960/src/test/java`. If no command-line sender exists, add a small test-only utility or use `nc` with a hex-decoded fixture:
+
+```bash
+xxd -r -p /tmp/gb32960-valid-realtime.hex | nc 127.0.0.1 32960
+```
+
+Expected:
+
+- Kafka topic `vehicle.raw.gb32960.v1` receives one raw record.
+- Kafka topic `vehicle.event.gb32960.v1` receives one or more normalized records.
+- Client receives a GB32960 success ACK.
+- `target/split-archive` contains a `.bin` file.
+- `target/split-event-store` contains `events.duckdb` or Parquet files after flush.
+- `target/split-vehicle-stat` contains updated stat files after analytics consumes the event.
+
+- [ ] **Step 6: Write runbook**
+
+Create `docs/operations/gb32960-service-split-runbook.md` with:
+
+```markdown
+# GB32960 Split Service Runbook
+
+## Services
+
+- `gb32960-ingest-app`: TCP 32960, HTTP 20100
+- `vehicle-history-app`: HTTP 20200
+- `vehicle-analytics-app`: HTTP 20300
+
+## Kafka Topics
+
+- `vehicle.raw.gb32960.v1`
+- `vehicle.event.gb32960.v1`
+- `vehicle.dlq.gb32960.v1`
+
+## ACK Semantics
+
+GB32960 success ACK is sent after required Kafka production succeeds. History and analytics failures do not block ACK.
+
+## Local Start
+
+Use the commands from Task 8 Step 3.
+
+## Verification
+
+Use health endpoints, Kafka topic consumption, archive file count, event-store query APIs, analytics output files, and ACK latency metrics.
+
+## Rollback
+
+Stop split services and start `bootstrap-all` with the previous environment. Keep `bootstrap-all` available until split service counts and sample queries match production expectations.
+```
+
+- [ ] **Step 7: Commit**
+
+```bash
+git add docs/operations/gb32960-service-split-runbook.md
+git commit -m "docs: add gb32960 split service runbook"
+```
+
+---
+
+### Task 9: Final Verification and Cutover Checklist
+
+**Files:**
+- Modify: `docs/operations/gb32960-service-split-runbook.md`
+
+- [ ] **Step 1: Run targeted tests**
+
+Run:
+
+```bash
+mvn -pl :sink-mq,:ingest-core,:protocol-gb32960,:event-history-service,:vehicle-state-service,:vehicle-stat-service test
+```
+
+Expected: BUILD SUCCESS.
+
+- [ ] **Step 2: Run split app tests**
+
+Run:
+
+```bash
+mvn -pl :gb32960-ingest-app,:vehicle-history-app,:vehicle-analytics-app test
+```
+
+Expected: BUILD SUCCESS.
+
+- [ ] **Step 3: Run package verification**
+
+Run:
+
+```bash
+mvn -pl :gb32960-ingest-app,:vehicle-history-app,:vehicle-analytics-app -am package -Dmaven.test.skip=true
+```
+
+Expected: BUILD SUCCESS.
+
+- [ ] **Step 4: Update runbook with measured results**
+
+Append a section:
+
+```markdown
+## Latest Local Verification
+
+- Build:
+- Unit tests:
+- Split app startup:
+- Kafka raw records:
+- Kafka event records:
+- Archive files:
+- Event-store files:
+- Analytics output:
+- ACK behavior:
+```
+
+Fill the fields with the actual command outputs and counts from the run.
+
+- [ ] **Step 5: Commit final verification docs**
+
+```bash
+git add docs/operations/gb32960-service-split-runbook.md
+git commit -m "docs: record gb32960 split verification"
+```
+
+---
+
+## Self-Review
+
+Spec coverage:
+
+- Three services are covered by Tasks 1, 2, 3, and 7.
+- Kafka topic and routing contract is covered by Task 4.
+- ACK after Kafka success is covered by Task 5.
+- History consumption and storage is covered by Task 6 and Task 8.
+- Analytics consumption and processing is covered by Task 6 and Task 8.
+- `bootstrap-all` remains untouched except for pre-existing unrelated worktree changes.
+- Verification and runbook are covered by Tasks 8 and 9.
+
+Implementation risk notes:
+
+- Task 5 is the most invasive task because it changes protocol ACK ordering. Implement it behind the smallest possible new API and keep `Dispatcher.dispatch(RawFrame)` behavior for other protocols.
+- If Kafka transactions are needed to atomically produce raw and normalized records, add that as a follow-up after the split app boundary is working. The first implementation can use idempotent producer plus shared `eventId`/`sourceRawEventId`.
+- If composition tests need additional auto-config classes, copy the exact pattern from `ProtocolAutoConfigurationCompositionTest` rather than widening app dependencies.