Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
299 changes: 272 additions & 27 deletions src/vulcanai/console/terminal_session.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,9 +13,12 @@
# limitations under the License.

import os
import re
import shutil
import subprocess
import sys
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Optional, Protocol


Expand All @@ -39,8 +42,8 @@ def write_terminal_sequence(sequence: str) -> None:
class TerminalAdapter(Protocol):
"""
Abstract parent class to enhance VulcanAI visualization in each terminal.
Currently supported: Gnome
Not yet implemented: Terminator, Zsh
Currently supported: Gnome, Terminator
Not yet implemented: Zsh
"""

name: str
Expand All @@ -57,26 +60,6 @@ def restore(self, state: Any) -> None: ...
# region gnome


def _run_gsettings(*args: str) -> Optional[str]:
"""
@brief Run gsettings and return trimmed stdout on success.
@param args Positional arguments forwarded to ``gsettings``.
@return Command stdout without trailing whitespace, or ``None`` on failure.
"""
try:
completed = subprocess.run(
["gsettings", *args],
check=False,
capture_output=True,
text=True,
)
except Exception:
return None
if completed.returncode != 0:
return None
return completed.stdout.strip()


@dataclass
class GnomeState:
"""@brief State required to restore GNOME Terminal settings."""
Expand All @@ -90,6 +73,26 @@ class GnomeTerminalAdapter:

name = "gnome-terminal"

@staticmethod
def _run_gsettings(*args: str) -> Optional[str]:
"""
@brief Run gsettings and return trimmed stdout on success.
@param args Positional arguments forwarded to ``gsettings``.
@return Command stdout without trailing whitespace, or ``None`` on failure.
"""
try:
completed = subprocess.run(
["gsettings", *args],
check=False,
capture_output=True,
text=True,
)
except Exception:
return None
if completed.returncode != 0:
return None
return completed.stdout.strip()

def detect(self) -> bool:
"""
@brief Detect whether the current terminal is GNOME Terminal.
Expand All @@ -108,7 +111,7 @@ def apply(self) -> Optional[GnomeState]:
@return ``GnomeState`` when the change is applied/confirmed, else ``None``.
"""
# The return value could be None, empty string or string with just single quotes
profile_id = _run_gsettings("get", "org.gnome.Terminal.ProfilesList", "default")
profile_id = self._run_gsettings("get", "org.gnome.Terminal.ProfilesList", "default")
if not profile_id:
return None
profile_id = profile_id.strip("'")
Expand All @@ -117,13 +120,13 @@ def apply(self) -> Optional[GnomeState]:

# GNOME stores per-profile keys under this dynamic schema path.
schema = f"org.gnome.Terminal.Legacy.Profile:/org/gnome/terminal/legacy/profiles:/:{profile_id}/"
current_policy = _run_gsettings("get", schema, "scrollbar-policy")
current_policy = self._run_gsettings("get", schema, "scrollbar-policy")
if not current_policy:
return None

# set only if needed
if current_policy != "'never'":
_run_gsettings("set", schema, "scrollbar-policy", "never")
self._run_gsettings("set", schema, "scrollbar-policy", "never")

return GnomeState(schema=schema, scrollbar_policy_backup=current_policy)

Expand All @@ -137,7 +140,241 @@ def restore(self, state: Optional[GnomeState]) -> None:
return
restore_value = state.scrollbar_policy_backup.strip("'")
if restore_value:
_run_gsettings("set", state.schema, "scrollbar-policy", restore_value)
self._run_gsettings("set", state.schema, "scrollbar-policy", restore_value)


# endregion

# region terminator

@dataclass
class TerminatorState:
terminal_uuid: str
previous_profile: str


class TerminatorTerminalAdapter:
"""@brief Terminator adapter that switches to a hidden-scroll profile temporarily."""

