diff --git a/src/spark_intelligence/self_awareness/operating_context.py b/src/spark_intelligence/self_awareness/operating_context.py index 06700e16..0665dd07 100644 --- a/src/spark_intelligence/self_awareness/operating_context.py +++ b/src/spark_intelligence/self_awareness/operating_context.py @@ -627,50 +627,74 @@ def _derived_access_action(*, access: dict[str, Any], execution_lane: dict[str, def _derived_access_lane(*, access: dict[str, Any], execution_lane: dict[str, Any]) -> str: - if str(access.get("effective_level") or "") == "5": - return "level5_operator" - if execution_lane.get("workspace_sandbox") is True or str(access.get("effective_level") or "") == "4": - return "spark_workspace" - return "access_status" + if not isinstance(access, str): access = str(access or '') + if not isinstance(execution_lane, str): execution_lane = str(execution_lane or '') + try: + if str(access.get("effective_level") or "") == "5": + return "level5_operator" + if execution_lane.get("workspace_sandbox") is True or str(access.get("effective_level") or "") == "4": + return "spark_workspace" + return "access_status" + + except Exception: + return "" def _run_policy_for_access_action(command: str) -> str: - command_text = str(command or "") - if "--enable-high-agency" in command_text: - return "explicit_opt_in" - if "disable-level5" in command_text or "docker smoke" in command_text or command_text == "spark restart": - return "confirm_once" - if " status" in command_text or "doctor" in command_text: + if not isinstance(command, str): command = str(command or '') + try: + command_text = str(command or "") + if "--enable-high-agency" in command_text: + return "explicit_opt_in" + if "disable-level5" in command_text or "docker smoke" in command_text or command_text == "spark restart": + return "confirm_once" + if " status" in command_text or "doctor" in command_text: + return "auto_read_only" + if "access setup" in command_text: + return "auto_safe" return "auto_read_only" - if "access setup" in command_text: - return "auto_safe" - return "auto_read_only" + + except Exception: + return "" def _confirmation_for_run_policy(run_policy: str) -> str | None: - if run_policy == "explicit_opt_in": - return "Enable whole-computer operator mode" - if run_policy == "confirm_once": - return "Confirm once before running this access action" - return None + if not isinstance(run_policy, str): run_policy = str(run_policy or '') + try: + if run_policy == "explicit_opt_in": + return "Enable whole-computer operator mode" + if run_policy == "confirm_once": + return "Confirm once before running this access action" + return None + + except Exception: + return "" def _command_text(command: object) -> str: - if isinstance(command, list): - parts = [str(part).strip() for part in command if str(part).strip()] - if parts and parts[0] != "spark": - parts.insert(0, "spark") - return " ".join(parts) - return str(command or "").strip() + try: + if isinstance(command, list): + parts = [str(part).strip() for part in command if str(part).strip()] + if parts and parts[0] != "spark": + parts.insert(0, "spark") + return " ".join(parts) + return str(command or "").strip() + + except Exception: + return "" def _confirmation_text(value: object) -> str | None: - text = str(value or "").strip() - if not text or text.casefold() == "none": - return None - return text + try: + text = str(value or "").strip() + if not text or text.casefold() == "none": + return None + return text + + except Exception: + return "" def _first_non_empty(*values: object) -> str: for value in values: text = str(value or "").strip()