research(rectification): nadi second-level study fails the preregistered gates (BUG-1240)

N1, N2, N3, and N5 do not pass. Switching ayanamsa moves level-5 lords and
equal-division D150 past the 20% wording line. Two full runs are byte-identical.
No production scoring change. closed_by_design.

docs/BUG_HISTORY.md

@@ -16627,3 +16627,19 @@
 - 相关记录：BUG-995（前端测试要同时看 cancelled 和退出码）。
 - 复发自：无
 - 修复版本：待发布
+
+## BUG-1240 | 秒级 / 纳迪校准从未经过检验
+
+- 状态：closed_by_design（2026-10-05 离线研究：预先登记的 N1、N2、N3、N5 都不过门；换岁差时第 5 层和等分 D150 的变化超过 20%，对外不说秒级）
+- 首次发现：2026-10-05
+- 最近更新：2026-10-05
+- 影响面：生时校正对外精度口径。主链 Vimshottari 只评前三层。第 4 层、第 5 层和 D150 不进入计分。
+- 用户现象：校正给出时间范围，不能把结果说成某一秒。公开评价集里 52/77 例的记录分钟是 5 的倍数，记录本身没有秒。
+- 触发条件：无用户路径。离线研究，任务书 `docs/tasks/TASK-rectification-nadi-seconds-research-20261005.md`。
+- 根因：在 77 例公开 AA 上，预先登记的规则没有胜过安慰剂。第 5 层在记录分钟上的拟合率 0.221，日期平移安慰剂 0.216，跨例换日期 0.225。六项主检验只有「第 4 层对日期平移」一项达到 p < 0.05，过门要求六项都达到。留出一件的提升区间含 0。六题后的区间里用第 5 层重排，±10 分钟从 25/77 降到 5/77。换一种岁差，第 5 层约 86%–92% 的日事件换主星，等分 D150 为 73/73 例换段。±15 秒已经让约六成日事件换掉第 5 层。
+- 修复：不改生产计分，不立实现单。研究记录见 `docs/research/rectification_nadi_seconds_2026_10_05.md`。
+- 验证：`tests/test_nadi_seconds_research.py` 10 项通过（前三层对账 0 差，77 例 964 件事件）。结果 JSON 由 `scripts/research/nadi_seconds_run.py` 写出。预登记提交 `57480db324f1602cfc422ff8c774d0d6e2113af6` 在第一次 v5 计分之前。两次复跑的字节核对写在进度记录里。
+- 防复发：再提秒级或纳迪校准时，先复跑这份预登记；不得用看过结果之后新加的规则判过门；不得重调 BUG-1091 已关闭的权重。
+- 相关记录：BUG-560、BUG-1090、BUG-1091、BUG-1105、BUG-1141。
+- 复发自：无
+- 修复版本：研究分支 `codex/rectification-nadi-seconds-research-20261005`（未推送、未部署）

docs/research/ACTIVE_FRONTS.md

@@ -131,3 +131,4 @@ Do not open new product surfaces before at least one of these four lanes is clos
 - 2026-09-29 起评价集换 **v5**（77 例公开 AA，`references/real_case_calibration/minute_rectification_holdout_v5.json`）：协议 `holdout_v5_protocol_2026_09_29.md`、构建与基线成绩单 `holdout_v5_build_2026_09_29.md`。后续打分判定以 v5 为准，v4 数字只作历史对照。
 - 2026-09-30 盘型口径研究（BUG-1105）：`<repo>/docs/research/rectification_varga_resolution_2026_09_30.md`——目标从分钟改为分盘上升段。v5 六题后 ±10 D9 / D10 头段 = 真值 88% / 86%，占比 ≥ 0.6 留一法 92% / 90%；±30 约 1/4–1/3 用户可给可信分盘；±60 分盘 blocked。按段选题 ≤ ±30 有 +3～+7 例；提前停不过红线；「不用校正」只对只需 D1 的问题成立。**建议立实现单**（§7）。
 - 2026-10-01 角度类时间技法研究（BUG-1141）：`<repo>/docs/research/rectification_angle_timing_2026_10_01.md`——行运压上升 / 天顶、次限推运角、太阳弧（1° ≈ 4 分钟），77 例 v5 真实日期对乱序日期 9 项预登记检验 0 项显著（最高胜过 58%，门槛 99.44%）；年龄结构保持对照同样无信号；`closed_by_design`。
+- 2026-10-05 纳迪秒级证伪（BUG-1240）：`<repo>/docs/research/rectification_nadi_seconds_2026_10_05.md`——第 4 / 第 5 层小运和等分 D150 对安慰剂日期，预先登记的 N1、N2、N3、N5 都不过门；换岁差时第 5 层和 D150 的变化超过 20%。`closed_by_design`。不立实现单，对外不说秒级。

docs/research/nadi_seconds_preregistration_2026_10_05.json

@@ -0,0 +1,116 @@
+{
+  "id": "nadi_seconds_preregistration_2026_10_05",
+  "dataset": "references/real_case_calibration/minute_rectification_holdout_v5.json",
+  "event_cutoff": "2026-09-14",
+  "ayanamsa": "raman",
+  "node_mode": "mean",
+  "locked_before_v5_scoring": true,
+  "grids": {
+    "coarse_radius_seconds": 600,
+    "coarse_step_seconds": 5,
+    "fine_radius_seconds": 120,
+    "fine_step_seconds": 1
+  },
+  "rules": {
+    "computed": ["A4", "A5", "B1", "B2", "B3", "B4", "B5", "N5A", "N5B"],
+    "n1_primary": ["A4", "A5", "B2"],
+    "n2_primary": "A5",
+    "n3_primary": "A5",
+    "n5_primary": ["N5A", "N5B"],
+    "definitions": {
+      "levels": "五层 Vimshottari 用生产递归：dasha_analyzer.build_dasha_timeline 之后再四次 build_antardasha。第 1 层是大运，第 4 层是 Sookshma，第 5 层是 Prana。前三层必须与 _active_vimshottari 对账。calculate_five_level_dasha 不是计分链。",
+      "targets": "领域目标宫主只读 DOMAIN_CONFIG 与 _house_lords，不改映射，也不乘辅助分。命中是二值。",
+      "A4": "第 4 层主星属于该事件领域的目标宫主。",
+      "A5": "第 5 层主星属于该事件领域的目标宫主。",
+      "B1": "五层里至少 1 层主星属于目标宫主。探索项，不是过门规则。",
+      "B2": "五层里至少 2 层主星属于目标宫主。",
+      "B3": "五层里至少 3 层。探索项。",
+      "B4": "五层里至少 4 层。探索项。",
+      "B5": "五层全部属于目标宫主。探索项。",
+      "N5A": "等分 D150（calc_custom_varga，N=150）上升星座的宫主星属于目标宫主。",
+      "N5B": "同一颗 D150 宫主星出现在当天五层小运主星里。",
+      "classical_index": "动宫顺排、固宫逆排、变宫从中段起排，状态 variant_unverified。不进入任何过门，也不进入 N4 的文案触发。不抄录纳迪描述原文。",
+      "named_text": "blocked。仓库内没有可合法使用的 Chandra Kala Nadi 描述文本。"
+    }
+  },
+  "samples": {
+    "n1_min_day_events": 3,
+    "n2_min_day_events": 4,
+    "n1_expected_n": 43,
+    "note": "N1 与 N5 只用日精度事件不少于 3 件、且两种安慰剂都非空的例子。开工时数过，43 例两种安慰剂都齐全。比较行的 n 不是 43，就不是本登记的样本，不能判过门。取整组与非取整组分开列，判定用合并样本。"
+  },
+  "tests": {
+    "permutations": 9999,
+    "bootstrap_resamples": 5000,
+    "n1_permutation_seed_base": 20261007,
+    "n2_bootstrap_seed": 20261008,
+    "n2_random_second_seed": 20261009,
+    "n3_bootstrap_seed": 20261010,
+    "n5_permutation_seed_base": 20261011,
+    "seed_order": "N1 与 N5 都按主规则列表的顺序，每条规则先 P1 再 P2，种子是基数加 0、1、2…。因此 N1 为 A4-P1=20261007、A4-P2=20261008、A5-P1=20261009、A5-P2=20261010、B2-P1=20261011、B2-P2=20261012。N5 为 N5A-P1=20261011、N5A-P2=20261012、N5B-P1=20261013、N5B-P2=20261014。N2 随机对照的种子是字符串 20261009:{case_id}:{event_id}。N2 安慰剂日期的 bootstrap 种子是 20261008+1。N3b 的 bootstrap 种子是 20261010 加半径分钟数，所以 ±10 为 20261020，±30 为 20261040，±60 为 20261070。"
+  },
+  "placebo": {
+    "shift_low_days": 30,
+    "shift_high_days": 180,
+    "shift_seed_template": "20261005:{case_id}",
+    "swap_seed": 20261006,
+    "p1": "每件日精度事件的日期用 Random(20261005:{case_id}) 平移。幅度 randint(30, 180)，符号 ±1。落在 (出生日, 2026-09-14] 之外就反向；仍在外面就按 365 天步长夹回，最后才用出生日之后 400 天并截到截止日期。领域留在原事件上。",
+    "p2": "把全部日精度事件按 (case_id, event_id) 排序后，用 Random(20261006) 打乱日期。领域留在接收方。日期夹回接收方的 (出生日, 截止日期]。置换以整例为单位。"
+  },
+  "statistic": {
+    "truth_minute_offsets": "0 到 59 秒，含两端。记录没有秒。真值分钟的拟合率是这 60 秒的平均。",
+    "fit_rate": "某一秒解释的日精度事件数，除以该例日精度事件数。",
+    "difference": "同一例上，真值分钟平均拟合率减去安慰剂日期在同样 60 秒上的平均拟合率。",
+    "test": "单侧配对符号翻转。极端值是置换均值大于或等于观察均值的次数。p = (极端值 + 1) / (置换次数 + 1)。过门看精确分数小于 0.05，并且均值大于 0。不看四舍五入到 6 位后的显示值。",
+    "window_mean": "粗网格上的平均拟合率，等于在粗网格里均匀抽一秒的期望。写入报告。它不是 N1 的过门对照。",
+    "fraction_all": "粗网格上能解释全部日精度事件的秒所占比例。A4、A5、B1、B2 都报。接近 100% 表示问前事对得上并不能分辨真假。这是诊断，不是过门。"
+  },
+  "n3_protocol": {
+    "radius_gate": 10,
+    "nadi_limit_seconds": 60,
+    "hit": "并列第一的偏移必须每一个都在限制内。空的并列是未命中。没有日精度事件时，全部秒并列，记为未命中。",
+    "n3a": "只按 A5 在粗网格上排序。报告 ≤60 秒与 ≤120 秒，并对照均匀随机（步长 5 秒时 |偏移|≤60 的格点数除以 241）和已发表的引擎先验头名 0.18。N3a 不参与过门。本运行不另做 N3a 的 ±30 / ±60 网格。",
+    "n3b": "复现六题回放的前半段：六次探测、不注入引导窗口，再取交付区间。区间内按 1 秒重排 A5。命中仍要求全部并列都在记录时间 60 秒内。对照是生产头名全部落在 1 分钟内。提升是纳迪命中减生产命中，整例 bootstrap，下界大于 0 才算提升。另报 any_exact，用来对照已发表的 0.64；那一列不是过门。真值在区间内须不少于 76/77。过门只看半径 10，且例数必须是 77。±30 与 ±60 只报告。",
+    "published_references": {
+      "engine_prior_top1_pm10": 0.18,
+      "six_question_any_exact_pm10": 0.64,
+      "cluster_head_pm10": 0.68,
+      "truth_in_range": "76/77",
+      "source": "docs/research/holdout_v5_build_2026_09_29.md"
+    }
+  },
+  "n4": {
+    "ayanamsas": ["raman", "lahiri", "kp", "true_citra"],
+    "time_offsets_seconds": [-60, -30, -15, 15, 30, 60],
+    "longitude_shifts_km": [10, -10],
+    "node_mode_compare": "mean_to_true",
+    "sample": "至少有一件日精度事件的例子。第 4 / 第 5 层按事件加权。D150 用等分段号，按例子计。没有日精度事件的 4 例不进入这张表。",
+    "wording_trigger": "Raman 分别换成 Lahiri、KP、True Chitra。三次里只要有一次第 5 层事件变化率大于 0.20，或等分 D150 的例子变化率大于 0.20，对外文案就不得出现秒级。时间和地点扰动写入表，本身不触发这条例。交点 mean 与 true 单列，预期只动罗睺计都。古典段号不触发。"
+  },
+  "n2": {
+    "rule": "A5",
+    "grid": "coarse",
+    "design": "日精度事件不少于 4 的例子，留一件。其余事件在粗网格上的最高分，并列全部保留。被留出的事件在这些秒上被解释的比例，减去同数量随机秒上的比例。随机种子是 20261009:{case_id}:{event_id}。并列不少于整张网格时，随机对照就是整张网格。例子内先平均，再对例子做 5000 次 bootstrap。过门只看相对随机秒的区间下界大于 0。同一批领先秒上、把留出位置换成 P1 日期的差，只报告，不是额外门槛。"
+  },
+  "gates": {
+    "n1": "A4、A5、B2 对 P1 和 P2 共 6 项全部通过，且每项 n 为 43，且不是 --limit 子集。",
+    "n2": "相对随机秒的 bootstrap 下界大于 0，且不是子集。",
+    "n3": "半径 10 上，纳迪相对生产 ≤1 分钟命中的提升，bootstrap 下界大于 0，并且真值在区间内不少于 76/77，例数 77。",
+    "n4": "不过方法门。只决定对外是否还能说秒级。",
+    "n5": "N5A、N5B 对 P1 和 P2 共 4 项全部通过，且不是子集。低优先。原文比对保持 blocked。"
+  },
+  "not_a_gate": [
+    "N3a",
+    "N3 的 ±30 与 ±60",
+    "取整组与非取整组的拆分",
+    "古典 D150 段号",
+    "calculate_five_level_dasha 与生产链的差异",
+    "B1、B3、B4、B5",
+    "窗口平均拟合率",
+    "能解释全部事件的秒所占比例",
+    "N2 的安慰剂日期对照",
+    "N3b 的 any_exact 复现列",
+    "N4 的时间扰动与地点扰动"
+  ],
+  "after_the_fact": "看过 v5 结果之后改过的规则、阈值或种子，只能写进研究文档的事后列，不能用来判过门。"
+}

