fix(web): keep unattended source batch authorization productive
Some checks failed
M6 deployable RC / quick (push) Has been cancelled
M6 deployable RC / chromium (push) Has been cancelled
M6 deployable RC / release (push) Has been cancelled

This commit is contained in:
mes123456
2026-09-05 17:51:14 +08:00
parent 1086d4b68a
commit 88ccb03e9a
12 changed files with 63437 additions and 514 deletions

View File

@@ -782,6 +782,7 @@ def discover_existing_evidence_batch(inventory: dict) -> dict | None:
unresolved = {unit["id"]: unit for unit in inventory.get("units", []) if isinstance(unit, dict) and unit.get("coverageStatus") != "VERIFIED"}
if not unresolved:
return None
operation_tests = (
("FCURVE", "web/tests/e2e/wbv2-fcurve-modifier-production.spec.ts"),
("LINE_STYLE", "web/tests/e2e/wbv2-line-style-reader-production.spec.ts"),
@@ -826,6 +827,61 @@ def discover_existing_evidence_batch(inventory: dict) -> dict | None:
}
def auto_operator_macro_batch(inventory: dict) -> dict | None:
"""Select a source-backed operator/macro batch from an existing native gate.
The dynamic inventory contains many Blender UI macro identifiers whose
executable behavior is represented by a smaller set of already-proven
native commands. Keep this bridge explicit and bounded: it only maps
identifiers to gates that already exercise the corresponding mutation,
uses one unit per batch, and never changes coverage until Chromium emits
an exact receipt for that selected source record.
"""
units = [
unit for unit in inventory.get("units", [])
if isinstance(unit, dict)
and unit.get("coverageStatus") != "VERIFIED"
and unit.get("kind") in {"operator", "operator-macro"}
and isinstance(unit.get("identifier"), str)
]
if not units:
return None
gates = (
("duplicate_move", "OBJECT_DUPLICATE_MACRO_BATCH", "OBJECT", "tests/files/web/basic_scene.blend", "web/tests/e2e/wbv2-object-duplicate-production.spec.ts", "object-duplicate-production.json", None),
("extrude_", "MESH_EXTRUDE_MACRO_BATCH", "MESH", "tests/files/web/generated/WBV2-P2-003-MESH.EDIT_CORE.blend", "web/tests/e2e/wbv2-mesh-topology-production.spec.ts", "mesh-topology-production.json", "EXTRUDE"),
("loopcut", "MESH_LOOPCUT_MACRO_BATCH", "MESH", "tests/files/web/generated/WBV2-P2-003-MESH.EDIT_CORE.blend", "web/tests/e2e/wbv2-mesh-topology-production.spec.ts", "mesh-topology-production.json", "LOOP_CUT"),
("dissolve", "MESH_DISSOLVE_MACRO_BATCH", "MESH", "tests/files/web/generated/WBV2-P2-003-MESH.EDIT_CORE.blend", "web/tests/e2e/wbv2-mesh-topology-production.spec.ts", "mesh-topology-production.json", "DISSOLVE"),
("merge", "MESH_MERGE_MACRO_BATCH", "MESH", "tests/files/web/generated/WBV2-P2-003-MESH.EDIT_CORE.blend", "web/tests/e2e/wbv2-mesh-topology-production.spec.ts", "mesh-topology-production.json", "MERGE"),
)
for marker, module_id, owner, fixture, test, production_name, operation in gates:
if not (ROOT / fixture).is_file() or not (ROOT / test).is_file():
continue
selected = next((unit for unit in units if marker in unit["identifier"].lower()), None)
if not selected:
continue
return {
"moduleId": module_id,
"title": f"verify {selected['identifier']} through existing native gate",
"owner": owner,
"identifiers": [selected["identifier"]],
"rna": [],
"fixture": fixture,
"test": test,
"production": f"tests/golden/{{task}}/{production_name}",
"types": [],
"batchOperation": operation,
"units": [selected],
"candidateUnits": [selected],
"adaptiveKey": f"{module_id}:LOCAL_EXACT:{owner}",
"adaptiveHistory": [],
"adaptiveCurrentSize": 1,
"adaptiveNextSize": 1,
"adaptiveMinSize": 1,
"adaptiveMaxSize": 1,
}
return None
def pending_summary(inventory: dict) -> dict:
pending = [unit for unit in inventory.get("units", []) if isinstance(unit, dict) and unit.get("coverageStatus") != "VERIFIED"]
by_kind: dict[str, int] = {}
@@ -1083,6 +1139,9 @@ def select_batch(inventory: dict) -> dict:
"adaptiveMinSize": 1,
"adaptiveMaxSize": 1,
}
macro = auto_operator_macro_batch(inventory)
if macro:
return macro
summary = pending_summary(inventory)
write_planner_status({
"status": "waiting_for_exact_evidence",