Launching processes

Starkite provides direct, structured process execution via the os module. All command execution is governed uniformly under standard process execution permissions.

The os module provides two primary execution functions:

• os.exec(cmd, args=[]) Executes a binary directly. If the command returns a non-zero exit code, it halts execution and raises a Starlark-level error.

• os.try_exec(cmd, args=[]) Executes a binary directly, returning an ExecResult for programmatic error handling.

---

Direct Process Execution (os.exec & os.try_exec)

os.exec() and os.try_exec() execute target binaries directly via standard system calls. Direct process execution avoids shell spawning overhead and prevents shell injection vulnerabilities.

Starkite supports two invocation styles:

1. Single Command String (Lexical Tokenization)

When passing a single command string, Starkite uses an internal lexical tokenizer to parse the command and its arguments without invoking an external shell:

def check_status():
    # Double quotes preserve arguments with internal spaces
    os.exec('git commit -m "initial commit"')

    # Single quotes preserve literal payloads (e.g. JSON or Regex)
    os.exec("grep -E '^[0-9]+' output.txt")

    # Backslash escapes outside quotes
    os.exec("cat file\\ with\\ spaces.txt")

    # Quoted executable paths
    os.exec('"/opt/custom tools/bin/app" --verbose')

Lexical Quoting Rules