docs/research/rectification_nadi_seconds_2026_10_05.md

@@ -0,0 +1,123 @@
+# 生时校正「纳迪秒级」证伪（2026-10-05，BUG-1240）
+
+> 任务书 `docs/tasks/TASK-rectification-nadi-seconds-research-20261005.md`。离线研究，生产代码没有改。
+> 数据：`references/real_case_calibration/minute_rectification_holdout_v5.json`（77 例公开 Rodden AA，244 件日精度事件）。默认岁差 Raman、交点 mean。
+> 规则在跑数之前单独提交：`57480db324f1602cfc422ff8c774d0d6e2113af6`。文件 `docs/research/nadi_seconds_preregistration_2026_10_05.json`。
+> 结果：`docs/research/nadi_seconds_results_2026_10_05.json`。Swiss Ephemeris 2.10.03。`PYTHONHASHSEED=0`。看过结果之后没有改规则、阈值或种子。
+
+## 1. 结论
+
+| 项 | 过门 | 关键数字 | 对照 |
+| --- | --- | --- | --- |
+| N1 拟合率 | 否 | 6 项里 1 项达到门槛：第 4 层对「日期平移」高 0.090，p = 0.0116。另外 5 项没有 | 日期平移，以及把日期换到别的例子上 |
+| N2 留出一件 | 否 | 33 例。留出事件比随机秒高 0.041，95% 区间 −0.021 到 0.101 | 随机秒。安慰剂日期的区间也跨过 0 |
+| N3 找回记录分钟 | 否 | ±10：纳迪 5/77，同一次回放的生产头名 25/77，提升 −0.260，区间整个在 0 下面 | 真值仍在交付区间内 76/77 |
+| N4 噪声地板 | 触发文案禁令 | 只换岁差，第 5 层有 86%–92% 的日事件换主星，等分 D150 有 73/73 例换段 | 时间和地点写入下表，不单独当方法门 |
+| N5 D150 结构 | 否 | 4 项都不达到门槛。按描述文字比对经历：blocked，仓库里没有可合法使用的原文 | 同 N1 的两种安慰剂 |
+
+不立实现单。第 4 层、第 5 层和 D150 不接进校正计分。BUG-1240 记为 `closed_by_design`。
+
+对外不能说秒级。N1–N3 没有过门，而且 N4 已经越过预先登记的 20% 线：换一种岁差算法，秒级记号就大面积变掉。
+
+## 2. N0 底座
+
+五层小运走生产递归（大运时间轴后再切四次），前三层与主链逐事件对账。77 例、964 件事件，不一致 0。等分 D150 用 `calc_custom_varga`（N=150）。古典「动宫顺排、固宫逆排、变宫从中段起排」标成 `variant_unverified`，不参与过门。
+
+记录分钟：52 例是 5 的倍数，25 例不是。日精度事件 244 件；不少于 3 件的 43 例，不少于 4 件的 33 例，不少于 5 件的 19 例，0 件的 4 例。记录里没有秒。真值分钟的拟合率是该分钟 0–59 秒的平均。非 5 倍数那一组仍然只知道分钟，不知道秒。
+
+## 3. N1
+
+样本是预先登记的 43 例，两种安慰剂都在。拟合率是「这一秒解释了多少件日精度事件」除以事件数。差是真值分钟的平均减去安慰剂日期在同样 60 秒上的平均。整例符号翻转 9999 次。过门要均值大于 0，并且精确 p < 0.05。六项都要达到。
+
+| 规则 | 对照 | 真值分钟 | 安慰剂 | 差 | 窗口平均 | p | 过门 |
+| --- | --- | --- | ---: | ---: | ---: | ---: | --- |
+| 第 4 层属于目标宫主 | 日期平移 | 0.259 | 0.169 | +0.090 | 0.214 | 0.0116 | 达到 |
+| 第 4 层属于目标宫主 | 跨例换日期 | 0.259 | 0.221 | +0.038 | 0.214 | 0.1476 | 否 |
+| 第 5 层属于目标宫主 | 日期平移 | 0.221 | 0.216 | +0.005 | 0.229 | 0.3794 | 否 |
+| 第 5 层属于目标宫主 | 跨例换日期 | 0.221 | 0.225 | −0.003 | 0.229 | 0.5526 | 否 |
+| 五层里至少两层命中 | 日期平移 | 0.329 | 0.339 | −0.009 | 0.325 | 0.6338 | 否 |
+| 五层里至少两层命中 | 跨例换日期 | 0.329 | 0.318 | +0.012 | 0.325 | 0.3485 | 否 |
+
+第 5 层在真值分钟上的拟合率（0.221）低于 ±10 分钟粗网格的平均（0.229）。记录中的那一分钟，并不比窗口里随便一秒更常被第 5 层解释。
+
+粗网格上能解释全部日精度事件的秒所占比例：第 4 层 0.0017，第 5 层 0.0047，至少一层 0.145，至少两层 0.011。严格规则并不是「几乎每一秒都能解释全部前事」。失败的原因是真值分钟并不比安慰剂更高。
+
+43 例里，分钟为 5 的倍数的有 31 例，其余 12 例。这是描述，没有单独做检验，也不能拿来过门。第 5 层真值分钟拟合率：5 的倍数 0.240，其余 0.171。其余那 12 例的真值分钟（0.171）低于它自己的窗口平均（0.199）。
+
+探索项 B3 / B4 / B5 不参与过门，数字在结果 JSON 的每例记录里。
+
+## 4. N2
+
+33 例、每例至少 4 件日精度事件。留一件，用其余事件在粗网格上取最高分，并列全留。再看被留出的那一件在这些秒上被第 5 层解释的比例，减去同样多的随机秒。
+
+相对随机秒：平均高 0.041，95% 区间 −0.021 到 0.101。相对安慰剂日期：平均高 0.026，区间 −0.037 到 0.087。过门只看随机秒，区间下界要大于 0。没有达到。
+
+5 的倍数 24 例，差的平均 0.047；其余 9 例，差的平均 0.024。两组都没有单独过门。
+
+## 5. N3
+
+命中的定义预先写死：并列第一的每一个偏移都要落在限制里。空并列算未命中。没有日精度事件的 4 例算未命中。
+
+### 单独用第 5 层排序（不参与过门）
+
+±10 分钟、每 5 秒一格。全部并列都落在记录时间 60 秒内：0/77。落在 120 秒内：2/77（0.026），其中 5 的倍数那组 0.038，另一组 0。同一网格上均匀抽一格，60 秒内的期望是 0.104，120 秒内是 0.203。已发表的引擎先验头名是 0.18。并列数的中位是 8；有 23 例超过 25 个并列，这已经塞不进 60 秒那一段。
+
+### 六题之后的交付区间里再排序（过门）
+
+沿用六题回放的前半段：六道选择题，不注入后面的引导窗口。区间内按 1 秒用第 5 层重排。回放 77 例没有报错。
+
+| 半径 | 真值在区间内 | 生产：全部并列都在 1 分钟内 | 纳迪：全部并列都在 60 秒内 | 提升 | 95% 区间 | 宽度中位 |
+| --- | --- | ---: | ---: | ---: | --- | ---: |
+| ±10（过门） | 76/77 | 25/77（0.325） | 5/77（0.065） | −0.260 | −0.377 到 −0.143 | 13 分钟 |
+| ±30（只报告） | 76/77 | 24/77（0.312） | 0/77 | −0.312 | −0.416 到 −0.208 | 33 分钟 |
+| ±60（只报告） | 76/77 | 13/77（0.169） | 0/77 | −0.169 | −0.260 到 −0.091 | 69 分钟 |
+
+±10 上，5 的倍数那组：生产 0.288，纳迪 0.058。另一组：生产 0.400，纳迪 0.080。两组都是纳迪更低。
+
+已发表的 ±10 成绩单是：引擎先验头名 0.18，六题回放头名 0.64，簇头名 0.68，真值在区间内 76/77，宽度中位 15 分钟（`docs/research/holdout_v5_build_2026_09_29.md`）。本轮真值覆盖同为 76/77。头名没有复现成 0.64。本轮的生产对照要求每一个并列分钟都在 1 分钟内；线上网格每 2 分钟一格，这等于「唯一头名就是记录的那一分钟」，得到 25/77。已发表的 0.64 是簇代表分钟的头名。宽度中位是 13 分钟，已发表是 15。按预登记，不把 25/77 改写成 0.64。过门比较的是同一次回放里的纳迪和生产，纳迪更低。
+
+## 6. N4 噪声地板
+
+73 例至少有一件日精度事件。第 4、第 5 层按事件加权。等分 D150 按例子计。没有日事件的 4 例不在这张表里。
+
+| 扰动 | 第 4 层事件变化 | 第 5 层事件变化 | 等分 D150 换段的例子 |
+| --- | ---: | ---: | ---: |
+| Raman → Lahiri | 0.873 | 0.857 | 73/73 |
+| Raman → KP | 0.848 | 0.922 | 73/73 |
+| Raman → True Chitra | 0.926 | 0.885 | 73/73 |
+| 出生时间 −15 秒 / +15 秒 | 0.107 / 0.115 | 0.607 / 0.598 | 12/73 / 29/73 |
+| 出生时间 −30 秒 / +30 秒 | 0.213 / 0.213 | 0.820 / 0.828 | 43/73 / 48/73 |
+| 出生时间 −60 秒 / +60 秒 | 0.418 / 0.393 | 0.918 / 0.902 | 70/73 / 72/73 |
+| 出生地向东 / 向西 10 公里 | 0 / 0 | 0 / 0 | 46/73 / 43/73 |
+
+交点从 mean 改成 true：73 例的月亮和上升都没变，罗睺都变了，第 5 层事件变化率是 0。小运只看月亮，这与预期一致。
+
+文案禁令看三次换岁差。三次的第 5 层事件变化率都大于 0.20，等分 D150 的例子变化率都是 1。禁令成立。±15 秒已经让大约六成日事件换掉第 5 层主星。东西方向移 10 公里不动第 5 层，但大约六成例子的等分 D150 会换段。
+
+## 7. N5
+
+等分 D150 的上升星座宫主。样本仍是 43 例，对照仍是两种安慰剂。
+
+| 规则 | 对照 | 真值分钟 | 安慰剂 | 差 | p | 过门 |
+| --- | --- | ---: | ---: | ---: | ---: | --- |
+| 宫主属于目标宫主 | 日期平移 | 0.272 | 0.272 | 0 | 1 | 否 |
+| 宫主属于目标宫主 | 跨例换日期 | 0.272 | 0.272 | 0 | 1 | 否 |
+| 宫主出现在五层小运里 | 日期平移 | 0.476 | 0.451 | +0.025 | 0.2224 | 否 |
+| 宫主出现在五层小运里 | 跨例换日期 | 0.476 | 0.490 | −0.014 | 0.6411 | 否 |
+
+第一条规则只用出生那一秒的盘，不用事件日期，所以两种改日期的安慰剂和真值相同。这是这条规则的结果，不是跑数之后改出来的。按描述文字比对经历保持 blocked。
+
+## 8. 对外口径草稿
+
+校正给你一个时间范围。你答完现在这几道题，范围里往往还剩大约一刻钟；你记得的那一分钟大多还在这个范围里，这些分钟分不出先后。我们拿时间记录到分钟的公开例子核对过：用更细的小运去挑秒，并不比把事情的日期挪开更准，也没有把范围收到你记录中的那一分钟。出生时间差半分钟，或者换一种岁差算法，这种细记号就会变。所以校正只说到现在的方法能分开的范围，不把结果说成秒级。
+
+草稿对照 `frontend/docs/VOICE.md`：直接对「你」说，不点名其他产品，不写秒级准确率。没有改 VOICE 文件本身。
+
+## 9. 复现
+
+```bash
+export PYTHONHASHSEED=0
+python scripts/research/nadi_seconds_run.py
+```
+
+预登记、库、跑数脚本、数据集的 sha256 写在结果 JSON 开头和结尾。本轮分别是 `dc012dd2b287b28ac2eefb75d5b9050e7788cb09d0064639878a37753ee8c2d3`、`ba170f3a0476fc41b3be8ac9f035d2f82575006753eb14992ccf124132be0f94`、`3c1af919408c637eead6e43252f6696beec73cb66480d8138a9075d3a8fb8ed6`、`ab365a3e3f4953d4b538f2c0889dc98ab89444a28340c3a7b29e2a864cdd687a`。同一命令再跑一次，两份结果的 SHA-256 都是 `4f1569c3e8794e0d2c35d244d32b3896e47cf8a1f7d03d16ce01d3f07b51ec69`，逐字节一致。