name = "terminator"
# Matches any top-level section like [profiles], [global_config], [layouts].
# Needed to detect where the [profiles] block ends.
_TOP_LEVEL_SECTION_RE = re.compile(r"^\s*\[[^\[\]].*\]\s*$")
# Matches profile headers inside [profiles], e.g. " [[default]]".
# Group 1 stores indentation so cloned profiles preserve style.
# Group 2 stores the profile name.
_PROFILE_HEADER_RE = re.compile(r"^(\s*)\[\[(.+?)\]\]\s*$")
# Matches generic "key = value" rows and captures indentation.
# Used when appending missing keys with consistent formatting.
_KEY_VALUE_RE = re.compile(r"^(\s*)[A-Za-z0-9_]+\s*=")
# Matches the specific scrollbar setting row.
# Used to rewrite current value to "disabled" without touching other keys.
_SCROLLBAR_RE = re.compile(r"^(\s*)scrollbar_position\s*=")

def __init__(self, config: "TerminalSessionConfig"):
self._config = config

# -- Utils ----------------------------------------------------------------

@staticmethod
def _run(*args: str) -> bool:
"""
Execute a command and return success status.

@return ``True`` when process exits with code ``0``, else ``False``.
"""
try:
completed = subprocess.run(
[*args],
check=False,
capture_output=True,
text=True,
)
except Exception:
return False
return completed.returncode == 0

@staticmethod
def _config_path() -> Path:
"""
Resolve Terminator config file location.

@return Absolute path to ``terminator/config`` under ``XDG_CONFIG_HOME`` or ``~/.config``.
"""
config_root = os.environ.get("XDG_CONFIG_HOME") or os.path.join(os.path.expanduser("~"), ".config")
return Path(config_root) / "terminator" / "config"

@classmethod
def _ensure_hidden_profile(cls, config_path: Path, base_profile: str, hidden_profile: str) -> bool:
"""
Ensure hidden profile exists and has ``scrollbar_position = disabled``.

@return ``True`` when config is ready for profile switching, else ``False``.
"""
try:
lines = config_path.read_text(encoding="utf-8").splitlines(keepends=True)
except Exception:
return False

profiles_start = next((i for i, line in enumerate(lines) if line.strip() == "[profiles]"), None)
if profiles_start is None:
return False

profiles_end = len(lines)
for index in range(profiles_start + 1, len(lines)):
if cls._TOP_LEVEL_SECTION_RE.match(lines[index]) and lines[index].strip() != "[profiles]":
profiles_end = index
break

profile_headers: list[tuple[str, int, str]] = []
for index in range(profiles_start + 1, profiles_end):
header_match = cls._PROFILE_HEADER_RE.match(lines[index].rstrip("\r\n"))
if header_match:
profile_headers.append((header_match.group(2).strip(), index, header_match.group(1)))

if not profile_headers:
return False

blocks: dict[str, tuple[int, int, str]] = {}
for idx, (name, start_idx, indent) in enumerate(profile_headers):
end_idx = profile_headers[idx + 1][1] if idx + 1 < len(profile_headers) else profiles_end
blocks[name] = (start_idx, end_idx, indent)

def ensure_disabled(block: list[str], section_indent: str) -> list[str]:
"""
Update one profile block so scrollbar is always disabled.
"""
updated = [block[0]]
scrollbar_found = False
key_indent = None
for line in block[1:]:
stripped = line.rstrip("\r\n")
if key_indent is None:
key_match = cls._KEY_VALUE_RE.match(stripped)
if key_match:
key_indent = key_match.group(1)
scrollbar_match = cls._SCROLLBAR_RE.match(stripped)
if scrollbar_match:
updated.append(f"{scrollbar_match.group(1)}scrollbar_position = disabled\n")
scrollbar_found = True
else:
updated.append(line)
if not scrollbar_found:
indent = key_indent if key_indent is not None else f"{section_indent} "
updated.append(f"{indent}scrollbar_position = disabled\n")
return updated

changed = False
if hidden_profile in blocks:
hidden_start, hidden_end, hidden_indent = blocks[hidden_profile]
hidden = ensure_disabled(lines[hidden_start:hidden_end], hidden_indent)
if hidden != lines[hidden_start:hidden_end]:
lines = lines[:hidden_start] + hidden + lines[hidden_end:]
changed = True
else:
if base_profile not in blocks:
return False
base_start, base_end, base_indent = blocks[base_profile]
hidden = [f"{base_indent}[[{hidden_profile}]]\n", *lines[base_start:base_end][1:]]
lines = lines[:profiles_end] + ensure_disabled(hidden, base_indent) + lines[profiles_end:]
changed = True

