Skip to content

atlas.core.runner

atlas.core.runner

Task execution via subprocess with local-first tool resolution.

resolve_tool 

resolve_tool(
    tool_name: str, project_dir: str
) -> str | None

Return the path to tool_name, preferring project-local installations.

Resolution cascade: 1. <project_dir>/.venv/bin/<tool> (Python virtual environment) 2. <project_dir>/node_modules/.bin/<tool> (Node.js local packages) 3. shutil.which(<tool>) (system PATH) 4. None — tool not found

Atlas NEVER installs packages; it only informs when a tool is missing.

Source code in src/atlas/core/runner.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
def resolve_tool(tool_name: str, project_dir: str) -> str | None:
    """Return the path to *tool_name*, preferring project-local installations.

    Resolution cascade:
    1. ``<project_dir>/.venv/bin/<tool>`` (Python virtual environment)
    2. ``<project_dir>/node_modules/.bin/<tool>`` (Node.js local packages)
    3. ``shutil.which(<tool>)`` (system PATH)
    4. None — tool not found

    Atlas NEVER installs packages; it only informs when a tool is missing.
    """
    candidates = [
        os.path.join(project_dir, ".venv", "bin", tool_name),
        os.path.join(project_dir, "node_modules", ".bin", tool_name),
    ]
    for path in candidates:
        if os.path.isfile(path) and os.access(path, os.X_OK):
            return path

    return shutil.which(tool_name)

run_task 

run_task(
    task_name: str,
    command: str,
    project_dir: str,
    timeout: int = 60,
) -> dict

Execute command as a shell command in project_dir.

Returns ok_result(task=task_name, output=..., returncode=...) on completion (even when the command exits non-zero, so the caller can decide how to handle failures).

Returns error_result when: - command is empty - The executable is not found (exit code 127) - The command times out (exit code 124)

Source code in src/atlas/core/runner.py 
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
def run_task(
    task_name: str,
    command: str,
    project_dir: str,
    timeout: int = 60,
) -> dict:
    """Execute *command* as a shell command in *project_dir*.

    Returns ``ok_result(task=task_name, output=..., returncode=...)`` on
    completion (even when the command exits non-zero, so the caller can
    decide how to handle failures).

    Returns ``error_result`` when:
    - *command* is empty
    - The executable is not found (exit code 127)
    - The command times out (exit code 124)
    """
    if not command or not command.strip():
        return error_result("INVALID_ARGUMENT", f"No command for task '{task_name}'")

    try:
        args = shlex.split(command)
    except ValueError as exc:
        return error_result("INVALID_ARGUMENT", f"Cannot parse command: {exc}")

    try:
        proc = subprocess.run(
            args,
            cwd=project_dir,
            capture_output=True,
            text=True,
            timeout=timeout,
            check=False,
        )
        output = (proc.stdout + proc.stderr).strip()
        return ok_result(task=task_name, output=output, returncode=proc.returncode)
    except FileNotFoundError:
        return error_result(
            "INVALID_ARGUMENT",
            f"Executable not found for task '{task_name}': {args[0]}",
        )
    except subprocess.TimeoutExpired:
        return error_result(
            "INVALID_ARGUMENT",
            f"Task '{task_name}' timed out after {timeout}s",
        )

find_rule_hint 

find_rule_hint(code: str, module_rules: dict) -> str

Return the rule hint for code from module_rules, or empty string.

Source code in src/atlas/core/runner.py 
84
85
86
def find_rule_hint(code: str, module_rules: dict) -> str:
    """Return the rule hint for *code* from *module_rules*, or empty string."""
    return module_rules.get(code, "")

augment_errors 

augment_errors(output: str, module_rules: dict) -> str

Scan tool output for error codes and append relevant rule hints inline.

Looks for patterns like E501, F401, W291 (letter + 3-4 digits). If a matching rule hint exists in module_rules, appends it on the next line with a 📎 prefix.

Source code in src/atlas/core/runner.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
def augment_errors(output: str, module_rules: dict) -> str:
    """Scan tool output for error codes and append relevant rule hints inline.

    Looks for patterns like E501, F401, W291 (letter + 3-4 digits).
    If a matching rule hint exists in *module_rules*, appends it on the
    next line with a 📎 prefix.
    """
    if not module_rules:
        return output
    lines = output.split("\n")
    augmented: list[str] = []
    for line in lines:
        augmented.append(line)
        code_match = re.search(r'\b([A-Z]\d{3,4})\b', line)
        if code_match:
            hint = find_rule_hint(code_match.group(1), module_rules)
            if hint:
                augmented.append(f"    \U0001f4ce {hint}")
    return "\n".join(augmented)