docs/tasks/PROGRESS-rectification-nadi-seconds-research-20261005.md

@@ -0,0 +1,47 @@
+# PROGRESS · 生时校正「纳迪秒级」证伪（2026-10-05，BUG-1240）
+
+任务书：`TASK-rectification-nadi-seconds-research-20261005.md`。分支 `codex/rectification-nadi-seconds-research-20261005`，基线 `origin/staging` @ `4c64733f`。只 commit，不 push。未部署。
+
+## 一句话
+
+第 4 层、第 5 层和等分 D150 没有胜过预先登记的安慰剂，也没有在六题后的区间里把记录分钟找得更准。换岁差时细记号大面积变化。BUG-1240 `closed_by_design`。不立实现单。
+
+## 提交
+
+| 提交 | 内容 |
+| --- | --- |
+| `217cd4a7` | N0 库、跑数脚本、对账测试、BUG-1240 先记为 investigating、状态板改为执行中 |
+| `6d51c2de` | 库自己把 `scripts/` 放进导入路径。直接跑脚本时否则找不到星历模块。没有改统计量 |
+| `57480db3` | 预登记 JSON，单独提交，在第一次 v5 计分之前 |
+| 本提交 | 结果 JSON、研究报告、ACTIVE_FRONTS、BUG 终态、状态板改为待验收、本进度 |
+
+预登记全文 SHA-256：`dc012dd2b287b28ac2eefb75d5b9050e7788cb09d0064639878a37753ee8c2d3`。
+
+## 硬红线
+
+| 要求 | 结果 |
+| --- | --- |
+| 规则先登记再跑 | 预登记 `57480db3` 早于结果。看过数字之后没有改规则、阈值或种子。事后列为空 |
+| 排序程序不读 `true_minute` | 测试扫描拟合函数源码，并用会拒绝这个键的字典做了运行检查 |
+| 每个「对上了」都有安慰剂 | N1 六项、N2、N5 四项都写了安慰剂。N3 的对照是同一次六题回放的生产头名 |
+| 负结果照写 | 六项里只有第 4 层对日期平移达到 p < 0.05。没有改成只保留这一项 |
+| 取整组和非取整组分开 | 写在研究报告 §3–§5。过门用合并样本 |
+| 生产代码零改动 | 改动在 `scripts/research/`、`tests/test_nadi_seconds_research.py` 和 `docs/` |
+| 不重调 BUG-1091 的权重 | 没有改 `DOMAIN_CONFIG`、计分权重或宫位映射 |
+| 不抄纳迪原文 | N5 的文字比对写 blocked |
+| 两次复跑逐字节一致 | 见下方「复跑」 |
+| 隐私 | 文档和 BUG 不写出生时间。案例编号只留在结果 JSON，与既有公开评价集相同 |
+
+## 复跑
+
+两次全量都用 `PYTHONHASHSEED=0`，半径 ±10 / ±30 / ±60。第一次写出 `docs/research/nadi_seconds_results_2026_10_05.json`（356,824 字节）。第二次写到临时文件。SHA-256 都是 `4f1569c3e8794e0d2c35d244d32b3896e47cf8a1f7d03d16ce01d3f07b51ec69`，逐字节一致。
+
+## 测试
+
+`tests/test_nadi_seconds_research.py` 与开工时那五个文件一起再跑，退出码 0。新测试 10 项通过，前三层对账 0 差，覆盖 77 例 964 件事件。新测试没有加进快速门的固定名单。
+
+Python 快速门用 `--profile quick --skip-frontend-runtime`。这个工作树没有 `frontend/node_modules`，本单也不改前端，所以没有跑 npm，也没有安装依赖。编译、JSON、能力清单和片段审计都通过。pytest 是 1056 passed、1 skipped、1 failed，约 604 秒。失败的是 `tests/test_consultation_native_layers.py::test_the_new_layer_leaves_every_existing_output_unchanged`。同一条单独再跑通过。它不在本单改过的文件里，快速门的固定名单也不包含新的纳迪测试。不把它算成本研究的回归。
+
+## 让步
+
+N0、N1、N2、N3、N4、N5 都做了。N3 的 ±30 / ±60 只报告。N6 没有另做图。

docs/tasks/README.md

@@ -422,4 +422,4 @@
 | `TASK-astrologer-rulings-batch6-20261004.md` | `PROGRESS-astrologer-rulings-batch6-20261004.md` | 第七轮裁定（共享仓书面回复，产品采用；问 1 选 A）：Rath 版双主星按 p.43 (a)–(e)（BUG-1227 解除 blocked，推翻第五批红线 2 的 Table 17 年数底线）；第 5 级宫主度数只倒算计都（p.71 脚注 42）；罗计旺陷 ±1 年；同宫两主比经度；BUG 从 1228 起 | 待验收 | 分支 `codex/astrologer-rulings-batch6-20261004`（BUG-1227～1229；校正分数文件未改、未升版本） |
 | `TASK-gate-log-volume-20261004.md` | `PROGRESS-gate-log-volume-20261004.md` | 门禁 validate 单步日志 4.4 万行 / 2 MB 网页打不开（run 3170）：快速门只打摘要（失败给末 200 行 + 日志文件）、前端测试门禁上 dot + 失败汇总（本机仍 TAP）、拆分 validate 为 7 个 step（产品授权改 workflow，只限拆分与重定向）；检查一项不少 | **已实现待验收**（BUG-1230；未推送、未部署；Gitea 各 step 页面是否打得开留待推送后由产品确认） | 分支 `codex/gate-log-volume-20261004` |
 | `TASK-consult-latency-quickwins-20261005.md` | `PROGRESS-consult-latency-quickwins-20261005.md` | 普通对话耗时两项快修：咨询链校正闸不再同步白等 VedAstro 官方快照子进程（每域约 4 s，复用顶层缓存 + 负缓存，不推翻 BUG-301）；补分段计时（第 0/1 步耗时、推理 token、分类耗时进日志与 usage）。推理强度/精简说明待模型 key 另单（BUG 从 1231 起） | 待领取 | 分支 `codex/consult-latency-quickwins-20261005` |
-| `TASK-rectification-nadi-seconds-research-20261005.md` | `PROGRESS-rectification-nadi-seconds-research-20261005.md` | **「纳迪秒级校准」可证伪检验（离线）**：竞品宣传「问前事到天 → 秒级」。本仓主链只用三层小运，Sookshma / Prana 与 D150 从未进评价集；v5 真值 52/77 是整 5 分钟（秒级无真值可对）。N0 五层小运 + D150 底座（前三层与主链对账 0 差）、N1 拟合率 vs 安慰剂日期（核心）、N2 留一件预测、N3 六题后区间内再细分能否提头名、N4 岁差 / 坐标 / 时间扰动的噪声地板、N5 D150 结构层（原文比对 blocked）、N6 结论 + 对外口径草稿。规则先登记再跑；不改生产代码；不得重调 BUG-1091 已关的权重 | 待领取 | 分支 `codex/rectification-nadi-seconds-research-20261005`（BUG-1240） |
+| `TASK-rectification-nadi-seconds-research-20261005.md` | `PROGRESS-rectification-nadi-seconds-research-20261005.md` | **「纳迪秒级校准」可证伪检验（离线）**：竞品宣传「问前事到天 → 秒级」。本仓主链只用三层小运，Sookshma / Prana 与 D150 从未进评价集；v5 真值 52/77 是整 5 分钟（秒级无真值可对）。N0 五层小运 + D150 底座（前三层与主链对账 0 差）、N1 拟合率 vs 安慰剂日期（核心）、N2 留一件预测、N3 六题后区间内再细分能否提头名、N4 岁差 / 坐标 / 时间扰动的噪声地板、N5 D150 结构层（原文比对 blocked）、N6 结论 + 对外口径草稿。规则先登记再跑；不改生产代码；不得重调 BUG-1091 已关的权重 | 待验收 | 分支 `codex/rectification-nadi-seconds-research-20261005`（BUG-1240 `closed_by_design`；未推送、未部署） |

scripts/research/nadi_seconds_lib.py