Syntax	Behavior	Example	Parsed Result
Single quotes ('...')	Preserves literal text inside quotes. No escape interpretation.	'{"k":"v"}'	{"k":"v"}
Double quotes ("...")	Preserves spaces; interprets backslash escapes (\", \\).	"hello \"world\""	hello "world"
Escaped space (\)	Preserves space as part of the current token.	file\ name.txt	file name.txt
Quoted binary	Quoted executable path containing spaces.	"/opt/my app/bin"	/opt/my app/bin

2. Structured Arguments List

To pass pre-constructed argument slices without string formatting, pass the binary name as the first argument and a list/tuple of strings as the second:

def check_os():
    # Passes arguments directly as argv elements
    os_info = os.exec("uname", ["-a"])
    print("System OS:", os_info.strip())

Running Shell Features (Pipes, Redirection, Chaining)

Because direct process execution does not use a shell, shell features like pipes (|), redirections (>, <), and chaining (&&, ;) can be invoked via os.shell() or the shell factory shortcuts:

def check_disk_space():
    sh = os.sh()
    disk_info = sh.exec("df -h / | tail -1")
    print("Disk Info:", disk_info.strip())

Permissions

Direct process execution requires the allow-all permission profile (or a custom profile explicitly granting os.exec permissions):

kite run ./script.star --permissions allow-all

---

Shell Execution (os.shell and Shortcuts)

For scripts requiring shell parsing, multi-command pipelines, or environment persistence, the os module provides the os.shell() constructor and factory shortcuts.

Shell Constructors and Shortcuts

Constructor	Default Command	Default Flag	Description
os.shell(...)	Platform default (/bin/sh or cmd.exe)	Platform default (-c or /c)	Configurable shell instance
os.sh(...)	/bin/sh	-c	Standard POSIX shell
os.bash(...)	/bin/bash	-c	Bash shell
os.zsh(...)	/bin/zsh	-c	Zsh shell
os.cmdexe(...)	cmd.exe	/c	Windows Command Prompt
os.powershell(...)	pwsh (or powershell.exe)	-Command	PowerShell shell

Each constructor accepts initial defaults: * command (string): Target shell binary or path. * flag (string): Argument flag preceding the script (-c, /c, -Command). * cwd (string): Default working directory. * env (dict): Bound environment variable map. * timeout (string): Bound execution timeout (default: "60s"). * userid / groupid (string | int): User and group execution identities (POSIX only).

Shell Methods: exec and try_exec

A Shell instance provides two execution methods: * sh.exec(script, ...): Executes the script string in the shell, returning standard output. Halts execution with an error on non-zero exit. * sh.try_exec(script, ...): Executes the script string, returning an ExecResult for programmatic exit code checks.

Both methods accept per-invocation overrides for cwd, env, timeout, userid, groupid, input, and output.

Examples

Pipelines and Redirections

def extract_process_info():
    bash = os.bash()
    top_proc = bash.exec("ps aux | sort -nrk 3,3 | head -n 5")
    print("Top processes by CPU:\n", top_proc)

Multi-line Script Execution

def run_setup():
    sh = os.sh()
    setup_script = """
    set -e
    mkdir -p build/logs
    touch build/logs/app.log
    """
    sh.exec(setup_script)

Bound Context and Per-Call Overrides

def run_build():
    # Configure base shell environment
    ci = os.bash(
        cwd = "/repo",
        env = {"CI": "true", "GOOS": "linux"},
        timeout = "5m",
    )

    # Per-call environment overrides merge with bound options
    ci.exec("make build", env = {"VERBOSE": "1"})

Permission Model

Constructing a Shell performs no I/O and requires no permissions. Calling sh.exec(script) delegates directly to os.exec([sh.command, sh.flag, script]) and validates against standard os.exec permission rules for the target shell binary.

---

Programmatic Error Handling (os.try_exec)

Use os.try_exec() when you need to handle exit status codes programmatically. It never raises a Starlark error on command failure; instead, it returns an ExecResult object.

def check_disk_space_safe():
    # Safe to handle failure programmatically
    disk = os.try_exec(
        "sh",
        ["-c", "df -h / | tail -1"],
        timeout = "5s",
        cwd = "/tmp",
        env = {"LANG": "C"},
    )

    if disk.ok:
        fields = disk.stdout.split()
        printf("Disk Available: %s (Used: %s)\n", fields[3], fields[4])
    else:
        printf("Warning: Disk check failed with code %d\n", disk.code)

---

Execution Options

All process execution functions accept the following optional keyword arguments:

Option	Type	Default	Purpose
cwd	string	""	Working directory in which to run the sub-process.
env	dict	None	Environment variable overrides (mapping string to string) for the command execution context.
timeout	string	"60s"	Time limit for execution (e.g., "10s", "5m"). The process is killed if the timeout is exceeded.
userid	string | int	None	User identity under which to run the process (POSIX only). See details below.
groupid	string | int	None	Group identity under which to run the process (POSIX only). See details below.
input	string | bytes | io.reader	None	Data or read stream to write to the process standard input.
output	io.writer	None	Write stream to redirect the process standard output to.

---

Handling Results

os.try_exec() returns an ExecResult struct containing the following attributes:

• .ok (bool): True if the command exited with code 0 and no internal errors occurred.
• .code (int): The integer process exit code returned by the command.
• .stdout (string): The standard output stream captured from the command.
• .stderr (string): The standard error stream captured from the command.
• .error (string): A combined error message if the command failed or timed out.

---

User and Group Execution

Local command execution natively supports running sub-processes under specified user and group identities (UID/GID) in POSIX environments (Linux, macOS).

Use the userid and groupid optional keyword arguments to configure the OS credentials of the spawned process:

• userid (string | int): The username or numeric User ID (UID) of the target user.
• groupid (string | int): The group name or numeric Group ID (GID) of the target group.

Examples

Running as a Specific User (Username)

To run a command as a specific user, pass the username to userid:

def query_database():
    # Runs the psql command directly as the postgres user
    result = os.try_exec("psql", ["-c", "SELECT version();"], userid="postgres")
    if result.ok:
        print("Database Version:", result.stdout.strip())

Running with Numeric IDs

To run a command using specific numeric UID and GID:

def run_unprivileged_task():
    # Runs the command as UID 65534 (nobody) and GID 65534 (nogroup)
    result = os.exec("id", userid=65534, groupid=65534)
    print("Identity:", result.strip())

Required Privileges

Changing process credentials (setuid/setgid) requires the parent kite process to have sufficient operating system privileges (typically running as root or having CAP_SETUID/CAP_SETGID capabilities). If kite is run under a standard non-privileged user, the OS kernel will reject the credential switch, and os.exec() will return a standard OS permission error (e.g., operation not permitted).

---

Streaming Input and Output

Subprocess execution integrates with the unified streaming contract via the input and output keyword arguments. This allows piping data directly into a process's standard input and redirecting its standard output.

Streamable Inputs

The input argument accepts: * string: Passed directly as standard input to the command. * bytes: Passed directly as standard input to the command. * io.reader: An active read stream (such as a stream returned by fs.path.get_reader() or an HTTP client response stream). The stream is copied to the subprocess standard input.

Once the command terminates (or times out), the input stream is automatically closed by the runtime.

Streamable Outputs

The output argument accepts: * io.writer: An active write stream (such as a stream returned by fs.path.get_writer()). The subprocess standard output is written directly to the stream.

Once the command terminates (or times out), the output stream is automatically flushed and closed by the runtime.

Examples

Piping a File to a Process

To pipe a file's content directly into the standard input of a process:

def count_lines():
    p = fs.path("data.txt")
    p.write_text("line 1\nline 2\nline 3\n")

    # Streams file data directly to the stdin of 'wc -l'
    reader = p.get_reader()
    result = os.exec("wc -l", input=reader)
    print("Line count:", result.strip())

Piping Process Output to a File

To redirect a process's standard output directly to a file:

def save_system_info():
    out_file = fs.path("sysinfo.txt")

    # Redirects stdout of 'uname -a' to the file writer
    writer = out_file.get_writer()
    os.exec("uname -a", output=writer)

    print("Saved content:", out_file.read_text().strip())

Multi-stage Pipeline (File-to-File via Subprocess)

To stream from an input file, process it with a command, and write the output directly to another file:

def process_pipeline():
    in_file = fs.path("input.log")
    out_file = fs.path("output.log")

    in_file.write_text("debug log entry\nerror log entry\ninfo log entry\n")

    r = in_file.get_reader()
    w = out_file.get_writer()

    # Pipes input.log into 'grep error' and writes the output directly to output.log
    os.exec("grep error", input=r, output=w)

    print("Filtered log:", out_file.read_text().strip())