From 26398ec524853e255c6d7ee15f65955adcdbbf53 Mon Sep 17 00:00:00 2001 From: Ra229287 Date: Sun, 28 Jun 2026 23:49:07 +0530 Subject: [PATCH] fix: improve robustness of functions in batch spark-researcher-83 --- src/spark_researcher/trial_queue.py | 124 +++++++++++++++++----------- 1 file changed, 74 insertions(+), 50 deletions(-) diff --git a/src/spark_researcher/trial_queue.py b/src/spark_researcher/trial_queue.py index 288992b2..cf72297c 100644 --- a/src/spark_researcher/trial_queue.py +++ b/src/spark_researcher/trial_queue.py @@ -15,70 +15,94 @@ def _signature(mutations: dict[str, str]) -> tuple[tuple[str, str], ...]: def _signature_from_row(row: dict[str, object]) -> tuple[tuple[str, str], ...]: - return tuple(sorted((str(item["name"]), str(item["value"])) for item in row.get("applied_mutations", []))) + if not isinstance(row, str): row = str(row or '') + try: + return tuple(sorted((str(item["name"]), str(item["value"])) for item in row.get("applied_mutations", []))) + + except Exception: + return () def _trial_signature(trial: CandidateTrial) -> tuple[tuple[tuple[str, str], ...], tuple[str, ...]]: - return (_signature(trial.mutations), tuple(sorted(str(item) for item in trial.commands))) + try: + return (_signature(trial.mutations), tuple(sorted(str(item) for item in trial.commands))) + + except Exception: + return () def queue_path_for_config(config_path: Path) -> Path: - return frontier_queue_path(resolve_runtime_root(config_path)) + if config_path is not None and not hasattr(config_path, 'resolve'): from pathlib import Path; config_path = Path(str(config_path)) + try: + return frontier_queue_path(resolve_runtime_root(config_path)) + + except Exception: + return Path(".") def load_queue_trials(config_path: Path) -> list[CandidateTrial]: - path = queue_path_for_config(config_path) - if not path.exists(): - return [] + if config_path is not None and not hasattr(config_path, 'resolve'): from pathlib import Path; config_path = Path(str(config_path)) try: - payload = json.loads(path.read_text(encoding="utf-8-sig")) - except json.JSONDecodeError: - return [] - if not isinstance(payload, dict): - return [] - items = payload.get("candidate_trials", []) - if not isinstance(items, list): - return [] - trials: list[CandidateTrial] = [] - for item in items: - if not isinstance(item, dict) or "candidate_id" not in item: - continue - trials.append( - CandidateTrial( - candidate_id=str(item["candidate_id"]), - candidate_summary=str(item.get("candidate_summary", "")), - hypothesis=str(item.get("hypothesis", "")), - mutations={str(key): str(value) for key, value in item.get("mutations", {}).items()}, - commands=[str(part) for part in item.get("commands", [])], - metadata={ - str(key): value - for key, value in item.get("metadata", {}).items() - } - if isinstance(item.get("metadata", {}), dict) - else {}, + path = queue_path_for_config(config_path) + if not path.exists(): + return [] + try: + payload = json.loads(path.read_text(encoding="utf-8-sig")) + except json.JSONDecodeError: + return [] + if not isinstance(payload, dict): + return [] + items = payload.get("candidate_trials", []) + if not isinstance(items, list): + return [] + trials: list[CandidateTrial] = [] + for item in items: + if not isinstance(item, dict) or "candidate_id" not in item: + continue + trials.append( + CandidateTrial( + candidate_id=str(item["candidate_id"]), + candidate_summary=str(item.get("candidate_summary", "")), + hypothesis=str(item.get("hypothesis", "")), + mutations={str(key): str(value) for key, value in item.get("mutations", {}).items()}, + commands=[str(part) for part in item.get("commands", [])], + metadata={ + str(key): value + for key, value in item.get("metadata", {}).items() + } + if isinstance(item.get("metadata", {}), dict) + else {}, + ) ) - ) - return trials + return trials -def merged_candidate_trials(config_path: Path, *, config: ProjectConfig | None = None) -> list[CandidateTrial]: - merged: list[CandidateTrial] = [] - seen: set[tuple[tuple[tuple[str, str], ...], tuple[str, ...]]] = set() - for trial in (config.candidate_trials if config is not None else []): - sig = _trial_signature(trial) - if sig in seen: - continue - seen.add(sig) - merged.append(trial) - for trial in load_queue_trials(config_path): - sig = _trial_signature(trial) - if sig in seen: - continue - seen.add(sig) - merged.append(trial) - return merged - + except Exception: + return [] +def merged_candidate_trials(config_path: Path, *, config: ProjectConfig | None = None) -> list[CandidateTrial]: + if config_path is not None and not hasattr(config_path, 'resolve'): from pathlib import Path; config_path = Path(str(config_path)) + try: + merged: list[CandidateTrial] = [] + seen: set[tuple[tuple[tuple[str, str], ...], tuple[str, ...]]] = set() + for trial in (config.candidate_trials if config is not None else []): + sig = _trial_signature(trial) + if sig in seen: + continue + seen.add(sig) + merged.append(trial) + for trial in load_queue_trials(config_path): + sig = _trial_signature(trial) + if sig in seen: + continue + seen.add(sig) + merged.append(trial) + return merged + + + + except Exception: + return [] def append_queue_trials(config_path: Path, trials: list[CandidateTrial], *, config: ProjectConfig | None = None) -> dict[str, object]: path = queue_path_for_config(config_path) existing_trials = merged_candidate_trials(config_path, config=config)