@@ -0,0 +1,574 @@
+#!/usr/bin/env python3
+"""Offline falsification helpers for second-scale nadi / sookshma / prana claims.
+
+Research only. This module imports production chart and Vimshottari functions
+and does not change them. Fitters and rankers never read ``true_minute``.
+
+The five-level lords are the production chain
+(``dasha_analyzer.build_antardasha`` applied twice more past pratyantar).
+That is the same recursion ``_active_vimshottari`` uses for the first three
+levels, including its midnight date anchor and 365.25-day year.
+``calculate_five_level_dasha`` stays available as a side check; it is not the
+scoring chain, because its 365.25636-day year does not reconcile to zero
+against the production lords.
+
+Classical D150 order (movable direct, fixed reverse, dual from the middle)
+is marked ``variant_unverified``. No Chandra Kala Nadi verse text is stored.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import math
+import sys
+from collections.abc import Mapping, Sequence
+from datetime import date, datetime, timedelta
+from pathlib import Path
+from random import Random
+from typing import Any
+
+_SCRIPTS = str(Path(__file__).resolve().parents[1])
+if _SCRIPTS not in sys.path:
+    sys.path.insert(0, _SCRIPTS)
+
+import dasha_analyzer
+import divisional_charts_extended
+import domain_calculation_service
+import narayana_dasha
+from scripts.active_rectification_event_engine import (
+    AYANAMSA,
+    DOMAIN_CONFIG,
+    NODE_MODE,
+    _active_vimshottari,
+    _event_datetime,
+    _house_lords,
+)
+
+ROOT = Path(__file__).resolve().parents[2]
+HOLDOUT_V5 = ROOT / "references" / "real_case_calibration" / "minute_rectification_holdout_v5.json"
+PREREG_PATH = ROOT / "docs" / "research" / "nadi_seconds_preregistration_2026_10_05.json"
+EVENT_CUTOFF = date(2026, 9, 14)
+LEVELS = ("md", "ad", "pd", "sookshma", "prana")
+# Structural reading only. See classical_nadi_index.
+CLASSICAL_NADI_STATUS = "variant_unverified"
+KM_PER_DEGREE_LAT = 111.32
+
+_VARGA = divisional_charts_extended.DivisionalChartsCalculator()
+_SKY_CACHE: dict[tuple, dict[str, Any]] = {}
+
+
+def load_cases(path: Path = HOLDOUT_V5) -> list[dict[str, Any]]:
+    return list(json.loads(path.read_text(encoding="utf-8"))["cases"])
+
+
+def canonical_bytes(payload: Any) -> bytes:
+    return (
+        json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
+        + "\n"
+    ).encode("utf-8")
+
+
+def sha256_bytes(raw: bytes) -> str:
+    return hashlib.sha256(raw).hexdigest()
+
+
+def file_sha256(path: Path) -> str:
+    return hashlib.sha256(path.read_bytes()).hexdigest()
+
+
+def round6(value: float) -> float:
+    return round(float(value), 6)
+
+
+def subdivide(period: Mapping[str, Any]) -> list[dict[str, Any]]:
+    """One production antardasha split. Adjacent rows abut by construction."""
+    return list(dasha_analyzer.build_antardasha(dict(period)))
+
+
+def chain_at(birth_date: str, moon_longitude: float, event_at: datetime) -> list[dict[str, Any]]:
+    """Five nested production periods active at ``event_at``.
+
+    ``birth_date`` is the candidate's local calendar date (YYYY-MM-DD), the
+    same anchor ``_active_vimshottari`` passes through.
+    """
+    return _chain_from_timeline(_timeline(birth_date, moon_longitude), event_at)
+
+
+def five_lords(birth_date: str, moon_longitude: float, event_at: datetime) -> tuple[str, str, str, str, str]:
+    return tuple(str(period["lord"]) for period in chain_at(birth_date, moon_longitude, event_at))  # type: ignore[return-value]
+
+
+def production_three(birth_date: str, moon_longitude: float, event_at: datetime) -> tuple[str, str, str]:
+    return _active_vimshottari(birth_date, float(moon_longitude) % 360.0, event_at)
+
+
+def d150_equal(longitude: float) -> dict[str, Any]:
+    """Equal-slice D150 via the repository ``calc_custom_varga`` (N=150)."""
+    row = _VARGA.calc_custom_varga(float(longitude) % 360.0, 150)
+    sign = str(row["sign"])
+    return {
+        "part_index": int(row["part_index"]),
+        "sign": sign,
+        "sign_idx": int(row["sign_idx"]),
+        "sign_lord": narayana_dasha.SIGN_LORDS[sign],
+        "source": "calc_custom_varga",
+    }
+
+
+def classical_nadi_index(longitude: float) -> dict[str, Any]:
+    """Reorder the 150 equal slices. Not a gate input.
+
+    Movable / fixed / dual follow ``DivisionalChartsCalculator`` sign classes.
+    Movable: part 0..149 in longitude order. Fixed: 149 down to 0.
+    Dual: start at the middle index 75 and wrap. The dual start is a
+    structural reading of "从中段起排", not a quotation, so the status is
+    ``variant_unverified``. Named nadi verses are not used.
+    """
+    longitude = float(longitude) % 360.0
+    sign_index = int(longitude // 30.0) % 12
+    part = int(d150_equal(longitude)["part_index"])
+    if sign_index in _VARGA.MOVABLE_SIGNS:
+        index = part
+        order = "direct"
+    elif sign_index in _VARGA.FIXED_SIGNS:
+        index = 149 - part
+        order = "reverse"
+    else:
+        index = (part + 75) % 150
+        order = "from_middle"
+    return {
+        "index": index,
+        "equal_part": part,
+        "sign_index": sign_index,
+        "order": order,
+        "status": CLASSICAL_NADI_STATUS,
+    }
+
+
+def domain_target_lords(ascendant_index: int, domain: str) -> frozenset[str]:
+    houses = DOMAIN_CONFIG[domain][1]
+    return frozenset(_house_lords(int(ascendant_index), houses))
+
+
+def explained(lords: Sequence[str], targets: frozenset[str], rule: str, segment_lord: str | None = None) -> bool:
+    """Binary 'this second explains this event' for one pre-registered rule.
+
+    A4 / A5: level 4 or 5 lord is a domain house lord.
+    B1..B5: at least that many of the five levels are domain house lords.
+    N5A: equal-division D150 lagna sign lord is a domain house lord.
+    N5B: that same sign lord appears in any of the five levels.
+    """
+    lords = tuple(lords)
+    if rule == "A4":
+        return lords[3] in targets
+    if rule == "A5":
+        return lords[4] in targets
+    if rule.startswith("B") and rule[1:].isdigit():
+        need = int(rule[1:])
+        return sum(lord in targets for lord in lords) >= need
+    if rule == "N5A":
+        return segment_lord is not None and segment_lord in targets
+    if rule == "N5B":
+        return segment_lord is not None and segment_lord in set(lords)
+    raise ValueError(f"unknown rule {rule}")
+
+
+def fit_count(
+    lord_rows: Sequence[Sequence[str]],
+    domains: Sequence[str],
+    ascendant_index: int,
+    rule: str,
+    segment_lord: str | None = None,
+) -> int:
+    """How many events ``rule`` explains. No clock and no truth label."""
+    total = 0
+    for lords, domain in zip(lord_rows, domains, strict=True):
+        targets = domain_target_lords(ascendant_index, domain)
+        if explained(lords, targets, rule, segment_lord):
+            total += 1
+    return total
+
+
+def rank_offsets(counts: Mapping[int, int]) -> list[int]:
+    """Offsets tied for the highest count, ascending. Ties are all kept."""
+    if not counts:
+        return []
+    best = max(counts.values())
+    return sorted(offset for offset, count in counts.items() if count == best)
+
+
+def candidate_moment(birth: Mapping[str, Any], offset_seconds: int) -> datetime:
+    base = datetime.strptime(f"{birth['date']} {str(birth['time'])[:5]}", "%Y-%m-%d %H:%M")
+    return base + timedelta(seconds=int(offset_seconds))
+
+
+def sky_at(
+    birth: Mapping[str, Any],
+    moment: datetime,
+    *,
+    ayanamsa: str = AYANAMSA,
+    node_mode: str = NODE_MODE,
+    longitude: float | None = None,
+) -> dict[str, Any]:
+    """Moon and ascendant from the production chart. Seconds are passed through."""
+    lat = round(float(birth["latitude"]), 6)
+    lon = round(float(birth["longitude"] if longitude is None else longitude), 6)
+    tz = round(float(birth["timezone_offset"]), 4)
+    key = (moment.year, moment.month, moment.day, moment.hour, moment.minute, moment.second, lat, lon, tz, str(ayanamsa), str(node_mode))
+    cached = _SKY_CACHE.get(key)
+    if cached is not None:
+        return cached
+    chart = domain_calculation_service.compute_chart({
+        "year": moment.year,
+        "month": moment.month,
+        "day": moment.day,
+        "hour": moment.hour,
+        "minute": moment.minute,
+        "second": moment.second,
+        "lat": lat,
+        "lon": lon,
+        "tz": tz,
+        "ayanamsa": ayanamsa,
+        "node_mode": node_mode,
+    })
+    moon = float(chart["planets"]["Moon"]["lon"]) % 360.0
+    asc = float(chart["ascendant"]["lon"]) % 360.0
+    rahu = chart.get("planets", {}).get("Rahu") or {}
+    rahu_lon = float(rahu["lon"]) % 360.0 if isinstance(rahu.get("lon"), (int, float)) else None
+    sky = {
+        "birth_date": moment.date().isoformat(),
+        "moon_lon": moon,
+        "asc_lon": asc,
+        "asc_index": int(asc // 30.0) % 12,
+        "rahu_lon": rahu_lon,
+    }
+    _SKY_CACHE[key] = sky
+    return sky
+
+
+def shift_longitude_km(longitude: float, latitude: float, kilometers_east: float) -> float:
+    """Move the birth longitude by an east-west ground distance.
+
+    One degree of longitude is ``111.32 * cos(latitude)`` kilometres.
+    Near a pole the east-west degree collapses and the shift is refused.
+    """
+    scale = KM_PER_DEGREE_LAT * math.cos(math.radians(float(latitude)))
+    if abs(scale) < 1e-3:
+        raise ValueError("east-west shift is undefined this close to a pole")
+    moved = float(longitude) + float(kilometers_east) / scale
+    return ((moved + 180.0) % 360.0) - 180.0
+
+
+def day_events(case: Mapping[str, Any]) -> list[dict[str, Any]]:
+    """Day-precision events copied without truth labels or narrative text."""
+    rows = []
+    for event in case["events"]:
+        if event.get("precision") != "day":
+            continue
+        rows.append({
+            "id": str(event["id"]),
+            "domain": str(event["domain"]),
+            "date": str(event["date"])[:10],
+            "precision": "day",
+        })
+    return rows
+
+
+def _clamp_life_date(day: date, birth: date, cutoff: date) -> date:
+    guard = 0
+    while day <= birth and guard < 160:
+        day = day + timedelta(days=365)
+        guard += 1
+    guard = 0
+    while day > cutoff and guard < 160:
+        day = day - timedelta(days=365)
+        guard += 1
+    if birth < day <= cutoff:
+        return day
+    fallback = birth + timedelta(days=400)
+    if fallback > cutoff:
+        fallback = cutoff
+    if fallback <= birth:
+        fallback = birth + timedelta(days=1)
+    return fallback
+
+
+def placebo_shift_dates(
+    events: Sequence[Mapping[str, Any]],
+    birth_date: str,
+    *,
+    seed: str,
+    low_days: int,
+    high_days: int,
+    cutoff: date = EVENT_CUTOFF,
+) -> list[dict[str, Any]]:
+    """Shift each day-event date by a seeded ±[low, high] day offset.
+
+    Domain stays. A result outside (birth, cutoff] is flipped, then repaired
+    by 365-day steps. The seed includes the case id so case order cannot
+    change one case's offsets.
+    """
+    rng = Random(seed)
+    birth = date.fromisoformat(birth_date)
+    shifted = []
+    for event in events:
+        original = date.fromisoformat(str(event["date"])[:10])
+        magnitude = rng.randint(int(low_days), int(high_days))
+        sign = rng.choice((-1, 1))
+        candidate = original + timedelta(days=sign * magnitude)
+        if not (birth < candidate <= cutoff):
+            candidate = original + timedelta(days=-sign * magnitude)
+        if not (birth < candidate <= cutoff):
+            candidate = _clamp_life_date(original + timedelta(days=magnitude), birth, cutoff)
+        shifted.append({**event, "date": candidate.isoformat()})
+    return shifted
+
+
+def placebo_swap_dates(
+    cases: Sequence[Mapping[str, Any]],
+    *,
+    seed: int,
+    cutoff: date = EVENT_CUTOFF,
+) -> dict[str, list[dict[str, Any]]]:
+    """Permute day-event dates across cases. Domains stay on the recipient event."""
+    slots: list[tuple[str, dict[str, Any]]] = []
+    for case in cases:
+        for event in day_events(case):
+            slots.append((str(case["case_id"]), event))
+    slots.sort(key=lambda item: (item[0], item[1]["id"]))
+    dates = [item[1]["date"] for item in slots]
+    order = list(range(len(dates)))
+    Random(int(seed)).shuffle(order)
+    by_case: dict[str, list[dict[str, Any]]] = {}
+    births = {str(case["case_id"]): str(case["birth"]["date"]) for case in cases}
+    for index, (case_id, event) in enumerate(slots):
+        raw = date.fromisoformat(dates[order[index]])
+        clamped = _clamp_life_date(raw, date.fromisoformat(births[case_id]), cutoff)
+        by_case.setdefault(case_id, []).append({**event, "date": clamped.isoformat()})
+    return by_case
+
+
+def event_moments(events: Sequence[Mapping[str, Any]]) -> list[datetime]:
+    return [_event_datetime(event) for event in events]  # type: ignore[arg-type]
+
+
+def _timeline(birth_date: str, moon_longitude: float) -> list[dict[str, Any]]:
+    nakshatra, progress, _pada = dasha_analyzer.lon_to_nakshatra(float(moon_longitude) % 360.0)
+    timeline, _elapsed, _remaining, _lord = dasha_analyzer.build_dasha_timeline(
+        birth_date, nakshatra, progress,
+    )
+    return timeline
+
+
+def _chain_from_timeline(timeline: Sequence[Mapping[str, Any]], event_at: datetime) -> list[dict[str, Any]]:
+    _index, major = dasha_analyzer.find_current(list(timeline), event_at)
+    levels = [major]
+    for _depth in range(4):
+        levels.append(dasha_analyzer.find_current_sub(subdivide(levels[-1]), event_at))
+    return levels
+
+
+def lords_for_events(
+    birth_date: str,
+    moon_longitude: float,
+    events: Sequence[Mapping[str, Any]],
+) -> list[tuple[str, str, str, str, str]]:
+    """Five lords for every event. The mahadasha timeline is built once."""
+    timeline = _timeline(birth_date, moon_longitude)
+    rows = []
+    for moment in event_moments(events):
+        rows.append(tuple(str(period["lord"]) for period in _chain_from_timeline(timeline, moment)))  # type: ignore[arg-type]
+    return rows
+
+
+def counts_by_offset(
+    skies: Mapping[int, Mapping[str, Any]],
+    events: Sequence[Mapping[str, Any]],
+    rule: str,
+    *,
+    use_segment_lord: bool,
+) -> dict[int, int]:
+    domains = [str(event["domain"]) for event in events]
+    counts: dict[int, int] = {}
+    for offset, sky in skies.items():
+        rows = lords_for_events(str(sky["birth_date"]), float(sky["moon_lon"]), events)
+        segment_lord = str(d150_equal(float(sky["asc_lon"]))["sign_lord"]) if use_segment_lord else None
+        counts[int(offset)] = fit_count(rows, domains, int(sky["asc_index"]), rule, segment_lord)
+    return counts
+
+
+def mean_rate(counts: Mapping[int, int], offsets: Sequence[int], event_count: int) -> float | None:
+    if event_count <= 0 or not offsets:
+        return None
+    total = sum(counts[offset] for offset in offsets)
+    return total / (len(list(offsets)) * event_count)
+
+
+def fraction_explaining_all(counts: Mapping[int, int], offsets: Sequence[int], event_count: int) -> float | None:
+    if event_count <= 0 or not offsets:
+        return None
+    hits = sum(1 for offset in offsets if counts[offset] == event_count)
+    return hits / len(list(offsets))
+
+
+def paired_sign_flip_p(
+    differences: Sequence[float],
+    *,
+    seed: int,
+    permutations: int,
+) -> dict[str, Any]:
+    """One-sided paired permutation: share of sign-flips with mean >= observed.
+
+    p = (count + 1) / (permutations + 1), counting the observed assignment.
+    """
+    values = [float(item) for item in differences]
+    n = len(values)
+    if n == 0:
+        return {"n": 0, "mean": None, "p": None, "extreme": None, "permutations": permutations}
+    observed = sum(values) / n
+    rng = Random(int(seed))
+    extreme = 0
+    for _ in range(int(permutations)):
+        total = 0.0
+        for value in values:
+            total += value if rng.randrange(2) == 0 else -value
+        if total / n >= observed - 1e-15:
+            extreme += 1
+    return {
+        "n": n,
+        "mean": round6(observed),
+        "p": round6((extreme + 1) / (int(permutations) + 1)),
+        "extreme": extreme,
+        "permutations": int(permutations),
+    }
+
+
+def bootstrap_mean_ci(
+    values: Sequence[float],
+    *,
+    seed: int,
+    resamples: int,
+) -> dict[str, Any]:
+    """Percentile 95% interval of the mean. Passes only when the lower bound is > 0."""
+    pool = [float(item) for item in values]
+    n = len(pool)
+    if n == 0:
+        return {"n": 0, "mean": None, "low": None, "high": None, "excludes_zero_positive": False}
+    rng = Random(int(seed))
+    means = []
+    for _ in range(int(resamples)):
+        draw = 0.0
+        for _item in range(n):
+            draw += pool[rng.randrange(n)]
+        means.append(draw / n)
+    means.sort()
+    low_index = int(0.025 * (len(means) - 1))
+    high_index = int(0.975 * (len(means) - 1))
+    low = means[low_index]
+    high = means[high_index]
+    return {
+        "n": n,
+        "mean": round6(sum(pool) / n),
+        "low": round6(low),
+        "high": round6(high),
+        "excludes_zero_positive": bool(low > 0.0),
+    }
+
+
+def minute_of(clock: str) -> int:
+    return int(str(clock)[3:5])
+
+
+def rounded_to_five_minutes(clock: str) -> bool:
+    return minute_of(clock) % 5 == 0
+
+
+def clock_delta_minutes(stamp: str, recorded: str) -> int:
+    def _clock(value: str) -> int:
+        hours, minutes = str(value)[:5].split(":")
+        return int(hours) * 60 + int(minutes)
+
+    delta = (_clock(stamp) - _clock(recorded)) % 1440
+    return delta - 1440 if delta > 720 else delta
+
+
+def leaders_within(offsets: Sequence[int], limit_seconds: int) -> bool:
+    """True only when every tied leader is inside the limit. An empty tie is a miss."""
+    return bool(offsets) and all(abs(int(offset)) <= int(limit_seconds) for offset in offsets)
+
+
+def truth_audit(cases: Sequence[Mapping[str, Any]]) -> dict[str, Any]:
+    """Evaluator-side audit. This function is allowed to read the recorded clock.
+
+    It does not read ``true_minute``; the recorded minute is ``birth.time``.
+    """
+    rounded: list[str] = []
+    unrounded: list[str] = []
+    minute_hist: dict[str, int] = {}
+    per_case = []
+    for case in cases:
+        clock = str(case["birth"]["time"])[:5]
+        minute = f"{minute_of(clock):02d}"
+        minute_hist[minute] = minute_hist.get(minute, 0) + 1
+        case_id = str(case["case_id"])
+        (rounded if rounded_to_five_minutes(clock) else unrounded).append(case_id)
+        day_count = sum(1 for event in case["events"] if event.get("precision") == "day")
+        per_case.append({"case_id": case_id, "minute": minute, "day_events": day_count})
+    day_counts = [row["day_events"] for row in per_case]
+    return {
+        "cases": len(per_case),
+        "rounded_5min_count": len(rounded),
+        "unrounded_count": len(unrounded),
+        "rounded_5min_case_ids": rounded,
+        "unrounded_case_ids": unrounded,
+        "minute_histogram": minute_hist,
+        "day_event_total": sum(day_counts),
+        "day_events_ge_3": sum(1 for count in day_counts if count >= 3),
+        "day_events_ge_4": sum(1 for count in day_counts if count >= 4),
+        "day_events_ge_5": sum(1 for count in day_counts if count >= 5),
+        "day_events_zero": sum(1 for count in day_counts if count == 0),
+        "per_case": per_case,
+    }
+
+
+def reconcile_case(case: Mapping[str, Any]) -> dict[str, Any]:
+    """First three lords versus ``_active_vimshottari`` for every event at the recorded second."""
+    moment = candidate_moment(case["birth"], 0)
+    sky = sky_at(case["birth"], moment)
+    mismatches = []
+    for event in case["events"]:
+        event_at = _event_datetime(event)  # type: ignore[arg-type]
+        ours = five_lords(sky["birth_date"], sky["moon_lon"], event_at)
+        theirs = production_three(sky["birth_date"], sky["moon_lon"], event_at)
+        if ours[:3] != theirs:
+            mismatches.append(str(event["id"]))
+    return {
+        "case_id": str(case["case_id"]),
+        "events": len(case["events"]),
+        "mismatches": mismatches,
+    }
+
+
+def public_fields(case: Mapping[str, Any]) -> dict[str, Any]:
+    """Copy the birth clock, place, and events a fitter may see."""
+    birth = case["birth"]
+    return {
+        "case_id": str(case["case_id"]),
+        "birth": {
+            "date": str(birth["date"]),
+            "time": str(birth["time"])[:5],
+            "latitude": float(birth["latitude"]),
+            "longitude": float(birth["longitude"]),
+            "timezone_offset": float(birth["timezone_offset"]),
+        },
+        "events": [
+            {
+                "id": str(event["id"]),
+                "domain": str(event["domain"]),
+                "date": str(event["date"]),
+                "precision": str(event["precision"]),
+            }
+            for event in case["events"]
+        ],
+    }

scripts/research/nadi_seconds_run.py

@@ -0,0 +1,676 @@
+#!/usr/bin/env python3
+"""Run the pre-registered nadi / sookshma / prana falsification study.
+
+Reads ``docs/research/nadi_seconds_preregistration_2026_10_05.json`` and
+refuses to score without it. The output JSON has no timestamps, so two runs
+with ``PYTHONHASHSEED=0`` are byte-identical. This script does not import
+``true_minute``.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+import traceback
+from datetime import date
+from pathlib import Path
+from random import Random
+from typing import Any, Mapping, Sequence
+
+ROOT = Path(__file__).resolve().parents[2]
+if str(ROOT) not in sys.path:
+    sys.path.insert(0, str(ROOT))
+
+from scripts.research import nadi_seconds_lib as lib  # noqa: E402
+
+REPORT = ROOT / "docs" / "research" / "nadi_seconds_results_2026_10_05.json"
+
+
+def load_prereg() -> dict[str, Any]:
+    if not lib.PREREG_PATH.is_file():
+        raise SystemExit(f"missing preregistration: {lib.PREREG_PATH}")
+    return json.loads(lib.PREREG_PATH.read_text(encoding="utf-8"))
+
+
+def offsets_of(radius: int, step: int) -> list[int]:
+    return list(range(-int(radius), int(radius) + 1, int(step)))
+
+
+def truth_minute_offsets() -> list[int]:
+    return list(range(0, 60))
+
+
+def collect_skies(birth: Mapping[str, Any], offsets: Sequence[int], **kwargs: Any) -> dict[int, dict[str, Any]]:
+    return {int(offset): lib.sky_at(birth, lib.candidate_moment(birth, int(offset)), **kwargs) for offset in offsets}
+
+
+def lord_bundle(skies: Mapping[int, Mapping[str, Any]], events: Sequence[Mapping[str, Any]]) -> dict[str, Any]:
+    lords: dict[int, list] = {}
+    asc: dict[int, int] = {}
+    segment: dict[int, str] = {}
+    part: dict[int, int] = {}
+    for offset, sky in skies.items():
+        lords[int(offset)] = lib.lords_for_events(str(sky["birth_date"]), float(sky["moon_lon"]), events)
+        asc[int(offset)] = int(sky["asc_index"])
+        row = lib.d150_equal(float(sky["asc_lon"]))
+        segment[int(offset)] = str(row["sign_lord"])
+        part[int(offset)] = int(row["part_index"])
+    return {"lords": lords, "asc": asc, "segment": segment, "part": part}
+
+
+def rule_counts(bundle: Mapping[str, Any], events: Sequence[Mapping[str, Any]], rule: str) -> dict[int, int]:
+    domains = [str(event["domain"]) for event in events]
+    use_segment = rule.startswith("N5")
+    counts: dict[int, int] = {}
+    for offset, rows in bundle["lords"].items():
+        targets = [lib.domain_target_lords(bundle["asc"][offset], domain) for domain in domains]
+        segment = bundle["segment"][offset] if use_segment else None
+        counts[offset] = sum(
+            lib.explained(lords, target, rule, segment)
+            for lords, target in zip(rows, targets, strict=True)
+        )
+    return counts
+
+
+def rule_bits(bundle: Mapping[str, Any], events: Sequence[Mapping[str, Any]], rule: str) -> dict[int, list[bool]]:
+    domains = [str(event["domain"]) for event in events]
+    use_segment = rule.startswith("N5")
+    bits: dict[int, list[bool]] = {}
+    for offset, rows in bundle["lords"].items():
+        targets = [lib.domain_target_lords(bundle["asc"][offset], domain) for domain in domains]
+        segment = bundle["segment"][offset] if use_segment else None
+        bits[offset] = [
+            lib.explained(lords, target, rule, segment)
+            for lords, target in zip(rows, targets, strict=True)
+        ]
+    return bits
+
+
+def mean_defined(values: Sequence[float | None]) -> float | None:
+    nums = [float(value) for value in values if value is not None]
+    if not nums:
+        return None
+    return lib.round6(sum(nums) / len(nums))
+
+
+def rate(flags: Sequence[bool]) -> float | None:
+    if not flags:
+        return None
+    return lib.round6(sum(bool(flag) for flag in flags) / len(flags))
+
+
+def split_rates(rows: Sequence[Mapping[str, Any]], key: str, rounded: set[str]) -> dict[str, float | None]:
+    return {
+        "all": rate([bool(row[key]) for row in rows]),
+        "rounded_5min": rate([bool(row[key]) for row in rows if row["case_id"] in rounded]),
+        "unrounded": rate([bool(row[key]) for row in rows if row["case_id"] not in rounded]),
+        "n_all": len(rows),
+        "n_rounded_5min": sum(1 for row in rows if row["case_id"] in rounded),
+        "n_unrounded": sum(1 for row in rows if row["case_id"] not in rounded),
+    }
+
+
+def delivery_offsets(start: str | None, end: str | None, recorded: str, radius_minutes: int) -> list[int]:
+    """One-second offsets covering each delivered clock minute, clipped to the search radius."""
+    if not start or not end:
+        return []
+    cap = int(radius_minutes) * 60
+    start_min = lib.clock_delta_minutes(start, recorded)
+    end_min = lib.clock_delta_minutes(end, recorded)
+
+    def clip(lo: int, hi: int) -> list[int]:
+        lo = max(lo, -cap)
+        hi = min(hi, cap)
+        if lo > hi:
+            return []
+        return list(range(lo, hi + 1))
+
+    if start_min <= end_min:
+        return clip(start_min * 60, end_min * 60 + 59)
+    return sorted(set(clip(start_min * 60, cap) + clip(-cap, end_min * 60 + 59)))
+
+
+def six_probe(case: Mapping[str, Any], radius: int) -> dict[str, Any]:
+    """futile_collect_stop_replay d1: six probes, no guided-window injection."""
+    from scripts.active_rectification_event_engine import AYANAMSA, NODE_MODE, compute_candidate_static_contexts
+    from scripts.rectification.event_probes import discriminating_event_probes
+    from scripts.rectification.refinement_packet import window_scan
+    from scripts.rectification.scoring_service import build_event_contribution_matrix, score_from_matrix
+    from scripts.research.fewer_probes_card_replay import _outcome
+    from scripts.research.guided_collect_holdout_replay import _hhmm, posterior_state
+    from scripts.research.minute_resolution_sweep import MINUTE_STEP, scoring_request_for
+    from scripts.research.probe_supply_after_six import ASK_COUNT, TODAY
+
+    recorded = str(case["birth"]["time"])[:5]
+    request = scoring_request_for(dict(case), radius)
+    request["ayanamsa"] = AYANAMSA
+    request["node_mode"] = NODE_MODE
+    request["minute_step"] = MINUTE_STEP
+    static_contexts = compute_candidate_static_contexts(request)
+    built = build_event_contribution_matrix(request, static_contexts=static_contexts)
+    rows = score_from_matrix(request, built)
+    times = [stamp for row in rows if (stamp := _hhmm(row.get("time")))]
+    probes = discriminating_event_probes(
+        {**request, "refresh_probes": False, "asked_probe_keys": []},
+        built,
+        scan=window_scan(built),
+        candidate_times=times,
+        representative_time=recorded,
+        today=TODAY,
+    )[:ASK_COUNT]
+    state = posterior_state(rows=rows, contexts=static_contexts, probes=probes, true_time=recorded)
+    outcome = _outcome(state, recorded)
+    scores = state["scores"]
+    valid = list(state["valid"])
+    if not valid:
+        leaders: list[str] = []
+    else:
+        def _score(row: Mapping[str, Any]) -> float:
+            stamp = _hhmm(row.get("time")) or ""
+            return float(scores.get(stamp, row.get("score") or 0))
+
+        best = max(_score(row) for row in valid)
+        leaders = sorted({
+            str(row.get("time"))[:5]
+            for row in valid
+            if abs(_score(row) - best) <= 1e-9 and str(row.get("time") or "")[:5]
+        })
+    deltas = [lib.clock_delta_minutes(stamp, recorded) for stamp in leaders]
+    return {
+        "start": outcome["start"],
+        "end": outcome["end"],
+        "width": outcome["width"],
+        "truth_in_range": bool(outcome["truth_in_range"]),
+        "leader_times": leaders,
+        "all_within_1min": bool(deltas) and all(abs(delta) <= 1 for delta in deltas),
+        "any_exact": recorded in leaders,
+        "error": None,
+    }
+
+
+def gate_p(result: Mapping[str, Any]) -> bool:
+    if result.get("mean") is None or result.get("extreme") is None:
+        return False
+    exact = (int(result["extreme"]) + 1) / (int(result["permutations"]) + 1)
+    return bool(result["mean"] > 0 and exact < 0.05)
+
+
+def summarize_rule(
+    rows: Sequence[Mapping[str, Any]],
+    *,
+    seed: int,
+    permutations: int,
+) -> dict[str, Any]:
+    diffs = [float(row["diff"]) for row in rows]
+    test = lib.paired_sign_flip_p(diffs, seed=seed, permutations=permutations)
+    return {
+        **test,
+        "pass": gate_p(test),
+        "truth_mean": mean_defined([row["truth"] for row in rows]),
+        "placebo_mean": mean_defined([row["placebo"] for row in rows]),
+        "window_mean": mean_defined([row["window"] for row in rows]),
+        "fraction_grid_explains_all_mean": mean_defined([row["fraction_all"] for row in rows]),
+    }
+
+
+def n2_case(
+    bits: Mapping[int, Sequence[bool]],
+    coarse: Sequence[int],
+    events: Sequence[Mapping[str, Any]],
+    placebo_bits: Mapping[int, Sequence[bool]],
+    *,
+    case_id: str,
+    seed: str,
+) -> dict[str, Any]:
+    deltas_random = []
+    deltas_placebo = []
+    held_rates = []
+    for index, event in enumerate(events):
+        counts = {
+            offset: sum(flag for event_index, flag in enumerate(flags) if event_index != index)
+            for offset, flags in bits.items()
+            if offset in coarse
+        }
+        leaders = [offset for offset in lib.rank_offsets(counts) if offset in set(coarse)]
+        if not leaders:
+            continue
+        held = sum(1 for offset in leaders if bits[offset][index]) / len(leaders)
+        rng = Random(f"{seed}:{case_id}:{event['id']}")
+        if len(leaders) >= len(coarse):
+            sample = list(coarse)
+        else:
+            sample = rng.sample(list(coarse), len(leaders))
+        random_rate = sum(1 for offset in sample if bits[offset][index]) / len(sample)
+        placebo_rate = sum(1 for offset in leaders if placebo_bits[offset][index]) / len(leaders)
+        held_rates.append(held)
+        deltas_random.append(held - random_rate)
+        deltas_placebo.append(held - placebo_rate)
+    return {
+        "held_rate": mean_defined(held_rates),
+        "delta_random": mean_defined(deltas_random),
+        "delta_placebo": mean_defined(deltas_placebo),
+    }
+
+
+def level_change(base: Sequence[Sequence[str]], other: Sequence[Sequence[str]], index: int) -> float | None:
+    if not base or len(base) != len(other):
+        return None
+    changed = sum(left[index] != right[index] for left, right in zip(base, other, strict=True))
+    return lib.round6(changed / len(base))
+
+
+def build(prereg: Mapping[str, Any], *, limit: int, radii: Sequence[int]) -> dict[str, Any]:
+    cases = [lib.public_fields(case) for case in lib.load_cases()]
+    if limit:
+        cases = cases[:limit]
+    grids = prereg["grids"]
+    coarse = offsets_of(grids["coarse_radius_seconds"], grids["coarse_step_seconds"])
+    fine = offsets_of(grids["fine_radius_seconds"], grids["fine_step_seconds"])
+    truth_offsets = truth_minute_offsets()
+    union = sorted(set(coarse) | set(fine) | set(truth_offsets))
+    rules = list(prereg["rules"]["computed"])
+    primary = list(prereg["rules"]["n1_primary"])
+    n1_min = int(prereg["samples"]["n1_min_day_events"])
+    n2_min = int(prereg["samples"]["n2_min_day_events"])
+    permutations = int(prereg["tests"]["permutations"])
+    resamples = int(prereg["tests"]["bootstrap_resamples"])
+    shift_low = int(prereg["placebo"]["shift_low_days"])
+    shift_high = int(prereg["placebo"]["shift_high_days"])
+    swap = lib.placebo_swap_dates(cases, seed=int(prereg["placebo"]["swap_seed"]))
+    rounded = {case_id for case_id in lib.truth_audit(cases)["rounded_5min_case_ids"]}
+    audit = lib.truth_audit(cases)
+    recon = [lib.reconcile_case(case) for case in cases]
+
+    n1_rows: dict[str, list[dict[str, Any]]] = {rule: [] for rule in primary}
+    n1_placebos = ("P1", "P2")
+    n5_rows: dict[str, list[dict[str, Any]]] = {rule: [] for rule in prereg["rules"]["n5_primary"]}
+    n2_rows = []
+    n3a_rows = []
+    n4_rows = []
+    fraction_rows: dict[str, list[float]] = {rule: [] for rule in ("A4", "A5", "B1", "B2")}
+    case_fit: list[dict[str, Any]] = []
+
+    for index, case in enumerate(cases, start=1):
+        events = lib.day_events(case)
+        birth = case["birth"]
+        print(f"nadi {index}/{len(cases)} {case['case_id']} day_events={len(events)}", flush=True)
+        skies = collect_skies(birth, union)
+        real = lord_bundle(skies, events) if events else None
+        shifted = lib.placebo_shift_dates(
+            events,
+            birth["date"],
+            seed=prereg["placebo"]["shift_seed_template"].format(case_id=case["case_id"]),
+            low_days=shift_low,
+            high_days=shift_high,
+        ) if events else []
+        p1 = lord_bundle(skies, shifted) if shifted else None
+        swapped = swap.get(case["case_id"], [])
+        p2 = lord_bundle(skies, swapped) if swapped else None
+        fit_row: dict[str, Any] = {
+            "case_id": case["case_id"],
+            "day_events": len(events),
+            "rounded_5min": case["case_id"] in rounded,
+        }
+        if real is not None:
+            for rule in rules:
+                if rule.startswith("N5") and rule not in prereg["rules"]["n5_primary"] and rule not in primary:
+                    continue
+                counts = rule_counts(real, events, rule)
+                truth = lib.mean_rate(counts, truth_offsets, len(events))
+                window = lib.mean_rate(counts, coarse, len(events))
+                fraction = lib.fraction_explaining_all(counts, coarse, len(events))
+                fit_row[rule] = {
+                    "truth": None if truth is None else lib.round6(truth),
+                    "window": None if window is None else lib.round6(window),
+                    "fraction_all": None if fraction is None else lib.round6(fraction),
+                }
+                if rule in fraction_rows and fraction is not None and len(events) >= n1_min:
+                    fraction_rows[rule].append(fraction)
+                if rule in primary and len(events) >= n1_min and p1 is not None and p2 is not None:
+                    for label, bundle in (("P1", p1), ("P2", p2)):
+                        placebo_rate = lib.mean_rate(rule_counts(bundle, shifted if label == "P1" else swapped, rule), truth_offsets, len(events))
+                        n1_rows[rule].append({
+                            "case_id": case["case_id"],
+                            "placebo": label,
+                            "truth": lib.round6(truth or 0.0),
+                            "placebo_rate": None if placebo_rate is None else lib.round6(placebo_rate),
+                            "diff": None if placebo_rate is None else lib.round6((truth or 0.0) - placebo_rate),
+                            "window": None if window is None else lib.round6(window),
+                            "fraction_all": None if fraction is None else lib.round6(fraction),
+                        })
+                if rule in n5_rows and len(events) >= n1_min and p1 is not None and p2 is not None:
+                    for label, bundle in (("P1", p1), ("P2", p2)):
+                        placebo_rate = lib.mean_rate(rule_counts(bundle, shifted if label == "P1" else swapped, rule), truth_offsets, len(events))
+                        n5_rows[rule].append({
+                            "case_id": case["case_id"],
+                            "placebo": label,
+                            "truth": lib.round6(truth or 0.0),
+                            "placebo_rate": None if placebo_rate is None else lib.round6(placebo_rate),
+                            "diff": None if placebo_rate is None else lib.round6((truth or 0.0) - placebo_rate),
+                            "window": None if window is None else lib.round6(window),
+                            "fraction_all": None if fraction is None else lib.round6(fraction),
+                        })
+            if len(events) >= n2_min and p1 is not None:
+                real_bits = rule_bits(real, events, prereg["rules"]["n2_primary"])
+                p1_bits = rule_bits(p1, shifted, prereg["rules"]["n2_primary"])
+                # P1 bits are aligned to shifted events, same index as real events.
+                n2_rows.append({
+                    "case_id": case["case_id"],
+                    "rounded_5min": case["case_id"] in rounded,
+                    **n2_case(
+                        real_bits, coarse, events, p1_bits,
+                        case_id=case["case_id"], seed=str(prereg["tests"]["n2_random_second_seed"]),
+                    ),
+                })
+            a5_counts = rule_counts(real, events, prereg["rules"]["n3_primary"]) if events else {}
+            leaders = lib.rank_offsets({offset: a5_counts.get(offset, 0) for offset in coarse}) if events else []
+            n3a_rows.append({
+                "case_id": case["case_id"],
+                "rounded_5min": case["case_id"] in rounded,
+                "day_events": len(events),
+                "hit_60s": lib.leaders_within(leaders, 60) if events else False,
+                "hit_120s": lib.leaders_within(leaders, 120) if events else False,
+                "tied_leaders": len(leaders),
+            })
+        else:
+            n3a_rows.append({
+                "case_id": case["case_id"],
+                "rounded_5min": case["case_id"] in rounded,
+                "day_events": 0,
+                "hit_60s": False,
+                "hit_120s": False,
+                "tied_leaders": len(coarse),
+            })
+        case_fit.append(fit_row)
+
+        # N4 at the recorded second only.
+        if events:
+            base_sky = skies[0]
+            base_lords = lib.lords_for_events(base_sky["birth_date"], base_sky["moon_lon"], events)
+            base_part = int(lib.d150_equal(base_sky["asc_lon"])["part_index"])
+            perturbations: dict[str, Any] = {}
+            for name in prereg["n4"]["ayanamsas"]:
+                if name == prereg["ayanamsa"]:
+                    continue
+                sky = lib.sky_at(birth, lib.candidate_moment(birth, 0), ayanamsa=name)
+                other = lib.lords_for_events(sky["birth_date"], sky["moon_lon"], events)
+                perturbations[f"ayanamsa:{name}"] = {
+                    "level4": level_change(base_lords, other, 3),
+                    "level5": level_change(base_lords, other, 4),
+                    "d150_changed": int(lib.d150_equal(sky["asc_lon"])["part_index"]) != base_part,
+                }
+            for seconds in prereg["n4"]["time_offsets_seconds"]:
+                sky = skies.get(int(seconds)) or lib.sky_at(birth, lib.candidate_moment(birth, int(seconds)))
+                other = lib.lords_for_events(sky["birth_date"], sky["moon_lon"], events)
+                perturbations[f"time:{int(seconds)}"] = {
+                    "level4": level_change(base_lords, other, 3),
+                    "level5": level_change(base_lords, other, 4),
+                    "d150_changed": int(lib.d150_equal(sky["asc_lon"])["part_index"]) != base_part,
+                }
+            for km in prereg["n4"]["longitude_shifts_km"]:
+                try:
+                    moved = lib.shift_longitude_km(float(birth["longitude"]), float(birth["latitude"]), float(km))
+                    sky = lib.sky_at(birth, lib.candidate_moment(birth, 0), longitude=moved)
+                    other = lib.lords_for_events(sky["birth_date"], sky["moon_lon"], events)
+                    perturbations[f"east_km:{km}"] = {
+                        "level4": level_change(base_lords, other, 3),
+                        "level5": level_change(base_lords, other, 4),
+                        "d150_changed": int(lib.d150_equal(sky["asc_lon"])["part_index"]) != base_part,
+                        "blocked": False,
+                    }
+                except ValueError:
+                    perturbations[f"east_km:{km}"] = {"blocked": True}
+            node_sky = lib.sky_at(birth, lib.candidate_moment(birth, 0), node_mode="true")
+            perturbations["node:true"] = {
+                "moon_changed": abs(float(node_sky["moon_lon"]) - float(base_sky["moon_lon"])) > 1e-6,
+                "asc_changed": abs(float(node_sky["asc_lon"]) - float(base_sky["asc_lon"])) > 1e-6,
+                "rahu_changed": node_sky["rahu_lon"] != base_sky["rahu_lon"],
+                "level5": level_change(base_lords, lib.lords_for_events(node_sky["birth_date"], node_sky["moon_lon"], events), 4),
+            }
+            n4_rows.append({"case_id": case["case_id"], "day_events": len(events), "perturbations": perturbations})
+
+    def pack_comparisons(bucket: Mapping[str, list[dict[str, Any]]], seed_base: int) -> dict[str, Any]:
+        packed = {}
+        cursor = 0
+        for rule, rows in bucket.items():
+            packed[rule] = {}
+            for label in n1_placebos:
+                subset = [
+                    {
+                        "diff": row["diff"],
+                        "truth": row["truth"],
+                        "placebo": row["placebo_rate"],
+                        "window": row["window"],
+                        "fraction_all": row["fraction_all"],
+                    }
+                    for row in rows
+                    if row["placebo"] == label and row["diff"] is not None
+                ]
+                packed[rule][label] = summarize_rule(subset, seed=seed_base + cursor, permutations=permutations)
+                cursor += 1
+        return packed
+
+    n1 = pack_comparisons(n1_rows, int(prereg["tests"]["n1_permutation_seed_base"]))
+    n5 = pack_comparisons(n5_rows, int(prereg["tests"]["n5_permutation_seed_base"]))
+    n1_pass = all(item["pass"] for rule in n1.values() for item in rule.values())
+    n5_pass = all(item["pass"] for rule in n5.values() for item in rule.values())
+    n2_values = [float(row["delta_random"]) for row in n2_rows if row["delta_random"] is not None]
+    n2_placebo_values = [float(row["delta_placebo"]) for row in n2_rows if row["delta_placebo"] is not None]
+    n2_ci = lib.bootstrap_mean_ci(n2_values, seed=int(prereg["tests"]["n2_bootstrap_seed"]), resamples=resamples)
+    n2_placebo_ci = lib.bootstrap_mean_ci(
+        n2_placebo_values, seed=int(prereg["tests"]["n2_bootstrap_seed"]) + 1, resamples=resamples,
+    )
+    uniform_60 = lib.round6(sum(1 for offset in coarse if abs(offset) <= 60) / len(coarse))
+    uniform_120 = lib.round6(sum(1 for offset in coarse if abs(offset) <= 120) / len(coarse))
+
+    # N3b. Separate from the fit loop so a replay failure does not drop N1.
+    n3b_rows = []
+    gate_radius = int(prereg["n3_protocol"]["radius_gate"])
+    for index, case in enumerate(cases, start=1):
+        print(f"replay {index}/{len(cases)} {case['case_id']}", flush=True)
+        by_radius = {}
+        for radius in radii:
+            try:
+                snapshot = six_probe(case, radius)
+            except Exception as exc:  # noqa: BLE001 — recorded as a miss, message dropped
+                snapshot = {
+                    "start": None,
+                    "end": None,
+                    "width": None,
+                    "truth_in_range": False,
+                    "leader_times": [],
+                    "all_within_1min": False,
+                    "any_exact": False,
+                    "error": type(exc).__name__,
+                }
+                traceback.print_exc(limit=2)
+            events = lib.day_events(case)
+            span = delivery_offsets(snapshot["start"], snapshot["end"], case["birth"]["time"], radius)
+            if events and span:
+                skies = collect_skies(case["birth"], span)
+                bundle = lord_bundle(skies, events)
+                counts = rule_counts(bundle, events, prereg["rules"]["n3_primary"])
+                leaders = lib.rank_offsets(counts)
+                nadi_hit = lib.leaders_within(leaders, int(prereg["n3_protocol"]["nadi_limit_seconds"]))
+                tied = len(leaders)
+            else:
+                nadi_hit = False
+                tied = 0
+            by_radius[str(radius)] = {
+                "truth_in_range": snapshot["truth_in_range"],
+                "width": snapshot["width"],
+                "baseline_all_within_1min": snapshot["all_within_1min"],
+                "baseline_any_exact": snapshot["any_exact"],
+                "nadi_all_within_60s": nadi_hit,
+                "nadi_tied_leaders": tied,
+                "improvement": int(nadi_hit) - int(snapshot["all_within_1min"]),
+                "error": snapshot["error"],
+            }
+        n3b_rows.append({
+            "case_id": case["case_id"],
+            "rounded_5min": case["case_id"] in rounded,
+            "radii": by_radius,
+        })
+
+    def n3b_summary(radius: int) -> dict[str, Any]:
+        key = str(radius)
+        improvements = [int(row["radii"][key]["improvement"]) for row in n3b_rows]
+        inside = sum(1 for row in n3b_rows if row["radii"][key]["truth_in_range"])
+        ci = lib.bootstrap_mean_ci(improvements, seed=int(prereg["tests"]["n3_bootstrap_seed"]) + radius, resamples=resamples)
+        return {
+            "radius": radius,
+            "truth_in_range": f"{inside}/{len(n3b_rows)}",
+            "truth_in_range_count": inside,
+            "truth_in_range_pass": inside >= 76 if len(n3b_rows) == 77 else None,
+            "baseline_all_within_1min": split_rates(
+                [{"case_id": row["case_id"], "hit": row["radii"][key]["baseline_all_within_1min"]} for row in n3b_rows],
+                "hit", rounded,
+            ),
+            "baseline_any_exact": rate([row["radii"][key]["baseline_any_exact"] for row in n3b_rows]),
+            "nadi_all_within_60s": split_rates(
+                [{"case_id": row["case_id"], "hit": row["radii"][key]["nadi_all_within_60s"]} for row in n3b_rows],
+                "hit", rounded,
+            ),
+            "improvement_ci": ci,
+            "pass": bool(ci["excludes_zero_positive"] and len(n3b_rows) == 77 and inside >= 76),
+        }
+
+    def n4_aggregate(label: str, field: str) -> dict[str, Any]:
+        values = []
+        d150_flags = []
+        weights = []
+        for row in n4_rows:
+            item = row["perturbations"].get(label) or {}
+            if item.get("blocked") or item.get(field) is None:
+                continue
+            values.append(float(item[field]))
+            weights.append(int(row["day_events"]))
+            if "d150_changed" in item:
+                d150_flags.append(bool(item["d150_changed"]))
+        event_rate = None
+        if weights and sum(weights):
+            event_rate = lib.round6(sum(value * weight for value, weight in zip(values, weights, strict=True)) / sum(weights))
+        return {
+            "event_rate": event_rate,
+            "case_rate": rate(d150_flags) if field == "level5" else None,
+            "d150_case_rate": rate(d150_flags),
+            "cases": len(values),
+        }
+
+    wording_hits = []
+    for name in prereg["n4"]["ayanamsas"]:
+        if name == prereg["ayanamsa"]:
+            continue
+        label = f"ayanamsa:{name}"
+        level5 = n4_aggregate(label, "level5")
+        wording_hits.append({
+            "switch": f"{prereg['ayanamsa']}->{name}",
+            "level5_event_rate": level5["event_rate"],
+            "d150_case_rate": level5["d150_case_rate"],
+            "over_20pct": bool(
+                (level5["event_rate"] is not None and level5["event_rate"] > 0.20)
+                or (level5["d150_case_rate"] is not None and level5["d150_case_rate"] > 0.20)
+            ),
+        })
+    wording_ban = any(item["over_20pct"] for item in wording_hits)
+
+    n3a = {
+        "grid": "coarse",
+        "uniform_within_60s": uniform_60,
+        "uniform_within_120s": uniform_120,
+        "published_engine_prior_top1_pm10": 0.18,
+        "hit_60s": split_rates(
+            [{"case_id": row["case_id"], "hit": row["hit_60s"]} for row in n3a_rows], "hit", rounded,
+        ),
+        "hit_120s": split_rates(
+            [{"case_id": row["case_id"], "hit": row["hit_120s"]} for row in n3a_rows], "hit", rounded,
+        ),
+    }
+    n3b = {str(radius): n3b_summary(radius) for radius in radii}
+    n3_pass = bool(n3b.get(str(gate_radius), {}).get("pass"))
+
+    return {
+        "prereg_sha256": lib.file_sha256(lib.PREREG_PATH),
+        "library_sha256": lib.file_sha256(Path(__file__).with_name("nadi_seconds_lib.py")),
+        "runner_sha256": lib.file_sha256(Path(__file__)),
+        "dataset_sha256": lib.file_sha256(lib.HOLDOUT_V5),
+        "swisseph": __import__("swisseph").version,
+        "limit": limit,
+        "n0": {
+            "audit": {key: value for key, value in audit.items() if key != "per_case"},
+            "per_case_day_events": audit["per_case"],
+            "reconciliation_mismatch_cases": [row["case_id"] for row in recon if row["mismatches"]],
+            "reconciliation_events": sum(row["events"] for row in recon),
+        },
+        "n1": {
+            "pass": n1_pass and not limit,
+            "comparisons": n1,
+            "fraction_grid_explains_all_mean": {rule: mean_defined(values) for rule, values in fraction_rows.items()},
+            "cases": case_fit,
+        },
+        "n2": {
+            "pass": bool(n2_ci["excludes_zero_positive"]) and not limit,
+            "versus_random": n2_ci,
+            "versus_placebo_date": n2_placebo_ci,
+            "cases": n2_rows,
+        },
+        "n3": {
+            "pass": n3_pass and not limit,
+            "n3a": n3a,
+            "n3b": n3b,
+            "cases_n3a": n3a_rows,
+            "cases_n3b": n3b_rows,
+        },
+        "n4": {
+            "wording_ban_second_scale": wording_ban,
+            "ayanamsa_switches": wording_hits,
+            "time": {str(seconds): n4_aggregate(f"time:{int(seconds)}", "level5") for seconds in prereg["n4"]["time_offsets_seconds"]},
+            "time_level4": {str(seconds): n4_aggregate(f"time:{int(seconds)}", "level4") for seconds in prereg["n4"]["time_offsets_seconds"]},
+            "longitude_km": {str(km): n4_aggregate(f"east_km:{km}", "level5") for km in prereg["n4"]["longitude_shifts_km"]},
+            "longitude_km_d150": {str(km): n4_aggregate(f"east_km:{km}", "level5")["d150_case_rate"] for km in prereg["n4"]["longitude_shifts_km"]},
+            "node_mean_to_true": {
+                "moon_changed_cases": sum(1 for row in n4_rows if row["perturbations"]["node:true"]["moon_changed"]),
+                "asc_changed_cases": sum(1 for row in n4_rows if row["perturbations"]["node:true"]["asc_changed"]),
+                "rahu_changed_cases": sum(1 for row in n4_rows if row["perturbations"]["node:true"]["rahu_changed"]),
+                "level5_event_rate": n4_aggregate("node:true", "level5")["event_rate"],
+            },
+            "cases": n4_rows,
+        },
+        "n5": {
+            "pass": n5_pass and not limit,
+            "named_text": "blocked",
+            "named_text_reason": "no legal Chandra Kala Nadi source in the repository",
+            "classical_index_status": lib.CLASSICAL_NADI_STATUS,
+            "comparisons": n5,
+        },
+        "gates": {
+            "n1": n1_pass and not limit,
+            "n2": bool(n2_ci["excludes_zero_positive"]) and not limit,
+            "n3": n3_pass and not limit,
+            "n4_blocks_second_scale_wording": wording_ban,
+            "n5": n5_pass and not limit,
+        },
+    }
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--limit", type=int, default=0)
+    parser.add_argument("--radii", default="10,30,60")
+    parser.add_argument("--out", default=str(REPORT))
+    args = parser.parse_args()
+    prereg = load_prereg()
+    radii = tuple(int(item) for item in str(args.radii).split(",") if item.strip())
+    payload = build(prereg, limit=int(args.limit), radii=radii)
+    text = json.dumps(payload, ensure_ascii=False, sort_keys=True, indent=2) + "\n"
+    out = Path(args.out)
+    out.parent.mkdir(parents=True, exist_ok=True)
+    out.write_text(text, encoding="utf-8")
+    print(json.dumps(payload["gates"], ensure_ascii=False, sort_keys=True), flush=True)
+    print(f"wrote {out}", flush=True)
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