if changed:
try:
config_path.write_text("".join(lines), encoding="utf-8")
except Exception:
return False
Comment on lines +282 to +286
Copy link

Copilot AI Mar 12, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

_ensure_hidden_profile() can permanently modify the user's Terminator config (e.g., creating a new profile or overwriting an existing profile's scrollbar_position) and there is no restoration/rollback in restore(). This has user-facing operational impact. Consider tracking whether the file/profile was changed and reverting it on session end, or only creating/modifying a uniquely-named VulcanAI profile when it doesn't already exist (without overwriting user-defined profiles).

Copilot uses AI. Check for mistakes.
return True

# -------------------------------------------------------------------------

def detect(self) -> bool:
"""
@brief Detect whether current terminal is Terminator.
@return ``True`` when Terminator environment markers are present.
"""
return (
"TERMINATOR_UUID" in os.environ
or "terminator" in os.environ.get("TERMINAL_EMULATOR", "").lower()
or "terminator" in os.environ.get("TERM_PROGRAM", "").lower()
)

def apply(self) -> Optional[TerminatorState]:
"""
@brief Switch current Terminator tab to hidden-scroll profile.
@return ``(uuid, base_profile)`` when switching succeeds, else ``None``.
"""
terminal_uuid = os.environ.get("TERMINATOR_UUID")
if not terminal_uuid:
return None

if not shutil.which("remotinator"):
return None

previous_profile = self._resolve_previous_profile()
if not previous_profile:
return None

config_path = self._config_path()
if not config_path.is_file():
return None

if not self._ensure_hidden_profile(
config_path=config_path,
base_profile=previous_profile,
hidden_profile=self._config.terminator_profile_hidden,
):
return None

switched = self._run(
"remotinator",
"switch_profile",
"-u",
terminal_uuid,
"-p",
self._config.terminator_profile_hidden,
)
if not switched:
return None

return TerminatorState(
terminal_uuid=terminal_uuid,
previous_profile=previous_profile,
)

def _resolve_previous_profile(self) -> Optional[str]:
candidates = (
self._config.terminator_profile_current,
os.environ.get("VULCANAI_TERMINATOR_PROFILE"),
self._config.terminator_profile_base,
)
for profile in candidates:
if profile is None:
continue
normalized = profile.strip()
if normalized:
return normalized
return None

def restore(self, state: Optional[TerminatorState]) -> None:
"""
@brief Restore previous Terminator profile.
@param state Previously saved state; no-op when ``None``.
@return None
"""
if not state:
return
if not shutil.which("remotinator"):
return

self._run(
"remotinator",
"switch_profile",
"-u",
state.terminal_uuid,
"-p",
state.previous_profile,
)


# endregion
Expand All @@ -158,6 +395,12 @@ class TerminalSessionConfig:
force_bg: bool = True
# Emit DEC private mode sequence to hide/show scrollbar.
hide_scrollbar: bool = True
# Terminator profile to restore once session ends.
terminator_profile_base: str = "default"
# Terminator profile used while session is running.
terminator_profile_hidden: str = "vulcanai-no-scroll"

terminator_profile_current: Optional[str] = None


class TerminalSession:
Expand All @@ -171,7 +414,9 @@ def __init__(
adapters: Optional[list[TerminalAdapter]] = None,
):
self.config = config if config is not None else TerminalSessionConfig()
self.adapters = adapters if adapters is not None else [GnomeTerminalAdapter()]
self.adapters = (
adapters if adapters is not None else [GnomeTerminalAdapter(), TerminatorTerminalAdapter(self.config)]
)
self._active: list[tuple[TerminalAdapter, Any]] = []

def start(self) -> None:
Expand Down
2 changes: 1 addition & 1 deletion src/vulcanai/core/plan_types.py
Original file line number Diff line number Diff line change
Expand Up @@ -117,7 +117,7 @@ def __str__(self) -> str:
if node.success_criteria:
# Succes Criteria: <node.success_criteria>
lines.append(
f"\<{color_tool}>tSuccess Criteria</{color_tool}>: "
f"\t<{color_tool}>Success Criteria</{color_tool}>: "
+ f"<{color_value}>{node.success_criteria}</{color_value}>"
)
if node.on_fail:
Expand Down
Loading