feat(auto): enable LLM-free fixos auto, autonomous repair, and bounded disk cleanup (ticket-088)

fixos/agent/autonomous.py

@@ -9,22 +9,22 @@
 from __future__ import annotations
 
 from .autonomous_session import (
-    AutonomousSession,
-    AgentReport,
-    FixAction,
-    SYSTEM_PROMPT_AUTONOMOUS,
     FORBIDDEN_COMMANDS,
     SUDO_PREFIXES,
+    SYSTEM_PROMPT_AUTONOMOUS,
+    AgentReport,
+    AutonomousSession,
+    FixAction,
 )
 
 __all__ = [
-    "run_autonomous_session",
-    "AutonomousSession",
-    "AgentReport",
-    "FixAction",
-    "SYSTEM_PROMPT_AUTONOMOUS",
     "FORBIDDEN_COMMANDS",
     "SUDO_PREFIXES",
+    "SYSTEM_PROMPT_AUTONOMOUS",
+    "AgentReport",
+    "AutonomousSession",
+    "FixAction",
+    "run_autonomous_session",
 ]
 
 
@@ -33,6 +33,9 @@ def run_autonomous_session(
     config,
     show_data: bool = True,
     max_fixes: int = 10,
+    auto_confirm: bool = False,
+    use_llm: bool | None = None,
+    dry_run: bool = False,
 ) -> AgentReport:
     """
     Uruchamia autonomiczny tryb agenta.
@@ -45,5 +48,8 @@ def run_autonomous_session(
         config=config,
         show_data=show_data,
         max_fixes=max_fixes,
+        auto_confirm=auto_confirm,
+        use_llm=use_llm,
+        dry_run=dry_run,
     )
     return session.run()

fixos/agent/autonomous_session.py

@@ -120,11 +120,18 @@ def __init__(
         config: FixOsConfig,
         show_data: bool = True,
         max_fixes: int = 10,
+        auto_confirm: bool = False,
+        use_llm: bool | None = None,
+        dry_run: bool = False,
     ):
         self.diagnostics = diagnostics
         self.config = config
         self.show_data = show_data
         self.max_fixes = max_fixes
+        self.auto_confirm = auto_confirm
+        self.use_llm = use_llm
+        self.dry_run = dry_run
+        self._heuristic_queue: list[dict[str, Any]] | None = None
         self.llm = LLMClient(config)
         self.report = AgentReport()
         self.fix_count = 0
@@ -133,6 +140,22 @@ def __init__(
         self.start_time = time.time()
         self._setup_timeout()
 
+    def _is_llm_active(self) -> bool:
+        """Return True if LLM calls should be used, False for LLM-free heuristic mode."""
+        if getattr(self, "use_llm", None) is False:
+            return False
+        if getattr(self, "use_llm", None) is True:
+            return True
+        llm_obj = getattr(self, "llm", None)
+        if llm_obj is not None and not isinstance(llm_obj, LLMClient):
+            return True
+        cfg = getattr(self, "config", None)
+        if cfg is None:
+            return False
+        if hasattr(cfg, "validate") and callable(cfg.validate):
+            return len(cfg.validate()) == 0
+        return bool(getattr(cfg, "api_key", None))
+
     def _setup_timeout(self) -> None:
         """Setup session timeout handler."""
 
@@ -148,24 +171,43 @@ def _clear_timeout(self) -> None:
 
     def _confirm_start(self) -> bool:
         """Ask user for confirmation before starting autonomous mode."""
+        import sys
+
+        llm_active = self._is_llm_active()
+        model_label = (
+            getattr(self.config, "model", "unknown")
+            if llm_active
+            else "heurystyki lokalne (bez LLM)"
+        )
         print(f"\n{'═' * UI_BORDER_WIDTH}")
         print("  ⚠️  TRYB AUTONOMICZNY – agent sam wykonuje komendy!")
         print("  Naciśnij Ctrl+C w dowolnym momencie aby przerwać.")
         print("═" * UI_BORDER_WIDTH)
         print(f"\n  Max napraw: {self.max_fixes}")
         print(f"  Timeout sesji: {self.config.session_timeout}s")
-        print(f"  Model: {self.config.model}")
+        print(f"  Model: {model_label}")
+
+        if getattr(self, "auto_confirm", False):
+            print("  Potwierdzenie: automatyczne (--yes / fixos auto)")
+            return True
 
-        confirm = input(
-            "\n  Czy na pewno chcesz uruchomić tryb autonomiczny? (yes/N): "
-        ).strip()
+        try:
+            confirm = input(
+                "\n  Czy na pewno chcesz uruchomić tryb autonomiczny? (yes/N): "
+            ).strip()
+        except EOFError:
+            if not (hasattr(sys.stdin, "isatty") and sys.stdin.isatty()):
+                print("  Brak interaktywnego terminala — anulowano (użyj --yes lub fixos auto).")
+            return False
         if confirm.lower() not in ("yes", "tak"):
             print("  Anulowano. Użyj --mode hitl dla trybu z potwierdzeniem.")
             return False
         return True
 
     def _initialize_messages(self) -> None:
         """Initialize LLM message history with system prompt and diagnostics."""
+        if not self._is_llm_active():
+            return
         anon_str, anon_report = anonymize(str(self.diagnostics))
         if self.show_data:
             display_anonymized_preview(anon_str, anon_report)
@@ -181,6 +223,156 @@ def _initialize_messages(self) -> None:
             },
         ]
 
+    @staticmethod
+    def _is_command_applicable_on_host(cmd: str) -> bool:
+        """Check whether a heuristic repair command targets tools/units present on this host."""
+        import shlex
+        import shutil
+        from pathlib import Path
+
+        try:
+            tokens = shlex.split(cmd)
+        except ValueError:
+            tokens = cmd.split()
+        idx = 0
+        while idx < len(tokens) and (
+            tokens[idx] in ("sudo", "env", "nohup") or "=" in tokens[idx] or tokens[idx].startswith("-")
+        ):
+            idx += 1
+        if idx >= len(tokens):
+            return False
+        prog = tokens[idx]
+        if (
+            prog in ("dnf", "yum", "pacman", "zypper", "apk", "ufw", "firewall-cmd")
+            and shutil.which(prog) is None
+        ):
+            return False
+        return not (
+            prog == "systemctl"
+            and "firewalld" in tokens
+            and shutil.which("firewall-cmd") is None
+            and not Path("/lib/systemd/system/firewalld.service").exists()
+            and not Path("/usr/lib/systemd/system/firewalld.service").exists()
+        )
+
+    def _build_heuristic_actions(self) -> list[dict[str, Any]]:
+        """Build deterministic repair and safe cleanup actions without LLM."""
+        from .session_core import package_cleanup_guard
+
+        actions: list[dict[str, Any]] = []
+        seen_commands: set[str] = set()
+
+        def _add_candidate(cmd: str, reason: str, severity: str = "warning") -> None:
+            cmd_clean = (cmd or "").strip()
+            if not cmd_clean or cmd_clean in seen_commands:
+                return
+            if self._is_forbidden(cmd_clean) or package_cleanup_guard(cmd_clean):
+                return
+            if not self._is_command_applicable_on_host(cmd_clean):
+                return
+            seen_commands.add(cmd_clean)
+            if reason and reason not in self.report.problems_found:
+                self.report.problems_found.append(reason)
+            actions.append(
+                {
+                    "action": "EXEC",
+                    "command": cmd_clean,
+                    "reason": reason or "Naprawa heurystyczna (bez LLM)",
+                    "analysis": f"[{severity}] {reason or cmd_clean}",
+                    "severity": severity,
+                }
+            )
+
+        diag = self.diagnostics if isinstance(self.diagnostics, dict) else {}
+
+        # 1. Inspect findings already present in diagnostics dict
+        for key, val in diag.items():
+            if isinstance(val, dict):
+                for item in val.get("findings", []) or val.get("problems", []) or []:
+                    if isinstance(item, dict) and item.get("command"):
+                        sev = str(item.get("severity", "warning")).lower()
+                        if sev in ("critical", "warning", "high", "error"):
+                            _add_candidate(
+                                str(item["command"]),
+                                str(item.get("title") or item.get("description") or key),
+                                sev,
+                            )
+
+        # 2. Run offline diagnostic plugins if no findings were pre-populated
+        if not actions and diag.get("_run_plugins", True) is not False:
+            try:
+                from ..plugins.base import Severity
+                from ..plugins.registry import PluginRegistry
+
+                registry = PluginRegistry()
+                registry.discover()
+                for res in registry.run():
+                    for finding in res.findings:
+                        if finding.command and finding.severity in (
+                            Severity.CRITICAL,
+                            Severity.WARNING,
+                        ):
+                            _add_candidate(
+                                finding.command,
+                                f"{finding.title}: {finding.description}",
+                                finding.severity.value,
+                            )
+            except Exception:  # noqa: BLE001, S110
+                pass
+
+        # 3. Extract safe disk cleanup suggestions if disk_analysis is present
+        disk_analysis = diag.get("disk_analysis")
+        if isinstance(disk_analysis, dict):
+            suggestions = disk_analysis.get("suggestions", [])
+            if suggestions:
+                try:
+                    from ..interactive.cleanup_planner import CleanupPlanner
+
+                    plan = CleanupPlanner().create_cleanup_plan(suggestions)
+                    for item in plan.get("prioritized_actions", []):
+                        if item.get("safe") and item.get("command"):
+                            _add_candidate(
+                                str(item["command"]),
+                                str(item.get("description") or "Bezpieczne czyszczenie dysku"),
+                                str(item.get("priority") or "medium"),
+                            )
+                except Exception:  # noqa: BLE001, S110
+                    pass
+
+        return actions
+
+    def _process_heuristic_turn(self) -> bool:
+        """Execute one deterministic turn without calling an LLM."""
+        self._check_timeout()
+        if self._heuristic_queue is None:
+            self._heuristic_queue = self._build_heuristic_actions()
+
+        if not self._heuristic_queue:
+            self._handle_done()
+            return False
+
+        remaining = self._get_remaining_time()
+        print(
+            f"  ⟳ Tura {self.fix_count + 1}/{self.max_fixes} | ⏰ {remaining}s pozostało (tryb bez LLM)"
+        )
+        action_data = self._heuristic_queue.pop(0)
+        analysis = action_data.get("analysis", "")
+        reason = action_data.get("reason", "")
+        cmd = action_data.get("command", "")
+        print(f"\n  📋 Analiza: {analysis}")
+        print(f"  🎯 Akcja: EXEC – {reason}")
+
+        if getattr(self, "dry_run", False):
+            print(f"  [DRY-RUN] Pominięto wykonanie: {cmd}")
+            self.report.fixes_applied.append(
+                FixAction(command=cmd, reason=reason, result="[DRY-RUN]", success=True)
+            )
+            self.fix_count += 1
+            return True
+
+        self._handle_exec(action_data)
+        return True
+
     def _get_remaining_time(self) -> int:
         """Get remaining session time in seconds."""
         from . import get_remaining_time
@@ -376,6 +568,9 @@ def _process_turn(self) -> bool:
 
         Returns False if session should end, True to continue.
         """
+        if not self._is_llm_active():
+            return self._process_heuristic_turn()
+
         self._check_timeout()
         remaining = self._get_remaining_time()
         print(
@@ -468,12 +663,18 @@ def run_autonomous_session(
     config: FixOsConfig,
     show_data: bool = True,
     max_fixes: int = 10,
+    auto_confirm: bool = False,
+    use_llm: bool | None = None,
+    dry_run: bool = False,
 ) -> AgentReport:
     """Run autonomous session (backward compatible wrapper)."""
     session = AutonomousSession(
         diagnostics=diagnostics,
         config=config,
         show_data=show_data,
         max_fixes=max_fixes,
+        auto_confirm=auto_confirm,
+        use_llm=use_llm,
+        dry_run=dry_run,
     )
     return session.run()