tests/test_nadi_seconds_research.py

@@ -0,0 +1,179 @@
+"""Nadi / five-level dasha research helpers. Public v5 cases only; no production edits."""
+
+from __future__ import annotations
+
+import inspect
+import math
+from datetime import datetime, timedelta
+
+import pytest
+
+from scripts.research import nadi_seconds_lib as lib
+
+
+class _Guard(dict):
+    """Raises if a fitter touches the truth label."""
+
+    def __getitem__(self, key):
+        if key == "true_minute":
+            raise AssertionError("read true_minute")
+        return dict.__getitem__(self, key)
+
+    def get(self, key, default=None):
+        if key == "true_minute":
+            raise AssertionError("read true_minute")
+        return dict.get(self, key, default)
+
+
+def _packet() -> dict:
+    moment = datetime(2010, 1, 1)
+    lords = lib.five_lords("1990-06-15", 120.0, moment)
+    return {
+        "lords": lords,
+        "again": lib.five_lords("1990-06-15", 120.0, moment),
+        "d150": lib.d150_equal(10.25),
+        "classical": lib.classical_nadi_index(40.1),
+        "p": lib.paired_sign_flip_p([0.1, -0.2, 0.3, 0.0], seed=20261005, permutations=200),
+        "ci": lib.bootstrap_mean_ci([1.0, 0.0, 1.0, 0.0], seed=20261008, resamples=300),
+    }
+
+
+def test_two_runs_are_byte_identical() -> None:
+    assert lib.canonical_bytes(_packet()) == lib.canonical_bytes(_packet())
+
+
+def test_five_levels_abut_nest_and_flip_on_the_boundary() -> None:
+    birth = "1990-06-15"
+    moon = 0.0
+    event = datetime(1990, 6, 25)
+    chain = lib.chain_at(birth, moon, event)
+    assert len(chain) == 5
+    for period in chain:
+        assert period["start"] <= event < period["end"]
+    for period in chain[:4]:
+        rows = lib.subdivide(period)
+        assert rows[0]["start"] == period["start"]
+        for left, right in zip(rows, rows[1:], strict=False):
+            assert left["end"] == right["start"]
+    rows = lib.subdivide(chain[0])
+    boundary = rows[0]["end"]
+    before = lib.five_lords(birth, moon, boundary - timedelta(seconds=1))
+    after = lib.five_lords(birth, moon, boundary)
+    assert before[1] == rows[0]["lord"]
+    assert after[1] == rows[1]["lord"]
+    assert before[1] != after[1]
+
+
+def test_d150_part_flips_across_slice_and_sign_boundaries() -> None:
+    assert lib.d150_equal(0.199)["part_index"] == 0
+    assert lib.d150_equal(0.201)["part_index"] == 1
+    assert lib.d150_equal(29.999)["part_index"] == 149
+    assert lib.d150_equal(30.001)["part_index"] == 0
+    assert lib.classical_nadi_index(0.1)["status"] == "variant_unverified"
+    assert lib.classical_nadi_index(0.1)["index"] == 0
+    assert lib.classical_nadi_index(30.1)["order"] == "reverse"
+    assert lib.classical_nadi_index(30.1)["index"] == 149
+    assert lib.classical_nadi_index(60.0)["order"] == "from_middle"
+    assert lib.classical_nadi_index(60.0)["index"] == 75
+
+
+def test_rule_families_use_domain_house_lords_only() -> None:
+    assert lib.domain_target_lords(0, "career") == frozenset({"Saturn"})
+    lords = ("Sun", "Moon", "Mars", "Mercury", "Jupiter")
+    targets = frozenset({"Mercury"})
+    assert lib.explained(lords, targets, "A4") is True
+    assert lib.explained(lords, targets, "A5") is False
+    assert lib.explained(lords, targets, "B1") is True
+    assert lib.explained(lords, targets, "B2") is False
+    assert lib.explained(lords, targets, "N5A", segment_lord="Saturn") is False
+    assert lib.explained(lords, targets, "N5B", segment_lord="Moon") is True
+    assert lib.fit_count([lords, lords], ["career", "career"], 0, "A5") == 0
+    assert lib.rank_offsets({-5: 1, 0: 3, 5: 3, 10: 2}) == [0, 5]
+
+
+def test_fitter_does_not_read_true_minute() -> None:
+    for function in (
+        lib.fit_count,
+        lib.rank_offsets,
+        lib.explained,
+        lib.counts_by_offset,
+        lib.public_fields,
+        lib.lords_for_events,
+    ):
+        assert "true_minute" not in inspect.getsource(function)
+    raw = lib.load_cases()[0]
+    guarded = _Guard(raw)
+    guarded["birth"] = _Guard(raw["birth"])
+    guarded["events"] = [_Guard(event) for event in raw["events"]]
+    public = lib.public_fields(guarded)
+    assert "true_minute" not in public
+    sky = lib.sky_at(public["birth"], lib.candidate_moment(public["birth"], 0))
+    events = lib.day_events(public)[:1]
+    assert events
+    rows = lib.lords_for_events(sky["birth_date"], sky["moon_lon"], events)
+    count = lib.fit_count(rows, [events[0]["domain"]], sky["asc_index"], "A5")
+    assert count in (0, 1)
+    assert lib.rank_offsets({0: 2, 5: 2, -5: 1}) == [0, 5]
+
+
+def test_midnight_candidate_uses_the_previous_local_date_and_matches_production() -> None:
+    birth = {
+        "date": "1990-06-15",
+        "time": "00:00",
+        "latitude": 31.2,
+        "longitude": 121.5,
+        "timezone_offset": 8.0,
+    }
+    moment = lib.candidate_moment(birth, -30)
+    assert moment == datetime(1990, 6, 14, 23, 59, 30)
+    sky = lib.sky_at(birth, moment)
+    assert sky["birth_date"] == "1990-06-14"
+    event = datetime(2001, 3, 4)
+    assert lib.five_lords(sky["birth_date"], sky["moon_lon"], event)[:3] == lib.production_three(
+        sky["birth_date"], sky["moon_lon"], event,
+    )
+
+
+def test_v5_truth_audit_matches_the_task_counts() -> None:
+    audit = lib.truth_audit(lib.load_cases())
+    assert audit["cases"] == 77
+    assert audit["rounded_5min_count"] == 52
+    assert audit["unrounded_count"] == 25
+    assert audit["day_event_total"] == 244
+    assert audit["day_events_ge_3"] == 43
+    assert audit["day_events_ge_5"] == 19
+    assert audit["day_events_zero"] == 4
+    assert sum(audit["minute_histogram"].values()) == 77
+
+
+def test_first_three_levels_match_production_on_every_v5_event() -> None:
+    mismatches = []
+    for case in lib.load_cases():
+        row = lib.reconcile_case(case)
+        if row["mismatches"]:
+            mismatches.append(row["case_id"])
+    assert mismatches == []
+
+
+def test_placebo_shift_is_seeded_and_stays_after_birth() -> None:
+    case = lib.public_fields(lib.load_cases()[0])
+    events = lib.day_events(case)
+    first = lib.placebo_shift_dates(
+        events, case["birth"]["date"], seed=f"20261005:{case['case_id']}", low_days=30, high_days=180,
+    )
+    second = lib.placebo_shift_dates(
+        events, case["birth"]["date"], seed=f"20261005:{case['case_id']}", low_days=30, high_days=180,
+    )
+    assert first == second
+    birth = case["birth"]["date"]
+    assert all(birth < row["date"] <= "2026-09-14" for row in first)
+    assert [row["domain"] for row in first] == [row["domain"] for row in events]
+    assert [row["date"] for row in first] != [row["date"] for row in events]
+
+
+def test_longitude_shift_is_about_ten_kilometres() -> None:
+    moved = lib.shift_longitude_km(121.5, 31.2, 10.0)
+    scale = 111.32 * math.cos(math.radians(31.2))
+    assert moved == pytest.approx(121.5 + 10.0 / scale)
+    with pytest.raises(ValueError):
+        lib.shift_longitude_km(0.0, 90.0, 10.0)