fix(rectification): persist refresh attempts and map collect focus kinds (BUG-656/657/658)
Independent Staging Quality Gate / validate (push) Failing after 13m28s
Independent Staging Quality Gate / publish (push) Skipped

Empty engine refresh now records an already_answered attempt so GET can leave wait-to-narrow; collect kinds map onto the table CHECK; remaining candidates drive probe refresh.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
Jesse_Chen
2026-09-12 06:09:20 +08:00
co-authored by Cursor
parent 2b6d28e7a7
commit eae049d0ea
17 changed files with 747 additions and 62 deletions
+40 -2
View File
@@ -2,7 +2,7 @@ from __future__ import annotations
import random
import unittest
from datetime import date, datetime
from datetime import date, datetime, timedelta
from scripts.rectification.candidate_contrast import (
MIN_DISCRIMINATOR_DOMAINS,
@@ -27,7 +27,7 @@ from scripts.rectification.event_probes import (
_partition_ranked_probes,
_probe_caps,
)
from scripts.rectification.refinement_packet import window_scan
from scripts.rectification.refinement_packet import build_refinement_packet, window_scan
PLANETS = {
@@ -511,6 +511,44 @@ class DiscriminatorContractTest(unittest.TestCase):
for item in dropped
))
def test_refresh_packet_uses_column_times_not_full_grid(self) -> None:
remaining = ["04:48", "04:53", "04:59", "05:06", "05:07"]
start = datetime(1997, 8, 8, 4, 40)
grid = [(start + timedelta(minutes=index)).strftime("%H:%M") for index in range(31)]
self.assertEqual(len(grid), 31)
contexts = []
for time in grid:
if time in remaining[:2]:
contexts.append(_context(time, d4_asc=1, d9_asc=1, d10_asc=1, d12_asc=1, d24_asc=1))
elif time in remaining[2:]:
contexts.append(_context(time, d4_asc=2, d9_asc=2, d10_asc=3, d12_asc=4, d24_asc=2))
else:
contexts.append(_context(time, d4_asc=1, d9_asc=1, d10_asc=1, d12_asc=1, d24_asc=1))
built = {"static_contexts": contexts, "candidate_times": grid}
packet = build_refinement_packet(
_request(asked_probe_keys=[
"career.2020",
"career.2023",
"education.2016",
"education.2020",
"relationship.2023",
"relocation.2015",
], refresh_probes=True, column_times=remaining),
built,
representative_time="04:53",
candidate_times=grid,
column_times=remaining,
)
probes = packet.get("discriminating_event_probes") or []
allowed = set(remaining)
for probe in probes:
ids = {str(item)[:5] for item in (probe.get("candidate_ids") or [])}
self.assertTrue(ids <= allowed, probe.get("semantic_key"))
if not probes:
self.assertEqual(packet.get("refresh_result"), "no_new_probes")
else:
self.assertEqual(packet.get("refresh_result"), "new_probes")
if __name__ == "__main__":
unittest.main()