mirrors/kitty
1
0
Fork 0
mirror of https://github.com/kovidgoyal/kitty synced 2026-07-22 00:08:04 +02:00
Code Issues Packages Projects Releases Wiki Activity

...

Browse Source
This commit is contained in:
Kovid Goyal
2026-07-21 15:34:38 +05:30
parent e36f6ac1b8
commit 51116e5df3
1 changed files with 41 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

56
kitty/child.py
View File

@@ -9,7 +9,7 @@ from collections.abc import Generator, Iterable, Mapping, Sequence
from contextlib import contextmanager, suppress
from itertools import count
from time import monotonic
from typing import TYPE_CHECKING, DefaultDict, Optional, TypedDict
from typing import TYPE_CHECKING, Any, DefaultDict, Optional, TypedDict
import kitty.fast_data_types as fast_data_types
@@ -51,14 +51,17 @@ else:
return list(filter(None, f.read().decode('utf-8').split('\0')))
if is_freebsd:
def cwd_of_process(pid: int) -> str:
import subprocess
cp = subprocess.run(['pwdx', str(pid)], capture_output=True)
if cp.returncode != 0:
raise ValueError(f'Failed to find cwd of process with pid: {pid}')
ans = cp.stdout.decode('utf-8', 'replace').split()[1]
return os.path.realpath(ans, strict=True)
else:
def cwd_of_process(pid: int) -> str:
# We use realpath instead of readlink to match macOS behavior where
# the underlying OS API returns real paths.
@@ -98,7 +101,6 @@ def checked_terminfo_dir() -> str | None:
class CachedProcessData:
cached_result: DefaultDict[int, list[int]] | None = None
cache_active: bool = False
cache_at: float = 0
@@ -150,6 +152,7 @@ def session_id(pids: Iterable[int]) -> int:
return sid
return -1
def parse_environ_block(data: str) -> dict[str, str]:
"""Parse a C environ block of environment variables into a dictionary."""
# The block is usually raw data from the target process. It might contain
@@ -158,15 +161,15 @@ def parse_environ_block(data: str) -> dict[str, str]:
pos = 0
while True:
next_pos = data.find("\0", pos)
next_pos = data.find('\0', pos)
# nul byte at the beginning or double nul byte means finish
if next_pos <= pos:
break
# there might not be an equals sign
equal_pos = data.find("=", pos, next_pos)
equal_pos = data.find('=', pos, next_pos)
if equal_pos > pos:
key = data[pos:equal_pos]
value = data[equal_pos + 1:next_pos]
value = data[equal_pos + 1 : next_pos]
ret[key] = value
pos = next_pos + 1
@@ -237,9 +240,9 @@ child_counter = count()
class Child:
child_fd: int | None = None
pid: int | None = None
initial_termios_state: list[Any] | None = None
forked = False
def __init__(
@@ -274,18 +277,23 @@ class Child:
self.stdin = stdin
self.env = env or {}
self.startup_command_via_shell_integration = startup_command_via_shell_integration
self.final_env:dict[str, str] = {}
self.final_env: dict[str, str] = {}
self.is_default_shell = bool(self.argv and self.argv[0] == shell_path)
self.should_run_via_run_shell_kitten = is_macos and self.is_default_shell
self.hold = hold
def get_final_env(self) -> tuple[dict[str, str], bool]:
from kitty.options.utils import DELETE_ENV_VAR
env = default_env().copy()
opts = fast_data_types.get_options()
boss = fast_data_types.get_boss()
if is_macos and env.get('LC_CTYPE') == 'UTF-8' and not getattr(sys, 'kitty_run_data').get(
'lc_ctype_before_python') and not getattr(default_env, 'lc_ctype_set_by_user', False):
if (
is_macos
and env.get('LC_CTYPE') == 'UTF-8'
and not getattr(sys, 'kitty_run_data').get('lc_ctype_before_python')
and not getattr(default_env, 'lc_ctype_set_by_user', False)
):
del env['LC_CTYPE']
env.update(self.env)
env['TERM'] = opts.term
@@ -314,6 +322,7 @@ class Child:
self.unmodified_argv = list(self.argv)
if not self.should_run_via_run_shell_kitten and 'disabled' not in opts.shell_integration:
from .shell_integration import modify_shell_environ
modify_shell_environ(opts, env, self.argv)
env = {k: v for k, v in env.items() if v is not DELETE_ENV_VAR}
if self.is_clone_launch:
@@ -327,6 +336,7 @@ class Child:
env['KITTY_SI_RUN_COMMAND_AT_STARTUP'] = self.startup_command_via_shell_integration
else:
from .shell_integration import join
scmd = self.argv or resolved_shell(fast_data_types.get_options())
try:
env['KITTY_SI_RUN_COMMAND_AT_STARTUP'] = join(scmd[0], self.startup_command_via_shell_integration)
@@ -356,7 +366,7 @@ class Child:
cwd = self.cwd
pass_fds = self.pass_fds
if self.remote_control_fd > -1:
pass_fds += self.remote_control_fd,
pass_fds += (self.remote_control_fd,)
if self.should_run_via_run_shell_kitten or must_run_startup_command_via_kitten:
# bash will only source ~/.bash_profile if it detects it is a login
# shell (see the invocation section of the bash man page), which it
@@ -373,6 +383,7 @@ class Child:
# xterm, urxvt, konsole and gnome-terminal do not do it in my
# testing.
import shlex
ksi = ' '.join(opts.shell_integration)
if ksi == 'invalid':
ksi = 'enabled'
@@ -388,6 +399,7 @@ class Child:
# login closes inherited file descriptors so dont use it when
# forward_stdio or pass_fds are used.
import pwd
user = pwd.getpwuid(os.geteuid()).pw_name
if cwd:
argv.append('--cwd=' + cwd)
@@ -400,8 +412,21 @@ class Child:
final_exe = argv[0]
env = tuple(f'{k}={v}' for k, v in self.final_env.items())
pid = fast_data_types.spawn(
final_exe, cwd, tuple(argv), env, master, slave, stdin_read_fd, stdin_write_fd,
ready_read_fd, ready_write_fd, tuple(handled_signals), kitten_exe(), opts.forward_stdio, pass_fds)
final_exe,
cwd,
tuple(argv),
env,
master,
slave,
stdin_read_fd,
stdin_write_fd,
ready_read_fd,
ready_write_fd,
tuple(handled_signals),
kitten_exe(),
opts.forward_stdio,
pass_fds,
)
os.close(slave)
self.pid = pid
self.child_fd = master
@@ -419,7 +444,7 @@ class Child:
except NotImplementedError:
pass
except OSError as err:
log_error("Could not move child process into a systemd scope: " + str(err))
log_error('Could not move child process into a systemd scope: ' + str(err))
return pid
def __del__(self) -> None:
@@ -569,6 +594,7 @@ class Child:
def send_signal_for_key(self, key_num: bytes) -> bool:
import signal
if self.child_fd is None:
return False
t = termios.tcgetattr(self.child_fd)
@@ -588,8 +614,8 @@ class Child:
return True
def reset_termios_state(self, when: int = termios.TCSANOW) -> None:
if (s := getattr(self, 'initial_termios_state', None)) and self.child_fd is not None:
if self.initial_termios_state is not None and self.child_fd is not None:
try:
termios.tcsetattr(self.child_fd, when, s)
termios.tcsetattr(self.child_fd, when, self.initial_termios_state)
except OSError:
pass