_obd_pipeline_smoke.nx source
↩ module page · 189 lines · 8123 B
1// _obd_pipeline_smoke.nx -- end-to-end OBD pipeline correctness gate.
2//
3// Proves the OBD predictive-maintenance pipeline works by stitching
4// every layer together + asserting expected outputs:
5//
6// 1. nx_obd2_shim.rx_fn parses a synthetic ISO-TP Mode 01
7// coolant-temp response -> emits EVT_COOLANT_TEMP event
8// 2. nx_pm_consumer_on_event ingests N drift samples ->
9// eventually classifies as ANOM_DRIFT_UP
10// 3. nx_tm_emit_anomaly serializes verdict to stdout NDJSON
11//
12// Synthetic dataset: coolant temperature rising linearly from 87C
13// to 108C across 100 samples spaced 1 hour apart (~4-day window).
14// Per OBD spec the high threshold is ~105C; the linear drift should
15// cross within the 7-day extrapolation horizon, firing ANOM_DRIFT_UP.
16//
17// Status: SEED v0.1.0. 2026-05-27.
18// WINNER-TIER: EXEMPT (correctness gate per NISHIBENCH_WINNER_STANDARD
19// §antipatterns; smokes are gates not perf claims)
20// INCUMBENTS: n/a (correctness gate)
21// EXEMPTION REASON: correctness validation; no performance claim made
22//
23// Exit-code convention:
24// 0 all assertions PASS
25// 1-9 setup-failure verdicts (memory alloc, init)
26// 10-19 shim-path verdicts (frame parse, event emit)
27// 20-29 consumer-path verdicts (register, on_event, classification)
28// 30-39 telemetry-path verdicts (bus init, emit, sink)
29
30import "nx_syscalls.nx"
31import "nx_telemetry.nx"
32import "shims/nx_obd2_shim.nx"
33import "consumers/nx_pred_maint_consumer.nx"
34
35// ===== Test parameters =================================================
36
37const SMOKE_N_SAMPLES: i64 = 100 // drift samples to inject
38const SMOKE_SAMPLE_PERIOD_NS: i64 = 3600000000000 // 1 hour per sample (ns)
39const SMOKE_START_TEMP_C: i64 = 87
40const SMOKE_END_TEMP_C: i64 = 108
41const SMOKE_THRESHOLD_LO: i64 = 70 // OBD spec low (degC)
42const SMOKE_THRESHOLD_HI: i64 = 105 // OBD spec high
43
44// ===== Phase 1: build synthetic OBD frame =================================================
45//
46// Synthesizes an ISO-TP Mode 01 response for coolant temp.
47// Frame shape (8 bytes per ISO-TP convention):
48// byte 0: 0x03 (SF type=0, len=3)
49// byte 1: 0x41 (Mode 01 response = 0x01 + 0x40)
50// byte 2: 0x05 (PID for coolant temp)
51// byte 3: temp_C + 40 (OBD encoding per SAE J1979)
52// byte 4-7: 0x55 padding
53
54func smoke_build_coolant_frame(out: *u8, temp_c: i64) -> i64 {
55 out[0] = 0x03 as u8
56 out[1] = 0x41 as u8
57 out[2] = 0x05 as u8 // PID coolant temp
58 out[3] = ((temp_c + 40) & 0xff) as u8 // OBD-encoded
59 out[4] = 0x55 as u8
60 out[5] = 0x55 as u8
61 out[6] = 0x55 as u8
62 out[7] = 0x55 as u8
63 return 8
64}
65
66// ===== Phase 2: shim parse round-trip check =================================================
67
68func smoke_assert_shim_parse(shim: *NxObd2Shim) -> i64 {
69 let frame: *u8 = sys_mmap(16)
70 smoke_build_coolant_frame(frame, 87) // canonical 87C sample
71
72 let evt_kinds: *i64 = (sys_mmap(8 * 4)) as *i64
73 let evt_values: *i64 = (sys_mmap(8 * 4)) as *i64
74
75 let n_events: i64 = nx_obd2_rx_fn(shim, frame, 8, evt_kinds, evt_values, 4)
76 if n_events != 1 { return 10 } // expect exactly 1 event
77 if evt_kinds[0] != NX_OBD2_EVT_COOLANT_TEMP { return 11 } // wrong event kind
78 if evt_values[0] != 87 { return 12 } // wrong decoded temp
79
80 return 0
81}
82
83// ===== Phase 3: consumer drift detection =================================================
84//
85// Feeds N samples of linear drift into the consumer; asserts that
86// at some point during the run the classifier returns ANOM_DRIFT_UP.
87
88func smoke_assert_drift_detection(consumer: *NxPredMaintConsumer,
89 stats_buf: *NxPidStats) -> i64 {
90 var i: i64 = 0
91 var saw_drift: i64 = 0
92 while i < SMOKE_N_SAMPLES {
93 // Linear interpolation start_temp..end_temp across N samples.
94 let span: i64 = SMOKE_END_TEMP_C - SMOKE_START_TEMP_C
95 let temp: i64 = SMOKE_START_TEMP_C + ((span * i) / SMOKE_N_SAMPLES)
96 let ts: i64 = i * SMOKE_SAMPLE_PERIOD_NS
97
98 let verdict: i64 = nx_pm_consumer_on_event(consumer,
99 NX_OBD2_EVT_COOLANT_TEMP,
100 temp, ts, stats_buf)
101 if verdict == NX_PM_ANOM_DRIFT_UP { saw_drift = 1 }
102 if verdict == NX_PM_ANOM_HIGH_3SIGMA { saw_drift = 1 } // also acceptable
103 i = i + 1
104 }
105
106 if saw_drift == 0 { return 25 } // never flagged
107 return 0
108}
109
110// ===== Phase 4: telemetry wire check =================================================
111//
112// After wiring the bus, re-running drift should produce stdout
113// NDJSON. V1 doesn't capture stdout in-process (would need fd
114// redirect via sys_pipe), so this phase just verifies the bus
115// emit_count increments + drop_count doesn't explode.
116
117func smoke_assert_telemetry_emit(bus: *NxTelemetryBus,
118 consumer: *NxPredMaintConsumer,
119 stats_buf: *NxPidStats) -> i64 {
120 let initial_emit: i64 = bus.emit_count
121 let initial_drops: i64 = bus.drop_count
122
123 // Replay drift; consumer should emit when classifier flags.
124 var i: i64 = 0
125 while i < SMOKE_N_SAMPLES {
126 let span: i64 = SMOKE_END_TEMP_C - SMOKE_START_TEMP_C
127 let temp: i64 = SMOKE_START_TEMP_C + ((span * i) / SMOKE_N_SAMPLES)
128 let ts: i64 = (i + SMOKE_N_SAMPLES) * SMOKE_SAMPLE_PERIOD_NS // continue after phase 3
129 nx_pm_consumer_on_event(consumer, NX_OBD2_EVT_COOLANT_TEMP,
130 temp, ts, stats_buf)
131 i = i + 1
132 }
133
134 if bus.emit_count == initial_emit { return 35 } // bus never received an emit
135 // Drop_count growing is OK (rate limiting); we just assert
136 // SOMETHING got through.
137 return 0
138}
139
140// ===== main =================================================
141
142func main() -> i64 {
143 // ----- setup -----
144 let shim: *NxObd2Shim = (sys_mmap(64)) as *NxObd2Shim
145 if nx_obd2_shim_init(shim) != NX_SHIM_OK { return 1 }
146
147 let times_buf: *i64 = (sys_mmap(8 * NX_PM_WINDOW_SIZE)) as *i64
148 let values_buf: *i64 = (sys_mmap(8 * NX_PM_WINDOW_SIZE)) as *i64
149 let rings: *NxPidRing = (sys_mmap(64 * NX_PM_MAX_PIDS)) as *NxPidRing
150
151 let consumer: *NxPredMaintConsumer = (sys_mmap(64)) as *NxPredMaintConsumer
152 if nx_pm_consumer_init(consumer, rings) != NX_PM_OK { return 2 }
153 if nx_pm_consumer_register(consumer, NX_OBD2_EVT_COOLANT_TEMP,
154 times_buf, values_buf,
155 SMOKE_THRESHOLD_LO, SMOKE_THRESHOLD_HI) != NX_PM_OK { return 3 }
156
157 let stats_buf: *NxPidStats = (sys_mmap(64)) as *NxPidStats
158
159 // ----- Phase 2: shim round-trip -----
160 let rc_shim: i64 = smoke_assert_shim_parse(shim)
161 if rc_shim != 0 { return rc_shim }
162
163 // ----- Phase 3: drift detection (without telemetry) -----
164 let rc_drift: i64 = smoke_assert_drift_detection(consumer, stats_buf)
165 if rc_drift != 0 { return rc_drift }
166
167 // ----- Phase 4: wire telemetry + assert emits arrive -----
168 let rl_keys: *i64 = (sys_mmap(8 * NX_TM_RL_SLOTS)) as *i64
169 let rl_tokens: *i64 = (sys_mmap(8 * NX_TM_RL_SLOTS)) as *i64
170 let rl_refill: *i64 = (sys_mmap(8 * NX_TM_RL_SLOTS)) as *i64
171 let rl: *NxTelemetryRateLimiter = (sys_mmap(64)) as *NxTelemetryRateLimiter
172 if nx_tm_rl_init(rl, rl_keys, rl_tokens, rl_refill, 1000) != NX_TM_OK { return 30 }
173
174 let sink: *NxTelemetrySink = (sys_mmap(32)) as *NxTelemetrySink
175 sink.emit_fn = 0 // V1 hardcoded to stdout-JSON; ignored
176 sink.rate_cap = 1000
177 sink.valid = 1
178
179 let bus: *NxTelemetryBus = (sys_mmap(64)) as *NxTelemetryBus
180 if nx_tm_bus_init(bus, sink, rl) != NX_TM_OK { return 31 }
181
182 if nx_pm_consumer_set_telemetry_bus(consumer, bus) != NX_PM_OK { return 32 }
183
184 let rc_telem: i64 = smoke_assert_telemetry_emit(bus, consumer, stats_buf)
185 if rc_telem != 0 { return rc_telem }
186
187 // All four phases green.
188 return 0
189}