feat(web): auto-discover exact evidence batches

my-tools/authorize_next_wbv2_task.py

@@ -277,6 +277,60 @@ def batch_contract(task: str, batch: dict) -> dict:
     return {"batchId": batch.get("batchId") or f"{task}:{batch['moduleId']}:{digest}", "selectedRecordIds": selected}
 
 
+def discover_existing_evidence_batch(inventory: dict) -> dict | None:
+    """Find exact Chromium receipts that list currently-unbound records.
+
+    This is an evidence-only discovery path: it never infers coverage from a
+    neighboring source file and never schedules a browser rerun. A new batch
+    receives a copied, task-bound evidence record through the module runner.
+    """
+    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"),
+        ("SHAPE_KEY", "web/tests/e2e/wbv2-shape-key-reader-production.spec.ts"),
+        ("MODIFIER", "web/tests/e2e/wbv2-modifier-reader-batch-two-production.spec.ts"),
+    )
+    for evidence_path in sorted(ROOT.glob("tests/golden/WBV2-P*-001/*.json"), key=lambda item: item.as_posix()):
+        try:
+            evidence = json.loads(evidence_path.read_text(encoding="utf-8"))
+        except (OSError, json.JSONDecodeError):
+            continue
+        if evidence.get("browser") != "chromium" or evidence.get("queueMutation") is not False:
+            continue
+        source_units = evidence.get("sourceUnits")
+        if not isinstance(source_units, list) or not source_units or not all(isinstance(unit, str) for unit in source_units):
+            continue
+        if not set(source_units).issubset(unresolved):
+            continue
+        operation = str(evidence.get("operation", ""))
+        test = next((path for marker, path in operation_tests if marker in operation), None)
+        fixture_paths = evidence.get("fixtures") if isinstance(evidence.get("fixtures"), list) else [evidence.get("fixture")]
+        fixture_paths = [fixture for fixture in fixture_paths if isinstance(fixture, str) and (ROOT / fixture).is_file()]
+        if not test or not (ROOT / test).is_file() or not fixture_paths:
+            continue
+        identifiers = [unit.split(":", 1)[1] for unit in source_units if unit.startswith("modifier-source:")]
+        rna = [unit.split(":", 1)[1] for unit in source_units if unit.startswith("rna-struct:")]
+        expected_types = evidence.get("expectedTypes") if isinstance(evidence.get("expectedTypes"), list) else []
+        return {
+            "moduleId": f"AUTO_EVIDENCE_REUSE_{evidence_path.parent.name}",
+            "title": f"reuse exact Chromium evidence from {evidence_path.parent.name}",
+            "owner": str(next((unresolved[unit].get("family") for unit in source_units if unresolved[unit].get("family")), "PROGRAM")),
+            "identifiers": identifiers,
+            "rna": rna,
+            "fixtures": fixture_paths,
+            "test": test,
+            "production": "tests/golden/{task}/auto-reused-production.json",
+            "types": expected_types,
+            "units": [unresolved[unit] for unit in source_units],
+            "candidateUnits": [unresolved[unit] for unit in source_units],
+            "reuseExistingEvidence": rel(evidence_path),
+            "reuseEvidenceSource": rel(evidence_path),
+        }
+
+
 def run_node(*args: str) -> None:
     result = subprocess.run(["node", *args], cwd=ROOT, text=True, capture_output=True, check=False)
     output = result.stdout or result.stderr
@@ -488,6 +542,9 @@ def select_batch(inventory: dict) -> dict:
                     next_size = current_size
                 chosen = unresolved[:max(min_size, min(max_size, len(unresolved), next_size))]
                 return {**candidate, "units": chosen, "candidateUnits": selected, "adaptiveKey": key, "adaptiveHistory": recent, "adaptiveCurrentSize": current_size, "adaptiveNextSize": len(chosen), "adaptiveMinSize": min_size, "adaptiveMaxSize": max_size}
+    discovered = discover_existing_evidence_batch(inventory)
+    if discovered:
+        return discovered
     raise SystemExit("no configured unresolved Blender source batch remains")
 
 
@@ -554,6 +611,8 @@ def main() -> int:
     module_record = {"id": module_task, "phase": str(phase), "title": batch["title"], "kind": "module", "owner": batch["owner"], "dependsOn": [root_task], "parityId": batch["moduleId"], "implementationClass": "LOCAL_EXACT", "parallelGroup": f"module-{batch['moduleId'].lower()}", "gate": "MODULE", "card": f"docs/tasks/{module_task}.md", "behavior": f"Verify the {batch['moduleId']} batch through Blender Main, WASM and Chromium recovery.", "outputs": ["aggregate module receipt", "coverage bindings", "manifest", "status"], "acceptance": ["source and protocol hashes are bound", "every selected Main reader type is proven", "negative no-mutation and recovery are proven", "Chromium and save/reopen evidence are exact"], "status": "in_progress", "ready": True, "queueMutationAllowed": True, "batchId": contract["batchId"], "selectedRecordIds": contract["selectedRecordIds"], "module": {"moduleId": batch["moduleId"], "batchId": contract["batchId"], "selectedRecordIds": contract["selectedRecordIds"], "sourceBacked": True, "capabilities": [f"CAP-{batch['moduleId']}-READER", f"CAP-{batch['moduleId']}-RECOVERY", f"CAP-{batch['moduleId']}-FAIL-CLOSED"], "sourcePaths": source_paths, "reuseReceipts": batch.get("reuseReceipts", [{"manifest": "tests/golden/WBV2-P1107-000/manifest.json", "report": "tests/golden/WBV2-P1107-000/module-receipt.json"}]), "dependencyReuse": batch.get("dependencyReuse", []), "positive": [f"{name} type is read from Blender Main" for name in batch["identifiers"]] + ["shared RNA metadata is preserved", "save/reopen preserves the batch metadata"], "negative": ["malformed blend is rejected without Main mutation", "unsupported batch data fails closed", "non-finite metadata is not synthesized"], "recovery": ["save/reopen", "worker termination", "deterministic second read"], "commands": commands, "fullFamilyParity": False, "evidencePaths": [batch["production"].format(task=module_task)], "coverageBindings": contract["selectedRecordIds"]}}
 
     module_record["module"]["commandBudget"] = {"maxCommands": 48, "maxReadCommands": 16}
+    if batch.get("reuseExistingEvidence"):
+        module_record["module"]["reuseExistingEvidence"] = batch["reuseExistingEvidence"]
     root_out = CONTROL_ROOT.parent / root_task
     root_plan_path = root_out / "module-execution-plan.json"
     root_plan = {**plan, "planId": f"BLENDER_WEB_MODULE_EXECUTION_P{phase}", "status": "ACTIVE_MODULE_QUEUE", "tasks": [*plan["tasks"], root_record, module_record], "modules": [*plan.get("modules", []), {"id": batch["moduleId"], "task": module_task, "owner": batch["owner"], "sourceBacked": True, "batchId": contract["batchId"], "selectedRecordIds": contract["selectedRecordIds"], "capabilities": module_record["module"]["capabilities"], "sourcePaths": source_paths, "reuseReceipts": module_record["module"]["reuseReceipts"], "fullFamilyParity": False}]}