diff --git a/batch_runner.py b/batch_runner.py index 9ceac8f5c8..347883a14d 100644 --- a/batch_runner.py +++ b/batch_runner.py @@ -208,7 +208,6 @@ def _prepare_container_image( except Exception as img_err: if config.get("verbose"): print(f" Prompt {prompt_index}: Docker image check failed: {img_err}", flush=True) - from tools.terminal_tool import register_task_env_overrides overrides = { "docker_image": container_image, @@ -233,7 +232,6 @@ def _process_single_prompt( """Run the agent on one prompt; returns trajectory, stats and metadata (or a failure result).""" prompt = prompt_data["prompt"] task_id = f"task_{prompt_index}" - failure = _prepare_container_image(prompt_index, prompt_data, batch_num, task_id, config) if failure is not None: return failure @@ -243,7 +241,6 @@ def _process_single_prompt( if config.get("verbose"): print(f" Prompt {prompt_index}: Using toolsets {selected_toolsets}") - agent = AIAgent( model=config["model"], max_iterations=config["max_iterations"], @@ -263,7 +260,6 @@ def _process_single_prompt( # Stats before conversion β€” keep the original evaluation order. tool_stats = _extract_tool_stats(result["messages"]) reasoning_stats = _extract_reasoning_stats(result["messages"]) - trajectory = agent._convert_to_trajectory_format(result["messages"], prompt, result["completed"]) return { @@ -278,7 +274,6 @@ def _process_single_prompt( "toolsets_used": selected_toolsets, "metadata": {"batch_num": batch_num, "timestamp": datetime.now().isoformat(), "model": config["model"]}, } - except Exception as e: print(f"❌ Error processing prompt {prompt_index}: {e}") if config.get("verbose"): @@ -300,12 +295,9 @@ def _process_batch_worker(args: Tuple) -> Dict[str, Any]: ``args`` is ``(batch_num, batch_data, output_dir, completed_prompts, config)``. """ batch_num, batch_data, output_dir, completed_prompts_set, config = args - output_dir = Path(output_dir) print(f"\nπŸ”„ Batch {batch_num}: Starting ({len(batch_data)} prompts)") - batch_output_file = output_dir / f"batch_{batch_num}.jsonl" - prompts_to_process = [ (idx, data) for idx, data in batch_data if idx not in completed_prompts_set @@ -314,9 +306,7 @@ def _process_batch_worker(args: Tuple) -> Dict[str, Any]: if not prompts_to_process: print(f"βœ… Batch {batch_num}: Already completed (skipping)") return {"batch_num": batch_num, "processed": 0, "skipped": len(batch_data), "tool_stats": {}, "completed_prompts": []} - print(f" Processing {len(prompts_to_process)} prompts (skipping {len(batch_data) - len(prompts_to_process)} already completed)") - batch_tool_stats = {} batch_reasoning_stats = dict.fromkeys(_REASONING_KEYS, 0) completed_in_batch = [] @@ -348,7 +338,6 @@ def _process_batch_worker(args: Tuple) -> Dict[str, Any]: tool_name: stats.get("failure", 0) for tool_name, stats in raw_tool_stats.items() } - _append_jsonl(batch_output_file, { "prompt_index": prompt_index, "conversations": result["trajectory"], @@ -360,7 +349,6 @@ def _process_batch_worker(args: Tuple) -> Dict[str, Any]: "tool_stats": _normalize_tool_stats(raw_tool_stats), # {tool: {count, success, failure}} "tool_error_counts": _normalize_tool_error_counts(raw_error_counts) # {tool: failure_count} }) - _merge_tool_stats(batch_tool_stats, result.get("tool_stats", {})) _merge_reasoning_stats(batch_reasoning_stats, result.get("reasoning_stats", {})) @@ -371,7 +359,6 @@ def _process_batch_worker(args: Tuple) -> Dict[str, Any]: print(f" {status} Prompt {prompt_index} completed") else: print(f" ❌ Prompt {prompt_index} failed (will retry on resume)") - print(f"βœ… Batch {batch_num}: Completed ({len(prompts_to_process)} prompts processed)") return { @@ -457,20 +444,16 @@ class BatchRunner: if not validate_distribution(distribution): raise ValueError(f"Unknown distribution: {distribution}. Available: {list(list_distributions().keys())}") - self.output_dir = Path("data") / run_name self.output_dir.mkdir(parents=True, exist_ok=True) self.checkpoint_file = self.output_dir / "checkpoint.json" self.stats_file = self.output_dir / "statistics.json" - self.dataset = self._load_dataset() if self.max_samples and self.max_samples < len(self.dataset): full_count = len(self.dataset) self.dataset = self.dataset[:self.max_samples] print(f"βœ‚οΈ Truncated dataset from {full_count} to {self.max_samples} samples (--max_samples)") - self.batches = self._create_batches() - print("πŸ“Š Batch Runner Initialized") print(f" Dataset: {self.dataset_file} ({len(self.dataset)} prompts)") print(f" Batch size: {self.batch_size}") @@ -487,7 +470,6 @@ class BatchRunner: """Load JSONL entries that have a ``prompt`` field; skip blank/invalid lines.""" if not self.dataset_file.exists(): raise FileNotFoundError(f"Dataset file not found: {self.dataset_file}") - dataset = [] with open(self.dataset_file, 'r', encoding='utf-8') as f: for line_num, line in enumerate(f, 1): @@ -537,7 +519,6 @@ class BatchRunner: def _save_checkpoint(self, checkpoint_data: Dict[str, Any], lock: Optional[Lock] = None): """Atomically write *checkpoint_data* (stamped ``last_updated``), under *lock* if given.""" checkpoint_data["last_updated"] = datetime.now().isoformat() - from utils import atomic_json_write if lock: with lock: @@ -557,7 +538,6 @@ class BatchRunner: if not batch_files: return completed_prompts - print(f"πŸ“‚ Scanning {len(batch_files)} batch files for completed prompts...") for batch_file in batch_files: @@ -608,15 +588,12 @@ class BatchRunner: if not completed_prompt_texts: return True print(f" Found {len(completed_prompt_texts)} already-completed prompts by content matching") - filtered_entries, skipped_indices = self._filter_dataset_by_completed(completed_prompt_texts) if not filtered_entries: print("\nβœ… All prompts have already been processed!") return False - self.batches = _chunk(filtered_entries, self.batch_size) - _banner("πŸ“Š RESUME SUMMARY") print(f" Original dataset size: {len(self.dataset):,} prompts") print(f" Already completed: {len(skipped_indices):,} prompts") @@ -644,7 +621,6 @@ class BatchRunner: ) else: worker_api_key = self.api_key - config = {key: getattr(self, key) for key in _AGENT_PASSTHROUGH} config["api_key"] = worker_api_key for key in ("distribution", "model", "max_iterations", "verbose", "log_prefix_chars"): @@ -661,7 +637,6 @@ class BatchRunner: (batch_num, batch_data, str(self.output_dir), completed_prompts_set, config) for batch_num, batch_data in enumerate(self.batches) ] - print(f"βœ… Created {len(tasks)} batch tasks") print("πŸš€ Starting parallel batch processing...\n") @@ -697,7 +672,6 @@ class BatchRunner: checkpoint_data.setdefault('batch_stats', {})[str(batch_num)] = { key: result.get(key, 0) for key in ('processed', 'skipped', 'discarded_no_reasoning') } - checkpoint_data['completed_prompts'] = sorted(completed_prompts_set) self._save_checkpoint(checkpoint_data, lock=checkpoint_lock) except Exception as ckpt_err: @@ -725,12 +699,10 @@ class BatchRunner: """ combined_file = self.output_dir / "trajectories.jsonl" print(f"\nπŸ“¦ Combining ALL batch files into {combined_file.name}...") - total_entries = 0 filtered_entries = 0 tombstone_entries = 0 batch_files_found = 0 - all_batch_files = sorted(self.output_dir.glob("batch_*.jsonl")) with open(combined_file, 'w', encoding='utf-8') as outfile: @@ -747,7 +719,6 @@ class BatchRunner: if data.get("discarded"): tombstone_entries += 1 continue - tool_stats = data.get('tool_stats', {}) invalid_tools = [k for k in tool_stats if k not in ALL_POSSIBLE_TOOLS] @@ -756,7 +727,6 @@ class BatchRunner: invalid_preview = invalid_tools[0][:50] + "..." if len(invalid_tools[0]) > 50 else invalid_tools[0] print(f" ⚠️ Filtering corrupted entry (batch {batch_num}): invalid tool '{invalid_preview}'") continue - outfile.write(line) except json.JSONDecodeError: filtered_entries += 1 @@ -779,14 +749,12 @@ class BatchRunner: if total_tool_stats: sorted_tools = sorted(total_tool_stats.items(), key=lambda x: x[1]["count"], reverse=True) - print(f"{'Tool Name':<25} {'Count':<10} {'Success':<10} {'Failure':<10} {'Success Rate':<12}") print("-" * 70) for tool_name, stats in sorted_tools: print(f"{tool_name:<25} {stats['count']:<10} {stats['success']:<10} {stats['failure']:<10} {stats['success_rate']:.1f}%") else: print("No tool calls were made during this run.") - total_discarded = sum(r.get("discarded_no_reasoning", 0) for r in results) print("\n🧠 Reasoning Coverage:") @@ -804,7 +772,6 @@ class BatchRunner: print(" No assistant turns recorded.") if total_discarded > 0: print(f" 🚫 Samples discarded (zero reasoning): {total_discarded:,}") - print(f"\nπŸ’Ύ Results saved to: {self.output_dir}") print(" - Trajectories: trajectories.jsonl (combined)") print(" - Individual batches: batch_*.jsonl (for debugging)") @@ -822,19 +789,15 @@ class BatchRunner: checkpoint_data = self._load_checkpoint() if checkpoint_data.get("run_name") != self.run_name: checkpoint_data = self._empty_checkpoint() - config = self._worker_config() # Index tracking is secondary to content matching (backward compatibility). completed_prompts_set = set(checkpoint_data.get("completed_prompts", [])) - start_time = time.time() # Checkpoint writes happen in the parent process; keep a lock for safety. checkpoint_lock = Lock() - results = self._run_pool(config, checkpoint_data, completed_prompts_set, checkpoint_lock) - total_tool_stats = {} total_reasoning_stats = dict.fromkeys(_REASONING_KEYS, 0) for batch_result in results: @@ -852,9 +815,7 @@ class BatchRunner: total_calls = stats["success"] + stats["failure"] stats["success_rate"] = round(stats["success"] / total_calls * 100, 2) if total_calls > 0 else 0.0 stats["failure_rate"] = round(stats["failure"] / total_calls * 100, 2) if total_calls > 0 else 0.0 - kept, batch_files_found = self._combine_batch_files() - final_stats = { "run_name": self.run_name, "distribution": self.distribution, @@ -871,7 +832,6 @@ class BatchRunner: with open(self.stats_file, 'w', encoding='utf-8') as f: json.dump(final_stats, f, indent=2, ensure_ascii=False) - self._print_summary(results, total_tool_stats, total_reasoning_stats, kept, batch_files_found, start_time) @@ -956,10 +916,8 @@ def main( """ if list_distributions: from toolset_distributions import print_distribution_info - print("πŸ“Š Available Toolset Distributions") print("=" * 70) - all_dists = list_distributions() for dist_name in sorted(all_dists.keys()): print_distribution_info(dist_name) @@ -990,7 +948,6 @@ def main( raise SystemExit(1) reasoning_config = {"enabled": True, "effort": reasoning_effort} print(f"🧠 Reasoning effort: {reasoning_effort}") - prefill_messages = None if prefill_messages_file: try: @@ -1027,9 +984,7 @@ def main( prefill_messages=prefill_messages, max_samples=max_samples, ) - runner.run(resume=resume) - except Exception as e: print(f"\n❌ Fatal error: {e}") if verbose: diff --git a/hermes_constants.py b/hermes_constants.py index ab2cfbea75..d8eff1a11b 100644 --- a/hermes_constants.py +++ b/hermes_constants.py @@ -145,7 +145,6 @@ def get_default_hermes_root() -> Path: memo_native, memo_env, memo_result = _default_hermes_root_memo if memo_native == str(native_home) and memo_env == env_home: return memo_result - result = native_home if env_home: env_path = Path(env_home) @@ -496,7 +495,6 @@ def _heal_managed_node_windows(home: Path | None = None) -> bool | None: node_arch = {"amd64": "x64", "x86_64": "x64", "arm64": "arm64", "x86": "x86"}.get(arch) if node_arch is None: return False - home = home or get_hermes_home() target = home / "node" @@ -522,24 +520,20 @@ def _heal_managed_node_windows(home: Path | None = None) -> bool | None: return response.read() except OSError: return None - index_url = f"https://nodejs.org/dist/latest-v{_HERMES_NODE_TARGET_MAJOR}.x/" index_bytes = _fetch(index_url, 60) if index_bytes is None: return False - match = re.search( rf"node-v{_HERMES_NODE_TARGET_MAJOR}\.\d+\.\d+-win-{node_arch}\.zip", index_bytes.decode("utf-8", errors="replace"), ) if not match: return False - zip_name = match.group(0) zip_bytes = _fetch(f"{index_url}{zip_name}", 300) if zip_bytes is None: return False - token = uuid.uuid4().hex[:8] staged = home / f"node.new-{token}" backup = home / f"node.old-{token}" @@ -559,7 +553,6 @@ def _heal_managed_node_windows(home: Path | None = None) -> bool | None: shutil.move(str(extracted), str(staged)) except OSError: return False - had_live = target.exists() if had_live: try: @@ -703,11 +696,9 @@ def find_node_executable_on_path(command: str) -> str | None: """ if sys.platform != "win32": return shutil.which(command) - command_str = str(command) if any(sep and sep in command_str for sep in (os.sep, os.altsep, "/", "\\")): return command_str if Path(command_str).is_file() else None - directories = [d for d in os.environ.get("PATH", "").split(os.pathsep) if d] for name in _candidate_node_command_names(command_str): for directory in directories: @@ -919,7 +910,6 @@ def get_subprocess_home(env: dict[str, str] | None = None) -> str | None: if mode == "profile": return profile_home - real_home = get_real_home(env) current_home = _env_get(env, "HOME") repaired = real_home if _norm_home_path(real_home) != _norm_home_path(current_home) else None @@ -988,7 +978,6 @@ def _canonical_model_variants(model: str) -> list[str]: """Add s plus its dots↔dashes and version-dot derivatives.""" dashed, dotted = s.replace('.', '-'), s.replace('-', '.') _add(s, dashed, dotted, _dash_to_dot(s), _dot_to_dash(s), _dash_to_dot(dashed), _dot_to_dash(dotted)) - _add_with_derivatives(model) parts = model.split('/') if len(parts) >= 2: # bare model (strip provider/aggregator prefix) @@ -1039,7 +1028,6 @@ def resolve_reasoning_config(cfg: dict | None, model: str = "") -> dict | None: if isinstance(model_cfg, dict): model_cfg = model_cfg.get("default") or model_cfg.get("model") or "" model = model_cfg.strip() if isinstance(model_cfg, str) else "" - overrides = agent_cfg.get("reasoning_overrides") or {} per_model = resolve_per_model_reasoning_effort(model, overrides) if per_model is not None: @@ -1185,7 +1173,6 @@ def apply_ipv4_preference(force: bool = False) -> None: if getattr(socket.getaddrinfo, "_hermes_ipv4_patched", False): return - _original_getaddrinfo = socket.getaddrinfo def _ipv4_getaddrinfo(host, port, family=0, type=0, proto=0, flags=0): @@ -1197,7 +1184,6 @@ def apply_ipv4_preference(force: bool = False) -> None: except socket.gaierror: # no A record β€” pure-IPv6 host return _original_getaddrinfo(host, port, family, type, proto, flags) return _original_getaddrinfo(host, port, family, type, proto, flags) - _ipv4_getaddrinfo._hermes_ipv4_patched = True # type: ignore[attr-defined] socket.getaddrinfo = _ipv4_getaddrinfo # type: ignore[assignment] diff --git a/hermes_logging.py b/hermes_logging.py index 2dfa09ad07..49990ff4ec 100644 --- a/hermes_logging.py +++ b/hermes_logging.py @@ -128,7 +128,6 @@ def _install_session_record_factory() -> None: except Exception: record.hermes_home = "" # type: ignore[attr-defined] return record - _session_record_factory._hermes_session_injector = True # type: ignore[attr-defined] logging.setLogRecordFactory(_session_record_factory) @@ -180,9 +179,7 @@ def setup_logging( global _logging_initialized home = hermes_home or get_hermes_home() log_dir = mkdir_under_hermes_home(home / "logs") - cfg_level, cfg_max_size, cfg_backup = _read_logging_config() - level_name = (log_level or cfg_level or "INFO").upper() level = getattr(logging, level_name, logging.INFO) max_bytes = (max_size_mb or cfg_max_size or 5) * 1024 * 1024 @@ -215,9 +212,7 @@ def setup_logging( # Root level must be low enough for the handlers to fire. if root.level == logging.NOTSET or root.level > level: root.setLevel(level) - _quiet_noisy_loggers() - _logging_initialized = True return log_dir @@ -230,7 +225,6 @@ def setup_verbose_logging() -> None: if any(getattr(h, "_hermes_verbose", False) for h in root.handlers): return - handler = logging.StreamHandler(_safe_stderr()) handler.setLevel(logging.DEBUG) handler.setFormatter(RedactingFormatter(_LOG_FORMAT_VERBOSE, datefmt="%H:%M:%S")) @@ -239,7 +233,6 @@ def setup_verbose_logging() -> None: if root.level > logging.DEBUG: root.setLevel(logging.DEBUG) - _quiet_noisy_loggers() # rex-deploy at INFO for sandbox status. logging.getLogger("rex-deploy").setLevel(logging.INFO) @@ -545,12 +538,10 @@ def enable_profile_log_routing(profile_homes: Sequence[str | Path]) -> bool: return False if any(isinstance(h, _ProfileRoutingFileHandler) for h in _queued_file_handlers): return True - listener = _queue_listener if listener is not None: listener.stop() _queue_listener = None - replacement = [] for existing in _queued_file_handlers: if isinstance(existing, RotatingFileHandler): @@ -558,7 +549,6 @@ def enable_profile_log_routing(profile_homes: Sequence[str | Path]) -> bool: _quietly(existing.close) else: replacement.append(existing) - _queued_file_handlers[:] = replacement if listener is not None: _start_queue_listener_locked() @@ -598,7 +588,6 @@ def _add_rotating_handler( and Path(getattr(existing, "baseFilename", "")).resolve() == resolved ): return - handler = _new_file_handler( path, level=level, max_bytes=max_bytes, backup_count=backup_count, formatter=formatter, )