async def analyze_lean_failure(
lean_code: str,
lean_executable_path: str,
cwd: str,
shared_lib_path: Optional[pathlib.Path],
timeout_seconds: int = DEFAULT_ANALYSIS_TIMEOUT,
fallback_error: str = "Build system analysis failed.",
) -> str:
"""
Analyzes failing Lean code using LSP for detailed diagnostics and goal states.
Orchestrates the process of preprocessing code, running an optional pre-check,
interacting with the Lean LSP server via LeanLspClient, collecting diagnostics
and goal states, and assembling an annotated code report. Annotations are
placed on separate lines before the corresponding numbered code line.
Args:
lean_code: The Lean code string to analyze.
lean_executable_path: Path to the 'lean' executable.
cwd: Working directory for the lean server process (temp project).
shared_lib_path: Path to the root of the shared library dependency, used
for setting LEAN_PATH for both pre-check and the LSP server.
timeout_seconds: Timeout for the entire analysis process, including
subprocess calls and individual LSP requests within the client.
fallback_error: Original error message (e.g., from `lake build`) to append
to the analysis output.
Returns:
Annotated Lean code string with goal states and LSP diagnostics on
separate lines, followed by numbered code lines, and the appended build
error message.
"""
# --- Pre-check [ ... No changes needed here ... ] ---
first_compiler_error_line = float("inf")
temp_precheck_filename = "precheck_analysis_file.lean"
temp_precheck_path = pathlib.Path(cwd) / temp_precheck_filename
logger.debug("Preprocessing Lean code for analysis...")
processed_code = _preprocess_lean_code(lean_code)
try:
# [ ... Rest of pre-check logic remains the same ... ]
logger.info(f"Pre-check: Writing processed code to {temp_precheck_path}")
temp_precheck_path.write_text(processed_code, encoding="utf-8")
logger.info(
f"Pre-check: Running: {lean_executable_path} {temp_precheck_filename}"
)
precheck_timeout = max(5, timeout_seconds // 3)
precheck_env = _get_lean_path_env(lean_executable_path, cwd, shared_lib_path)
result = await asyncio.to_thread(
subprocess.run,
[lean_executable_path, temp_precheck_filename],
capture_output=True,
text=True,
cwd=cwd,
timeout=precheck_timeout,
encoding="utf-8",
errors="replace",
env=precheck_env,
check=False,
)
error_pattern = re.compile(r".*?:(\d+):\d+:\s*error:")
output_to_scan = result.stdout + "\n" + result.stderr
output_lines = output_to_scan.splitlines()
logger.debug(
f"Pre-check stdout/stderr ({len(output_lines)} lines): "
f"{output_to_scan[:500]}..."
)
for line_num_str in (
match.group(1)
for match in (error_pattern.match(line) for line in output_lines)
if match
):
try:
first_compiler_error_line = int(line_num_str)
logger.info(
"Pre-check: Found first compiler error on line: "
f"{first_compiler_error_line} (in processed code)"
)
break
except ValueError:
logger.warning(
f"Pre-check: Failed to parse line number from match: {line_num_str}"
)
if first_compiler_error_line == float("inf"):
logger.info(
"Pre-check: No errors found matching pattern in compiler output."
)
except FileNotFoundError:
logger.error(
f"Pre-check Failed: Lean executable not found at '{lean_executable_path}'."
)
except subprocess.TimeoutExpired:
logger.warning(f"Pre-check: Timed out after {precheck_timeout}s.")
except Exception as precheck_e:
logger.warning(
f"Pre-check: Failed with unexpected error: {precheck_e}", exc_info=True
)
finally:
if temp_precheck_path.exists():
try:
temp_precheck_path.unlink()
logger.debug(f"Pre-check: Cleaned up {temp_precheck_path}")
except OSError as e:
logger.warning(
f"Pre-check: Could not delete temp file {temp_precheck_path}: {e}"
)
# --- End of Pre-check ---
# --- Initialize LSP Client [ ... No changes needed here ... ] ---
client_timeout = max(15, timeout_seconds // 2)
client = LeanLspClient(
lean_executable_path,
cwd,
timeout=client_timeout,
shared_lib_path=shared_lib_path,
)
# --- Analysis Variables [ ... No changes needed here ... ] ---
analysis_succeeded = False
collected_diagnostics: List[Dict[str, Any]] = []
diagnostics_by_line: Dict[int, List[str]] = defaultdict(list)
first_lsp_error_line_idx = float("inf")
effective_error_line_idx = float("inf")
# --- File URI [ ... No changes needed here ... ] ---
temp_filename = "temp_analysis_file.lean"
temp_file_path = pathlib.Path(cwd) / temp_filename
temp_file_uri = temp_file_path.as_uri()
# Wrap core LSP interaction in try/finally to ensure client close
final_output_lines: List[str] = []
code_lines_buffer: List[
str
] = [] # Define earlier for use in goal loop and finally block
try:
# --- Start LSP and Open Document [ ... No changes needed here ... ] ---
logger.info("Starting LSP client...")
await client.start_server()
await client.initialize()
code_lines_buffer = processed_code.splitlines() # Assign here
await client.did_open(temp_file_uri, "lean", 1, processed_code)
logger.info(f"Sent textDocument/didOpen for URI {temp_file_uri}")
# --- Wait for initial diagnostics [ ... No changes needed here ... ] ---
initial_wait_time = min(10.0, client_timeout / 2)
logger.info(f"Waiting {initial_wait_time:.1f}s for initial diagnostics...")
await asyncio.sleep(initial_wait_time)
initial_diagnostics = await client.get_diagnostics(timeout=1.0)
collected_diagnostics.extend(initial_diagnostics)
logger.info(f"Collected {len(initial_diagnostics)} initial diagnostics.")
# --- Determine effective error line [ ... No changes needed here ... ] ---
for diag in collected_diagnostics:
if diag.get("severity") == 1:
try:
line_idx = (
diag.get("range", {}).get("start", {}).get("line", float("inf"))
)
first_lsp_error_line_idx = min(
first_lsp_error_line_idx, float(line_idx)
)
except (ValueError, TypeError):
logger.warning(
f"Could not parse line number from diagnostic: {diag}"
)
effective_error_line_idx = min(
float(first_compiler_error_line) - 1, first_lsp_error_line_idx
)
if effective_error_line_idx != float("inf"):
effective_error_line_idx = int(effective_error_line_idx)
logger.info(
"Effective first error detected near line index: "
f"{effective_error_line_idx} (Line {effective_error_line_idx + 1} "
"in processed code)"
)
else:
logger.info("No errors detected from pre-check or initial LSP diagnostics.")
effective_error_line_idx = -1
# --- Goal fetching loop ---
logger.info(
"Starting line-by-line goal annotation (skipping blank/comment-only "
"lines)..." # Updated log message (kept within limit)
)
goal_lines_buffer: List[Optional[str]] = [] # Store cleaned goals or None
for i, line_content in enumerate(code_lines_buffer):
goal_to_append: Optional[str] = None # Default to adding nothing
# Check if the line is blank or consists only of a comment (and whitespace)
stripped_line = line_content.strip()
is_comment_or_blank = not stripped_line or stripped_line.startswith("--")
if is_comment_or_blank:
logger.debug(f"Skipping goal fetch for blank/comment line {i + 1}")
# Append None, as no goal should be displayed for this line
goal_lines_buffer.append(None)
continue # Move to the next line
# --- Goal fetching logic (only runs for non-comment/non-blank lines) ---
goal_result = None # Define here for scope
try:
# Attempt to get the goal state for the current line
goal_result = await client.get_goal(temp_file_uri, line=i, character=0)
if goal_result is not None:
parsed_goal = _parse_goal_result(goal_result)
# Log errors/warnings during parsing, but don't store them as goals
if parsed_goal.startswith("Error:"):
logger.warning(
"Goal parsing resulted in error for line "
f"{i + 1}: {parsed_goal}"
)
elif parsed_goal.startswith("Goal state (unknown format"):
logger.warning(
"Goal parsing resulted in unknown format for line "
f"{i + 1}: {parsed_goal}"
)
else:
# Use the cleaned, potentially multi-line goal string
cleaned_goal, _ = _clean_goal_string(parsed_goal)
# Store the cleaned goal if it's not just "goals accomplished"
# or empty, as those don't need explicit display.
if cleaned_goal and cleaned_goal != "goals accomplished":
goal_to_append = cleaned_goal
elif cleaned_goal == "goals accomplished":
logger.debug(
"Goals accomplished at line %d, not storing explicit "
"goal.",
i + 1,
)
else:
logger.debug(
"Empty cleaned goal after parsing for line %d, "
"skipping.",
i + 1,
)
# If goal_result was None initially, goal_to_append remains None.
except ConnectionError as goal_conn_e:
logger.error(
"Connection error during goal retrieval loop for line "
f"{i + 1}: {goal_conn_e}"
)
# goal_to_append remains None
except Exception as goal_e:
logger.error(
"Unexpected error during goal processing for line "
f"{i + 1}: {goal_e}",
exc_info=True,
)
# goal_to_append remains None
# Append None or the cleaned goal string to the buffer
goal_lines_buffer.append(goal_to_append)
# --- Collect final diagnostics and process [ ** MINOR CHANGE ** ] ---
# (The change is in how the output of _format_diagnostic_line is used later)
diagnostic_collection_timeout = min(5.0, client_timeout)
logger.info(
f"Collecting final diagnostics (timeout="
f"{diagnostic_collection_timeout:.1f}s)..."
)
more_diagnostics = await client.get_diagnostics(
timeout=diagnostic_collection_timeout
)
collected_diagnostics.extend(more_diagnostics)
logger.info(f"Total collected diagnostics: {len(collected_diagnostics)}")
# Process all collected diagnostics for inline reporting
if collected_diagnostics:
logger.debug(
f"Processing {len(collected_diagnostics)} total diagnostics..."
)
formatted_diags_count = 0
seen_diags = set()
for diag in collected_diagnostics:
# [ ... De-duplication logic remains the same ... ]
diag_sig = (
diag.get("severity"),
diag.get("range", {}).get("start", {}).get("line"),
diag.get("range", {}).get("start", {}).get("character"),
diag.get("range", {}).get("end", {}).get("line"),
diag.get("range", {}).get("end", {}).get("character"),
diag.get("message"),
)
if diag_sig in seen_diags:
continue
seen_diags.add(diag_sig)
start_line_idx = diag.get("range", {}).get("start", {}).get("line", -1)
if start_line_idx != -1 and start_line_idx < len(code_lines_buffer):
# Get the potentially multi-line formatted diagnostic string
diag_string = _format_diagnostic_line(diag, start_line_idx)
if diag_string is not None: # Only store Errors/Warnings
diagnostics_by_line[start_line_idx].append(diag_string)
formatted_diags_count += 1
else:
logger.warning(
"Diagnostic reported for invalid line index "
f"{start_line_idx}: {diag}"
)
logger.info(
f"Processed {formatted_diags_count} unique diagnostics into line map."
)
# --- Assemble final output [ ** MODIFIED ** ] ---
logger.info("Assembling final annotated output...")
num_lines = len(code_lines_buffer)
# Calculate padding width for line numbers
padding_width = len(str(num_lines)) if num_lines > 0 else 1
for i, code_line in enumerate(code_lines_buffer):
# Create prefix only for the code line number
line_num_str = f"{i + 1}"
line_prefix = f"{line_num_str:<{padding_width}} | "
# 1. Add Goal State (if any) on its own lines
goal_content = goal_lines_buffer[i]
if goal_content is not None:
# Add a separator/header before the goal content
final_output_lines.append("--- Goal ---")
final_output_lines.append(
goal_content
) # Append raw multi-line goal string
# 2. Add Diagnostics (if any) on their own lines
if i in diagnostics_by_line:
# Add separator only if there are diagnostics for this line
final_output_lines.append("--- Diagnostics ---")
for diag_string in diagnostics_by_line[i]:
# Append the formatted, potentially multi-line diagnostic string
final_output_lines.append(diag_string)
logger.debug(
f"Inserted {len(diagnostics_by_line[i])} diagnostics before "
f"line index {i}"
)
# 3. Add the processed code line with line number prefix
final_output_lines.append(line_prefix + code_line)
logger.info("Finished assembling annotated output.")
analysis_succeeded = True
# --- Exception Handling [ ... No changes needed here ... ] ---
except ConnectionError as e:
logger.error(f"LSP Connection Error during analysis: {e}")
final_output_lines.append(f"--- Error: LSP Connection Failed: {e} ---")
analysis_succeeded = False
except asyncio.TimeoutError as e:
logger.error(f"LSP Overall Timeout Error during analysis: {e}")
final_output_lines.append(
f"--- Error: LSP Timeout during analysis setup: {e} ---"
)
analysis_succeeded = False
except Exception as e:
logger.exception(f"Unhandled exception during LSP analysis orchestration: {e}")
final_output_lines.append(f"--- Error: Unexpected analysis failure: {e} ---")
analysis_succeeded = False
finally:
# Ensure LSP Client is closed gracefully
if "client" in locals() and client:
logger.info("Shutting down LSP client...")
await client.close()
# --- Append Build Error Section [ ** MODIFIED ** ] ---
# Ensure final_output_lines exists even if analysis failed very early
if not final_output_lines and not analysis_succeeded:
final_output_lines = ["--- Analysis failed before generating LSP output ---"]
# Add appropriate header based on success/failure
if not analysis_succeeded and (
not final_output_lines or not final_output_lines[0].startswith("--- Error:")
):
final_output_lines.insert(0, "--- LSP Analysis Incomplete ---")
# Append the original fallback error message without prefixing/padding
final_output_lines.append("\n--- Build System Output ---") # Clear header
if fallback_error and fallback_error.strip():
# Append raw lines from the fallback error
final_output_lines.extend(fallback_error.strip().splitlines())
else:
final_output_lines.append(
"(Original build system output not provided or empty)"
)
# --- Return Final Result ---
return "\n".join(final_output_lines)