guptaaryan16/observability_env_test
0
1import json2import os3import re4import subprocess5from datetime import datetime, timezone6from typing import Any, Dict, List, Tuple7 8from openai import OpenAI9 10from observer_env.main import Action, TracingEnvironment11 12 13API_BASE_URL = os.getenv("API_BASE_URL", "https://router.huggingface.co/v1")14MODEL_NAME = os.getenv("MODEL_NAME", "openai/gpt-oss-120b:free")15FALLBACK_CHAT_MODEL = os.getenv("FALLBACK_CHAT_MODEL", "google/gemma-4-31B-it:novita")16HF_TOKEN = os.getenv("HF_TOKEN")17BENCHMARK_NAME = os.getenv("BENCHMARK_NAME", "observability-rca")18TASK_INDEX = int(os.getenv("TASK_INDEX", "2"))19MAX_STEPS = int(os.getenv("MAX_STEPS", "8"))20WORKSPACE_DIR = os.getenv("WORKSPACE_DIR", "workspace")21DATASET_MODE = os.getenv("DATASET_MODE", "baseline")22REAL_CLOSE_ROOT = os.getenv("REAL_CLOSE_ROOT", "real_example/real_close_dataset")23REAL_CLOSE_REPO = os.getenv("REAL_CLOSE_REPO", "real_example/microservices-demo")24LLM_TIMEOUT_SECONDS = float(os.getenv("LLM_TIMEOUT_SECONDS", "30"))25DEBUG_LLM = os.getenv("DEBUG_LLM", "0") == "1"26DEBUG_LLM_FIRST_ONLY = os.getenv("DEBUG_LLM_FIRST_ONLY", "1") == "1"27DEBUG_LLM_MAX_CHARS = int(os.getenv("DEBUG_LLM_MAX_CHARS", "1600"))28DEBUG_LLM_LOG_PATH = os.getenv("DEBUG_LLM_LOG_PATH", "")29RUN_ARTIFACT_PATH = os.getenv("RUN_ARTIFACT_PATH", "")30PROMPT_CONTEXT_MAX_CHARS = int(os.getenv("PROMPT_CONTEXT_MAX_CHARS", "5000"))31 32 33# OpenAI client is initialised lazily in run_episode() so that importing34# this module (e.g. during `openenv validate`) does not crash when HF_TOKEN35# is absent.36client: OpenAI | None = None37ACTIVE_MODEL_NAME = MODEL_NAME38_DEBUG_LLM_TARGET_STEP: int | None = None39_RUN_STATS: Dict[str, int] = {40 "auto_finish_used": 0,41 "recovery_overrides": 0,42 "invalid_action_recovered": 0,43 "cache_actions": 0,44 "cache_hits": 0,45 "cache_misses": 0,46}47 48 49def _preview(value: Any, max_chars: int = DEBUG_LLM_MAX_CHARS) -> str:50 text = value if isinstance(value, str) else json.dumps(value, ensure_ascii=False)51 if len(text) <= max_chars:52 return text53 return text[:max_chars] + "...<truncated>"54 55 56def _debug_step_enabled(step_num: int) -> bool:57 global _DEBUG_LLM_TARGET_STEP58 if not DEBUG_LLM:59 return False60 if not DEBUG_LLM_FIRST_ONLY:61 return True62 if _DEBUG_LLM_TARGET_STEP is None:63 _DEBUG_LLM_TARGET_STEP = step_num64 return step_num == _DEBUG_LLM_TARGET_STEP65 66 67def _debug_emit(step_num: int, event: str, payload: Any) -> None:68 if not _debug_step_enabled(step_num):69 return70 preview = _preview(payload)71 print(f"[DEBUG] step={step_num} event={event} payload={preview}")72 if DEBUG_LLM_LOG_PATH:73 row = {"step": step_num, "event": event, "payload": payload}74 with open(DEBUG_LLM_LOG_PATH, "a", encoding="utf-8") as f:75 f.write(json.dumps(row, ensure_ascii=False) + "\n")76 77 78def _normalize_real_close_tasks(task_mapping: Dict[str, Any]) -> List[Dict[str, Any]]:79 normalized: List[Dict[str, Any]] = []80 for task in task_mapping.get("tasks", []):81 task_id = str(task.get("id", ""))82 expected = task.get("expected", {})83 84 if "checkout" in task_id:85 normalized.append(86 {87 "id": "task_1_syntax_trace",88 "difficulty": "Easy",89 "description": (90 "Customers are reporting checkout failures. Inspect logs and source code "91 "to locate ERR_CART_OVERLOAD and identify exact file and line."92 ),93 "workflow": [94 "Read checkout logs",95 "Search code for ERR_CART_OVERLOAD",96 "Return buggy file and line",97 ],98 "expected_answer": {99 "buggy_file": expected.get("buggy_file"),100 "line_number": expected.get("line_number"),101 "explanation": "checkout fails with ERR_CART_OVERLOAD due to injected overload condition",102 },103 "evidence_signals": {104 "must": ["ERR_CART_OVERLOAD", "checkoutservice", "main.go"],105 "helpful": ["PlaceOrder", "ResourceExhausted"],106 },107 "branch": expected.get("branch"),108 "commit": expected.get("commit"),109 }110 )111 elif "email" in task_id:112 normalized.append(113 {114 "id": "task_2_config_mismatch",115 "difficulty": "Medium",116 "description": (117 "Emailservice startup fails. Find missing env var from logs and infer expected "118 "default from source context."119 ),120 "workflow": [121 "Read email startup log",122 "Find missing variable",123 "Inspect source for expected port behavior",124 ],125 "expected_answer": {126 "missing_var": "SMTP_PORT",127 "default_value": "1025",128 "explanation": "emailservice crashes at startup because SMTP_PORT is missing",129 },130 "evidence_signals": {131 "must": ["SMTP_PORT", "KeyError", "email_server.py"],132 "helpful": ["os.environ", "startup"],133 },134 "branch": expected.get("branch"),135 "commit": expected.get("commit"),136 }137 )138 elif "shipping" in task_id:139 normalized.append(140 {141 "id": "task_3_silent_logical_bug",142 "difficulty": "Hard",143 "description": (144 "Shipping quote is unexpectedly low. Trace logs and source to identify the "145 "mathematical regression."146 ),147 "workflow": [148 "Read shipping logs",149 "Locate quote logic in source",150 "Explain quote regression",151 ],152 "expected_answer": {153 "buggy_file": expected.get("buggy_file"),154 "bug_function": "CreateQuoteFromCount",155 "explanation": "shipping quote changed from 8.99 to 0.89 causing undercharge",156 },157 "evidence_signals": {158 "must": ["shippingservice", "0.89", "8.99", "quote"],159 "helpful": ["CreateQuoteFromCount", "TestGetQuote"],160 },161 "branch": expected.get("branch"),162 "commit": expected.get("commit"),163 }164 )165 166 if len(normalized) < 3:167 raise ValueError("real_close task mapping must contain at least 3 tasks")168 return normalized169 170 171def _prepare_real_close_workspace() -> Tuple[str, str, str]:172 root = os.path.abspath(REAL_CLOSE_ROOT)173 repo = os.path.abspath(REAL_CLOSE_REPO)174 mapping_path = os.path.join(root, "metadata", "task_mapping.json")175 176 with open(mapping_path, "r", encoding="utf-8") as f:177 mapping = json.load(f)178 179 tasks = _normalize_real_close_tasks(mapping)180 181 if TASK_INDEX < 0 or TASK_INDEX >= len(tasks):182 raise ValueError(f"TASK_INDEX out of bounds for real_close mode: {TASK_INDEX}")183 184 selected = tasks[TASK_INDEX]185 selected_branch = selected.get("branch")186 selected_commit = selected.get("commit")187 188 # Prefer exact commit pinning for deterministic bug context, then fall back to branch.189 if selected_commit:190 try:191 subprocess.run(192 ["git", "-C", repo, "checkout", selected_commit],193 check=True,194 capture_output=True,195 text=True,196 )197 except subprocess.CalledProcessError:198 if selected_branch:199 subprocess.run(200 ["git", "-C", repo, "checkout", selected_branch],201 check=True,202 capture_output=True,203 text=True,204 )205 elif selected_branch:206 subprocess.run(207 ["git", "-C", repo, "checkout", selected_branch],208 check=True,209 capture_output=True,210 text=True,211 )212 213 logs_src = os.path.join(root, "logs")214 logs_dst = os.path.join(repo, "logs")215 os.makedirs(logs_dst, exist_ok=True)216 for name in os.listdir(logs_src):217 src = os.path.join(logs_src, name)218 dst = os.path.join(logs_dst, name)219 if os.path.isfile(src):220 with open(src, "r", encoding="utf-8") as fsrc:221 content = fsrc.read()222 with open(dst, "w", encoding="utf-8") as fdst:223 fdst.write(content)224 225 resolved_dataset_path = os.path.abspath(".resolved_dataset_real_close.json")226 with open(resolved_dataset_path, "w", encoding="utf-8") as f:227 json.dump(tasks, f, indent=2)228 229 return repo, resolved_dataset_path, selected.get("id", "real_close_task")230 231 232def _resolve_runtime_inputs() -> Tuple[str, str]:233 mode = DATASET_MODE.strip().lower()234 if mode == "real_close":235 workspace_dir, dataset_path, _ = _prepare_real_close_workspace()236 return workspace_dir, dataset_path237 238 workspace_dir = WORKSPACE_DIR if os.path.isdir(WORKSPACE_DIR) else "."239 return workspace_dir, "dataset.json"240 241 242def _format_bool(value: bool) -> str:243 return "true" if value else "false"244 245 246def _format_reward(value: float) -> str:247 return f"{value:.2f}"248 249 250def _action_to_string(action_dict: Dict[str, Any]) -> str:251 if "command" in action_dict:252 return f"command({action_dict['command']})"253 if "read_code" in action_dict:254 return f"read_code({action_dict['read_code']})"255 if "get_cached_result" in action_dict:256 compact = json.dumps(action_dict["get_cached_result"], separators=(",", ":"))257 return f"get_cached_result({compact})"258 if "submit_rca" in action_dict:259 compact = json.dumps(action_dict["submit_rca"], separators=(",", ":"))260 return f"submit_rca({compact})"261 return "noop"262 263 264def _build_prompt(task_description: str, last_output: str, step_num: int) -> str:265 workspace_hints = (266 "Workspace hint: avoid assuming fixed paths. First discover paths with 'find . -maxdepth 3 -type d'. "267 "If logs/src are missing, inspect available directories and adapt."268 )269 return (270 "You are solving a log-to-code RCA task. "271 "Return JSON with exactly one key among command, read_code, get_cached_result, submit_rca. "272 "Use get_cached_result when previous command output likely already contains the needed evidence. "273 "Prefer short commands with grep/find/cat/read_code and submit when confident. "274 "Do NOT place shell commands inside read_code; read_code must be a file path or file:start-end.\n"275 f"{workspace_hints}\n"276 f"Step: {step_num}\n"277 f"Task: {task_description}\n"278 f"Last output:\n{last_output}\n"279 "JSON only."280 )281 282 283def _last_n(items: List[Any], n: int) -> List[Any]:284 if n <= 0:285 return []286 return items[-n:] if len(items) > n else list(items)287 288 289def _truncate_for_prompt(text: str, max_chars: int = PROMPT_CONTEXT_MAX_CHARS) -> str:290 if len(text) <= max_chars:291 return text292 return text[:max_chars] + "\n...<context truncated>"293 294 295def _build_memory_context(296 state_snapshot: Dict[str, Any],297 workspace_paths: Dict[str, str],298) -> str:299 investigation = state_snapshot.get("investigation", {}) if isinstance(state_snapshot, dict) else {}300 reviewed_files = investigation.get("reviewed_files", []) if isinstance(investigation, dict) else []301 reviewed_logs = investigation.get("reviewed_logs", []) if isinstance(investigation, dict) else []302 evidence_timeline = investigation.get("evidence_timeline", []) if isinstance(investigation, dict) else []303 command_history = investigation.get("command_history", []) if isinstance(investigation, dict) else []304 cache_history = investigation.get("cache_read_history", []) if isinstance(investigation, dict) else []305 command_cache_index = investigation.get("command_cache_index", []) if isinstance(investigation, dict) else []306 file_history = investigation.get("file_read_history", []) if isinstance(investigation, dict) else []307 metrics = investigation.get("metrics", {}) if isinstance(investigation, dict) else {}308 discovered_signals = state_snapshot.get("discovered_signals", []) if isinstance(state_snapshot, dict) else []309 310 recent_commands = _last_n(command_history, 3)311 recent_file_reads = _last_n(file_history, 3)312 recent_cache_reads = _last_n(cache_history, 3)313 recent_cache_index = _last_n(command_cache_index, 5)314 recent_evidence = _last_n(evidence_timeline, 8)315 316 context_obj = {317 "known_workspace_paths": workspace_paths,318 "state": {319 "task_id": state_snapshot.get("task_id") if isinstance(state_snapshot, dict) else None,320 "step_count": state_snapshot.get("step_count") if isinstance(state_snapshot, dict) else None,321 "done": state_snapshot.get("done") if isinstance(state_snapshot, dict) else None,322 },323 "memory": {324 "reviewed_logs": reviewed_logs,325 "reviewed_files": reviewed_files,326 "discovered_signals": discovered_signals,327 "metrics": metrics,328 "recent_evidence": recent_evidence,329 "recent_commands": [330 {331 "step": item.get("step"),332 "command": item.get("command"),333 "return_code": item.get("return_code"),334 "error": item.get("error"),335 }336 for item in recent_commands337 ],338 "recent_file_reads": [339 {340 "step": item.get("step"),341 "filepath": item.get("filepath"),342 "start_line": item.get("start_line"),343 "end_line": item.get("end_line"),344 "error": item.get("error"),345 }346 for item in recent_file_reads347 ],348 "recent_cache_reads": [349 {350 "step": item.get("step"),351 "cache_hit": item.get("cache_hit"),352 "resolved_keys": item.get("resolved_keys"),353 "error": item.get("error"),354 }355 for item in recent_cache_reads356 ],357 "command_cache_index": [358 {359 "key": item.get("key"),360 "step": item.get("step"),361 "command": item.get("command"),362 "output_preview": item.get("output_preview"),363 }364 for item in recent_cache_index365 ],366 },367 }368 return _truncate_for_prompt(json.dumps(context_obj, ensure_ascii=False, indent=2))369 370 371def _build_prompt_with_memory(372 task_description: str,373 last_output: str,374 step_num: int,375 state_snapshot: Dict[str, Any],376 workspace_paths: Dict[str, str],377) -> str:378 base = _build_prompt(task_description, last_output, step_num)379 memory_context = _build_memory_context(state_snapshot, workspace_paths)380 return (381 f"{base}\n\n"382 "Persistent investigation memory (use this to remember exact paths/signals across steps):\n"383 f"{memory_context}\n\n"384 "When possible, prefer get_cached_result over rerunning duplicate commands; prioritize exact file paths discovered in memory and submit_rca once bug file + line are confirmed."385 )386 387 388def _extract_json_object(content: str) -> Dict[str, Any]:389 text = (content or "").strip()390 if not text:391 raise ValueError("Empty LLM response")392 393 if text.startswith("```"):394 text = re.sub(r"^```(?:json)?", "", text).strip()395 text = re.sub(r"```$", "", text).strip()396 397 try:398 parsed = json.loads(text)399 if isinstance(parsed, dict):400 return parsed401 except Exception:402 pass403 404 start = text.find("{")405 end = text.rfind("}")406 if start == -1 or end == -1 or end <= start:407 raise ValueError("No JSON object found in response")408 409 parsed = json.loads(text[start : end + 1])410 if not isinstance(parsed, dict):411 raise ValueError("Parsed payload is not an object")412 return parsed413 414 415def _existing_dir(base: str, candidate: str) -> str:416 path = os.path.join(base, candidate)417 if os.path.isdir(path):418 return candidate419 return ""420 421 422def _discover_workspace_paths(workspace_dir: str) -> Dict[str, str]:423 logs_dir = _existing_dir(workspace_dir, "logs")424 src_dir = _existing_dir(workspace_dir, "src")425 426 if not logs_dir:427 for cand in ["workspace/logs", "real_example/real_close_dataset/logs"]:428 if _existing_dir(workspace_dir, cand):429 logs_dir = cand430 break431 432 if not src_dir:433 for cand in ["workspace/src", "env", "real_example/microservices-demo/src"]:434 if _existing_dir(workspace_dir, cand):435 src_dir = cand436 break437 438 return {439 "logs_dir": logs_dir,440 "src_dir": src_dir,441 }442 443 444def _detect_tool_capabilities(workspace_dir: str) -> Dict[str, bool]:445 try:446 probe = subprocess.run(447 "command -v rg >/dev/null 2>&1",448 shell=True,449 cwd=workspace_dir,450 capture_output=True,451 text=True,452 timeout=5,453 )454 has_rg = probe.returncode == 0455 except Exception:456 has_rg = False457 return {"has_rg": has_rg}458 459 460def _portable_search_command(targets: List[str]) -> str:461 cleaned = [t for t in targets if t]462 target_str = " ".join(cleaned) if cleaned else "."463 pattern = "ERR|error|Exception|KeyError|Quote|shipping|checkout|SMTP_PORT"464 return (465 "if command -v rg >/dev/null 2>&1; then "466 f"rg -n -S '{pattern}' {target_str}; "467 "else "468 f"grep -RInE '{pattern}' {target_str}; "469 "fi"470 )471 472 473def _looks_like_shell_command(text: str) -> bool:474 lowered = text.lower()475 shell_markers = ["|", "&&", ";", "find ", "grep ", "rg ", "xargs ", "2>/dev/null"]476 return any(marker in lowered for marker in shell_markers)477 478 479def _coerce_action_payload(480 payload: Dict[str, Any],481 workspace_paths: Dict[str, str],482 capabilities: Dict[str, bool],483 step_num: int,484 task_id: str,485 last_output: str,486 state_snapshot: Dict[str, Any],487) -> Dict[str, Any]:488 if not isinstance(payload, dict):489 return _exploration_action(workspace_paths, step_num)490 491 if "get_cached_result" in payload:492 cache_req = payload["get_cached_result"]493 if isinstance(cache_req, str):494 payload = dict(payload)495 payload["get_cached_result"] = {"query": cache_req}496 elif cache_req is None:497 payload = dict(payload)498 payload["get_cached_result"] = {}499 elif not isinstance(cache_req, dict):500 return _exploration_action(workspace_paths, step_num)501 502 if "cache_query" in payload and "get_cached_result" not in payload:503 payload = {"get_cached_result": {"query": str(payload.get("cache_query", "")).strip()}}504 505 if "cmd" in payload and isinstance(payload["cmd"], list):506 cmd_parts = [str(p) for p in payload["cmd"] if str(p).strip()]507 if cmd_parts:508 return {"command": " ".join(cmd_parts)}509 510 if not any(k in payload for k in ["command", "read_code", "get_cached_result", "submit_rca"]):511 rca_like_keys = {"file", "buggy_file", "line", "line_number", "explanation", "bug_function", "function", "function_name"}512 if any(k in payload for k in rca_like_keys):513 payload = {"submit_rca": payload}514 515 if "read_code" in payload and isinstance(payload["read_code"], str):516 rc = payload["read_code"].strip()517 if _looks_like_shell_command(rc):518 return {"command": rc}519 if rc == "" or rc.endswith("/"):520 return _exploration_action(workspace_paths, step_num)521 522 if "command" in payload and isinstance(payload["command"], str):523 cmd = payload["command"].strip()524 if cmd == "":525 return _exploration_action(workspace_paths, step_num)526 if ("rg " in cmd or "xargs rg" in cmd) and not capabilities.get("has_rg", False):527 return {"command": _portable_search_command([workspace_paths.get("logs_dir", ""), workspace_paths.get("src_dir", "")])}528 payload = dict(payload)529 payload["command"] = cmd530 531 if "submit_rca" in payload:532 submit_raw = payload["submit_rca"]533 if isinstance(submit_raw, str):534 submit_obj: Dict[str, Any] = {"explanation": submit_raw}535 elif isinstance(submit_raw, dict):536 submit_obj = dict(submit_raw)537 else:538 return _exploration_action(workspace_paths, step_num)539 540 explanation = str(submit_obj.get("explanation", ""))541 542 if task_id == "task_1_syntax_trace":543 buggy_file = submit_obj.get("buggy_file") or submit_obj.get("file") or _extract_candidate_bug_file(last_output, "src/checkoutservice/main.go")544 line_number = submit_obj.get("line_number") or submit_obj.get("line")545 if line_number is None:546 file_line = _extract_file_and_line_from_output(last_output, str(buggy_file))547 if file_line is not None:548 _, line_number = file_line549 payload["submit_rca"] = {550 "buggy_file": str(buggy_file),551 "line_number": int(line_number) if str(line_number).isdigit() else submit_obj.get("line_number", 0),552 "explanation": explanation,553 }554 555 elif task_id == "task_2_config_mismatch":556 missing_var = submit_obj.get("missing_var") or submit_obj.get("var") or submit_obj.get("env_var")557 default_value = submit_obj.get("default_value") or submit_obj.get("default")558 payload["submit_rca"] = {559 "missing_var": str(missing_var or "SMTP_PORT"),560 "default_value": str(default_value or "1025"),561 "explanation": explanation,562 }563 564 elif task_id == "task_3_silent_logical_bug":565 buggy_file = submit_obj.get("buggy_file") or submit_obj.get("file") or _extract_candidate_bug_file(last_output, "src/shippingservice/quote.go")566 bug_function = submit_obj.get("bug_function") or submit_obj.get("function") or submit_obj.get("function_name")567 exp_l = explanation.lower()568 if not bug_function:569 if "calculatequote" in exp_l:570 bug_function = "CalculateQuote"571 elif "createquotefromcount" in exp_l or "quote.go" in str(buggy_file).lower():572 bug_function = "CreateQuoteFromCount"573 else:574 bug_function = "CreateQuoteFromCount"575 payload["submit_rca"] = {576 "buggy_file": str(buggy_file),577 "bug_function": str(bug_function),578 "explanation": explanation,579 }580 581 return payload582 583 584def _extract_file_and_line_from_output(output: str, filepath_hint: str = "") -> tuple[str, int] | None:585 if not output:586 return None587 patterns = []588 if filepath_hint:589 escaped = re.escape(filepath_hint)590 patterns.append(rf"((?:\./)?{escaped}):(\d+):")591 patterns.append(r"([\w./-]+\.(?:go|py|js|ts|java)):(\d+):")592 593 # For safety, only accept line numbers tied to concrete source file paths.594 for pattern in patterns:595 match = re.search(pattern, output)596 if match:597 try:598 file_path = match.group(1)599 line = int(match.group(2))600 return file_path.lstrip("./"), line601 except Exception:602 continue603 return None604 605 606def _extract_candidate_bug_file(output: str, fallback: str = "src/checkoutservice/main.go") -> str:607 if not output:608 return fallback609 match = re.search(r"([\w./-]+\.(?:go|py|js|ts|java))", output)610 if match:611 return match.group(1)612 return fallback613 614 615def _auto_finish_action(616 task_id: str,617 state_snapshot: Dict[str, Any],618 last_output: str,619) -> Dict[str, Any] | None:620 if task_id != "task_1_syntax_trace":621 return None622 623 discovered_signals = set(str(v) for v in state_snapshot.get("discovered_signals", []))624 investigation = state_snapshot.get("investigation", {}) if isinstance(state_snapshot, dict) else {}625 metrics = investigation.get("metrics", {}) if isinstance(investigation, dict) else {}626 must_signals_discovered = int(metrics.get("must_signals_discovered", 0) or 0)627 628 has_core_evidence = (629 "ERR_CART_OVERLOAD" in discovered_signals630 and ("checkoutservice" in discovered_signals or "main.go" in discovered_signals)631 )632 if not has_core_evidence and must_signals_discovered < 2:633 return None634 635 candidate_file = _extract_candidate_bug_file(last_output)636 file_line = _extract_file_and_line_from_output(last_output, candidate_file)637 if file_line is not None:638 final_file, line_hint = file_line639 return {640 "submit_rca": {641 "buggy_file": final_file,642 "line_number": line_hint,643 "explanation": "checkout fails with ERR_CART_OVERLOAD due to injected overload condition",644 }645 }646 647 return {"command": "grep -RIn 'ERR_CART_OVERLOAD' src | head -n 10"}648 649 650def _exploration_action(workspace_paths: Dict[str, str], step_num: int = 1) -> Dict[str, Any]:651 logs_dir = workspace_paths.get("logs_dir", "")652 src_dir = workspace_paths.get("src_dir", "")653 654 phase = ((step_num - 1) % 3) + 1655 if phase == 1:656 if logs_dir and src_dir:657 return {"command": f"ls -la {logs_dir} && ls -la {src_dir}"}658 if logs_dir:659 return {"command": f"ls -la {logs_dir} && find . -maxdepth 3 -type d"}660 if src_dir:661 return {"command": f"ls -la {src_dir} && find . -maxdepth 3 -type d"}662 return {"command": "find . -maxdepth 3 -type d"}663 664 if phase == 2:665 return {"command": _portable_search_command([logs_dir, src_dir])}666 667 if src_dir:668 return {"command": f"find {src_dir} -maxdepth 4 -type f | head -n 80"}669 return {"command": "find . -maxdepth 4 -type f | head -n 80"}670 671 672def _recovery_action(673 workspace_paths: Dict[str, str],674 failure_streak: int,675 step_num: int,676) -> Dict[str, Any]:677 logs_dir = workspace_paths.get("logs_dir", "")678 src_dir = workspace_paths.get("src_dir", "")679 680 if failure_streak <= 1:681 return _exploration_action(workspace_paths, step_num)682 683 return {"command": _portable_search_command([logs_dir, src_dir])}684 685 686def _llm_action(687 task_description: str,688 last_output: str,689 step_num: int,690 workspace_paths: Dict[str, str],691 state_snapshot: Dict[str, Any],692) -> Dict[str, Any]:693 global ACTIVE_MODEL_NAME694 prompt = _build_prompt_with_memory(695 task_description=task_description,696 last_output=last_output,697 step_num=step_num,698 state_snapshot=state_snapshot,699 workspace_paths=workspace_paths,700 )701 _debug_emit(step_num, "llm_prompt", prompt)702 try:703 response = client.chat.completions.create(704 model=ACTIVE_MODEL_NAME,705 messages=[{"role": "user", "content": prompt}],706 timeout=LLM_TIMEOUT_SECONDS,707 )708 content = (response.choices[0].message.content or "{}").strip()709 _debug_emit(step_num, "llm_raw_response", content)710 parsed = _extract_json_object(content)711 _debug_emit(step_num, "llm_parsed_action", parsed)712 return parsed713 except Exception as exc:714 msg = str(exc).lower()715 exc_name = type(exc).__name__.lower()716 if "not a chat model" in msg and ACTIVE_MODEL_NAME != FALLBACK_CHAT_MODEL:717 ACTIVE_MODEL_NAME = FALLBACK_CHAT_MODEL718 _debug_emit(step_num, "llm_model_fallback", {"active_model": ACTIVE_MODEL_NAME})719 response = client.chat.completions.create(720 model=ACTIVE_MODEL_NAME,721 messages=[{"role": "user", "content": prompt}],722 timeout=LLM_TIMEOUT_SECONDS,723 )724 content = (response.choices[0].message.content or "{}").strip()725 _debug_emit(step_num, "llm_raw_response", content)726 parsed = _extract_json_object(content)727 _debug_emit(step_num, "llm_parsed_action", parsed)728 return parsed729 if (730 "permission" in msg731 or "authentication" in msg732 or "invalid_api_key" in msg733 or "incorrect api key" in msg734 or "permissiondenied" in exc_name735 or "authenticationerror" in exc_name736 ):737 _debug_emit(step_num, "llm_auth_error", str(exc))738 raise RuntimeError(f"LLM auth/permission error: {exc}") from exc739 _debug_emit(step_num, "llm_non_auth_error", str(exc))740 return _exploration_action(workspace_paths, step_num)741 742 743def run_episode() -> Tuple[bool, int, List[float]]:744 global _RUN_STATS, client, ACTIVE_MODEL_NAME745 # Lazy OpenAI client init — deferred so module import works without HF_TOKEN.746 if client is None:747 token = HF_TOKEN748 if not token:749 raise RuntimeError(750 "HF_TOKEN environment variable is required to run inference. "751 "Set it before calling `python inference.py`."752 )753 client = OpenAI(base_url=API_BASE_URL, api_key=token)754 _RUN_STATS = {755 "auto_finish_used": 0,756 "recovery_overrides": 0,757 "invalid_action_recovered": 0,758 "cache_actions": 0,759 "cache_hits": 0,760 "cache_misses": 0,761 }762 workspace_dir, dataset_path = _resolve_runtime_inputs()763 workspace_paths = _discover_workspace_paths(workspace_dir)764 capabilities = _detect_tool_capabilities(workspace_dir)765 env = TracingEnvironment(workspace_dir=workspace_dir, dataset_path=dataset_path)766 obs = env.reset(task_index=TASK_INDEX)767 task_name = obs.task_id768 769 print(f"[START] task={task_name} env={BENCHMARK_NAME} model={ACTIVE_MODEL_NAME}")770 771 rewards: List[float] = []772 step_idx = 0773 done = False774 success = False775 prev_action_str = ""776 repeated_failure_streak = 0777 run_summary: Dict[str, Any] = {}778 779 try:780 while not done and step_idx < MAX_STEPS:781 step_idx += 1782 state_snapshot = env.state()783 784 auto_payload = _auto_finish_action(785 task_id=obs.task_id,786 state_snapshot=state_snapshot,787 last_output=obs.last_action_output,788 )789 if auto_payload is not None:790 action_payload = auto_payload791 _RUN_STATS["auto_finish_used"] += 1792 _debug_emit(step_idx, "auto_finish_action", action_payload)793 else:794 try:795 action_payload = _llm_action(796 obs.description,797 obs.last_action_output,798 step_idx,799 workspace_paths,800 state_snapshot,801 )802 except RuntimeError as llm_exc:803 error_str = str(llm_exc).replace("\n", " ")804 print(805 "[STEP] "806 f"step={step_idx} "807 "action=llm_unavailable "808 f"reward={_format_reward(0.0)} "809 "done=true "810 f"error={error_str}"811 )812 done = True813 break814 action_payload = _coerce_action_payload(815 action_payload,816 workspace_paths,817 capabilities,818 step_idx,819 obs.task_id,820 obs.last_action_output,821 state_snapshot,822 )823 if repeated_failure_streak >= 2:824 action_payload = _recovery_action(workspace_paths, repeated_failure_streak, step_idx)825 _RUN_STATS["recovery_overrides"] += 1826 _debug_emit(step_idx, "action_final", action_payload)827 action_str = _action_to_string(action_payload)828 829 try:830 action = Action(**action_payload)831 except Exception:832 action_payload = _recovery_action(workspace_paths, repeated_failure_streak, step_idx)833 _RUN_STATS["invalid_action_recovered"] += 1834 action = Action(**action_payload)835 action_str = _action_to_string(action_payload)836 837 obs, reward, done, info = env.step(action)838 rewards.append(reward.value)839 error = info.get("last_action_error")840 if "get_cached_result" in action_payload:841 _RUN_STATS["cache_actions"] += 1842 if bool(info.get("cache_hit", False)):843 _RUN_STATS["cache_hits"] += 1844 else:845 _RUN_STATS["cache_misses"] += 1846 847 if error and action_str == prev_action_str:848 repeated_failure_streak += 1849 elif error:850 repeated_failure_streak = 1851 else:852 repeated_failure_streak = 0853 prev_action_str = action_str854 855 error_str = "null" if error in (None, "") else str(error).replace("\n", " ")856 857 print(858 "[STEP] "859 f"step={step_idx} "860 f"action={action_str} "861 f"reward={_format_reward(reward.value)} "862 f"done={_format_bool(done)} "863 f"error={error_str}"864 )865 866 grading = (env.result or {}).get("grading", {})867 success = bool(grading.get("passed", False)) if done else False868 final_state = env.state()869 run_summary = {870 "timestamp_utc": datetime.now(timezone.utc).isoformat(),871 "benchmark_name": BENCHMARK_NAME,872 "model_name": ACTIVE_MODEL_NAME,873 "api_base_url": API_BASE_URL,874 "dataset_mode": DATASET_MODE,875 "workspace_dir": workspace_dir,876 "dataset_path": dataset_path,877 "task_index": TASK_INDEX,878 "task_id": task_name,879 "max_steps": MAX_STEPS,880 "steps_executed": step_idx,881 "success": success,882 "rewards": rewards,883 "reward_sum": round(sum(rewards), 4),884 "run_stats": dict(_RUN_STATS),885 "capabilities": capabilities,886 "workspace_paths": workspace_paths,887 "result": env.result,888 "final_state": final_state,889 }890 return success, step_idx, rewards891 finally:892 env.close()893 rewards_str = ",".join(_format_reward(v) for v in rewards)894 print(895 "[END] "896 f"success={_format_bool(success)} "897 f"steps={step_idx} "898 f"rewards={rewards_str}"899 )900 if RUN_ARTIFACT_PATH and run_summary:901 artifact_dir = os.path.dirname(RUN_ARTIFACT_PATH)902 if artifact_dir:903 os.makedirs(artifact_dir, exist_ok=True)904 with open(RUN_ARTIFACT_PATH, "w", encoding="utf-8") as f:905 json.dump(run_summary, f, ensure_ascii=False, indent=2)906 907 908if __name__ == "__main__":909 run_episode()910