scripts/p5_wave2b_core_patches.py
Rama main · réplica de lectura del repositorio original.
#!/usr/bin/env python3
"""P5 wave2b: targeted mypy strict fixes for remaining core/*_ops.py."""
from __future__ import annotations
import re
import subprocess
import sys
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
CORE = ROOT / "butler/core"
def patch(path: Path, old: str, new: str) -> None:
src = path.read_text(encoding="utf-8")
if old not in src:
raise SystemExit(f"missing patch in {path}: {old[:60]!r}")
path.write_text(src.replace(old, new, 1), encoding="utf-8")
def main() -> int:
# agent_loop_ops
p = CORE / "agent_loop_ops.py"
patch(
p,
" return synthetic_result(decision)\n except Exception:",
" return str(synthetic_result(decision))\n except Exception:",
)
patch(
p,
" return synthetic_result(decision)\n return None",
" return str(synthetic_result(decision))\n return None",
)
patch(
p,
" return filter_fallback_chain(chain)",
" out = filter_fallback_chain(chain)\n return out if isinstance(out, list) else chain",
)
patch(p, " def _run():\n\n return run_stop_hooks(", " def _run() -> Any:\n\n return run_stop_hooks(")
# agents_md_sections_ops
p = CORE / "agents_md_sections_ops.py"
src = p.read_text(encoding="utf-8")
if "cast" not in src:
src = src.replace(
"from pathlib import Path\n",
"from pathlib import Path\nfrom typing import cast\n",
)
src = src.replace(
" return resolve_active_project_workspace_safe()",
" return cast(Path | None, resolve_active_project_workspace_safe())",
)
p.write_text(src, encoding="utf-8")
# batch_sequence_guard_ops
patch(
CORE / "batch_sequence_guard_ops.py",
" return tc.args_dict()",
" raw = tc.args_dict()\n return dict(raw) if isinstance(raw, dict) else {}",
)
# context_pipeline_ops
p = CORE / "context_pipeline_ops.py"
patch(
p,
" return get_audit_session_key(fallback=fallback)",
" return str(get_audit_session_key(fallback=fallback) or fallback)",
)
for fn in (
"apply_unified_tool_masking",
"compress_inline_tool_messages",
"apply_model_transforms",
):
patch(
p,
f" return {fn}(list(messages))",
f" out = {fn}(list(messages))\n return out if isinstance(out, list) else list(messages)",
)
patch(
p,
" return out\n except Exception as exc:",
" return out if isinstance(out, list) else list(messages)\n except Exception as exc:",
)
# llm_retry_invoke_ops
patch(
CORE / "llm_retry_invoke_ops.py",
" on_tool_call_ready: Callable[[int, str, str, dict], None] | None,",
" on_tool_call_ready: Callable[[int, str, str, dict[str, Any]], None] | None,",
)
# message_context_adapter_ops
patch(
CORE / "message_context_adapter_ops.py",
' role: ApiRole = role_raw # type: ignore[assignment]\n',
" role: ApiRole = role_raw\n",
)
# mode_classifier_ops
patch(
CORE / "mode_classifier_ops.py",
"from typing import Literal",
"from typing import Any, Literal",
)
# review_context_adapter_ops
patch(
CORE / "review_context_adapter_ops.py",
" severity=sev, # type: ignore[arg-type]\n",
" severity=sev,\n",
)
# parallel_tools_ops
p = CORE / "parallel_tools_ops.py"
src = p.read_text(encoding="utf-8")
src = src.replace("def parse_tool_args_safe(tc: Any) -> dict:", "def parse_tool_args_safe(tc: Any) -> dict[str, Any]:")
src = src.replace(" def _run() -> dict:", " def _run() -> dict[str, Any]:")
src = src.replace(
" return dispatch_fn(name, args, tool_call_id=tool_id)",
" return str(dispatch_fn(name, args, tool_call_id=tool_id))",
)
src = src.replace(
" return finalize(\n name,\n args,\n {\n \"error\": f\"Tool execution failed: {exc}\",\n \"code\": \"TOOL_DISPATCH_ERROR\",\n },\n )",
" return str(\n finalize(\n name,\n args,\n {\n \"error\": f\"Tool execution failed: {exc}\",\n \"code\": \"TOOL_DISPATCH_ERROR\",\n },\n )\n )",
)
p.write_text(src, encoding="utf-8")
# tool_batch_runner_ops
p = CORE / "tool_batch_runner_ops.py"
src = p.read_text(encoding="utf-8")
src = src.replace("def parse_tool_call_args_safe(tc: Any) -> dict:", "def parse_tool_call_args_safe(tc: Any) -> dict[str, Any]:")
src = src.replace(" def _run() -> dict:", " def _run() -> dict[str, Any]:")
src = src.replace(" return tc.args_dict()", " raw = tc.args_dict()\n return dict(raw) if isinstance(raw, dict) else {}")
p.write_text(src, encoding="utf-8")
# model_context_ops
patch(
CORE / "model_context_ops.py",
" return safe_best_effort(_run, label=\"model_context.max_output_tokens\", default=None)",
" result = safe_best_effort(_run, label=\"model_context.max_output_tokens\", default=None)\n"
" return int(result) if isinstance(result, int) else None",
)
# prompt_renderer_ops
patch(
CORE / "prompt_renderer_ops.py",
" return static_system_reminder_enabled()",
" return bool(static_system_reminder_enabled())",
)
# reasoning_trace_ops
patch(
CORE / "reasoning_trace_ops.py",
" return enrich_fix_hint(level, state)",
" return str(enrich_fix_hint(level, state) or \"\")",
)
# reflection_closure_ops
p = CORE / "reflection_closure_ops.py"
src = p.read_text(encoding="utf-8")
if "cast" not in src:
src = src.replace("from pathlib import Path\n", "from pathlib import Path\nfrom typing import cast\n")
src = src.replace(" return _reflex_path()", " return cast(Path, _reflex_path())")
src = src.replace(
" return env_truthy(\"BUTLER_REFLECTION_CLOSURE_WRITE\", default=False)",
" return bool(env_truthy(\"BUTLER_REFLECTION_CLOSURE_WRITE\", default=False))",
)
p.write_text(src, encoding="utf-8")
# session_hydration_ops
p = CORE / "session_hydration_ops.py"
src = p.read_text(encoding="utf-8")
if "cast" not in src:
src = src.replace("from typing import Any\n", "from typing import Any, cast\n")
src = src.replace(
" return effective_workspace(Path(workspace))",
" return cast(Path, effective_workspace(Path(workspace)))",
)
src = src.replace(
" return safe_best_effort(_run, label=\"session_hydration.transcript_ts\", default=None)",
" result = safe_best_effort(_run, label=\"session_hydration.transcript_ts\", default=None)\n"
" return float(result) if isinstance(result, (int, float)) else None",
)
p.write_text(src, encoding="utf-8")
# session_transcript_ops
for label, fn in (
("session_transcript.load_tail_index", "load_tail_rows_safe"),
("session_transcript.load_tail_full_read", "load_tail_full_read_safe"),
):
pass
p = CORE / "session_transcript_ops.py"
src = p.read_text(encoding="utf-8")
src = src.replace(
" return safe_best_effort(\n _run,\n label=\"session_transcript.load_tail_index\",\n default=None,\n )",
" result = safe_best_effort(\n _run,\n label=\"session_transcript.load_tail_index\",\n default=None,\n )\n"
" return result if isinstance(result, list) else None",
)
src = src.replace(
" return safe_best_effort(\n _run,\n label=\"session_transcript.load_tail_full_read\",\n default=None,\n )",
" result = safe_best_effort(\n _run,\n label=\"session_transcript.load_tail_full_read\",\n default=None,\n )\n"
" return result if isinstance(result, list) else None",
)
p.write_text(src, encoding="utf-8")
# tool_batch_finalize_ops
p = CORE / "tool_batch_finalize_ops.py"
src = p.read_text(encoding="utf-8")
src = src.replace(
" return finalize_unenveloped_failure_result(name, args, result)",
" return str(finalize_unenveloped_failure_result(name, args, result))",
)
src = src.replace(
" return finalize_fallback_tool_result(\n name,\n args,\n {\n \"error\": f\"Tool execution failed: {exc}\",\n \"code\": \"TOOL_DISPATCH_ERROR\",\n },\n )",
" return str(\n finalize_fallback_tool_result(\n name,\n args,\n {\n \"error\": f\"Tool execution failed: {exc}\",\n \"code\": \"TOOL_DISPATCH_ERROR\",\n },\n )\n )",
)
p.write_text(src, encoding="utf-8")
# tool_dispatch_doom_ops
p = CORE / "tool_dispatch_doom_ops.py"
src = p.read_text(encoding="utf-8")
src = src.replace(
" return finalize_fallback_tool_result(name, args, synthetic_result(ask_dec))",
" return str(finalize_fallback_tool_result(name, args, synthetic_result(ask_dec)))",
)
src = src.replace(
" return finalize_fallback_tool_result(name, args, synthetic_result(before))",
" return str(finalize_fallback_tool_result(name, args, synthetic_result(before)))",
)
p.write_text(src, encoding="utf-8")
# tool_dispatch_ops
patch(
CORE / "tool_dispatch_ops.py",
" return annotate_mutation_not_landed(tool_name, result)",
" out = annotate_mutation_not_landed(tool_name, result)\n"
" if isinstance(out, tuple) and len(out) == 2:\n"
" return str(out[0]), bool(out[1])\n"
" return result, False",
)
# tool_orchestrator_ops
patch(
CORE / "tool_orchestrator_ops.py",
" return check_approval(command, cwd=cwd, session_key=session_key)",
" block = check_approval(command, cwd=cwd, session_key=session_key)\n"
" return str(block) if block else None",
)
# tool_retry_ops
patch(
CORE / "tool_retry_ops.py",
" return classify_tool_error(result) == ToolErrorKind.retry",
" return bool(classify_tool_error(result) == ToolErrorKind.retry)",
)
# transcript_search_ops
patch(
CORE / "transcript_search_ops.py",
" return safe_best_effort(\n _run,\n label=\"transcript_search.fts\",\n default=None,\n )",
" result = safe_best_effort(\n _run,\n label=\"transcript_search.fts\",\n default=None,\n )\n"
" return result if isinstance(result, list) else None",
)
# transcript_export_ops
p = CORE / "transcript_export_ops.py"
src = p.read_text(encoding="utf-8")
src = src.replace(
" return safe_best_effort(\n _run,\n label=\"transcript_export.recent_tasks\",\n default=None,\n )",
" result = safe_best_effort(\n _run,\n label=\"transcript_export.recent_tasks\",\n default=None,\n )\n"
" return result if isinstance(result, list) else None",
)
src = src.replace(
" return safe_best_effort(\n _run,\n label=\"transcript_export.workspace\",\n default=None,\n )",
" result = safe_best_effort(\n _run, label=\"transcript_export.workspace\", default=None)\n"
" return result if isinstance(result, Path) else None",
)
p.write_text(src, encoding="utf-8")
print("Applied wave2b patches")
return 0
if __name__ == "__main__":
raise SystemExit(main())