0.7.3

Author: ninthreezy

datatorch/agent/client.py

@@ -15,14 +15,14 @@ class Log(TypedDict):
 
 
 class AgentStepDispatch(TypedDict):
-    """A ready step pushed to this agent (agent protocol v2).
+    """A ready step pushed to this runner (runner protocol v2).
 
     ``input`` arrives server-resolved: ``steps.*`` and ``input.*``
     references are already substituted with concrete values.
     """
 
     stepId: str
-    agentId: str
+    runnerId: str
     action: str
     stepName: Optional[str]
     stepIndex: int
@@ -40,10 +40,10 @@ class AgentStepRequest(TypedDict):
 
 
 class AgentStepCancel(TypedDict):
-    """A stop signal for a step this agent is currently running."""
+    """A stop signal for a step this runner is currently running."""
 
     stepId: str
-    agentId: str
+    runnerId: str
 
 
 class AgentStepCancelRequest(TypedDict):
@@ -52,7 +52,7 @@ class AgentStepCancelRequest(TypedDict):
 
 STEP_DISPATCH_FIELDS = """
     stepId
-    agentId
+    runnerId
     action
     stepName
     stepIndex
@@ -71,11 +71,11 @@ def __init__(self, session: AsyncClientSession):
         self.session = session
 
     def agent_steps(self):
-        """Subscribe to ready steps dispatched to this agent."""
+        """Subscribe to ready steps dispatched to this runner."""
         # fmt: off
         sub = gql("""
             subscription {
-                step: agentSteps {
+                step: runnerSteps {
                     %s
                 }
             }
@@ -84,17 +84,17 @@ def agent_steps(self):
         return cast(AsyncGenerator[AgentStepRequest, None], self.session.subscribe(sub))
 
     def agent_step_cancels(self):
-        """Subscribe to stop signals for steps this agent is running.
+        """Subscribe to stop signals for steps this runner is running.
 
         Best effort: the server has already moved the step terminal, so a
         missed signal only means the step runs to harmless completion.
         """
         # fmt: off
         sub = gql("""
             subscription {
-                cancel: agentStepCancels {
+                cancel: runnerStepCancels {
                     stepId
-                    agentId
+                    runnerId
                 }
             }
         """)
@@ -105,15 +105,15 @@ def agent_step_cancels(self):
         )
 
     async def pending_steps(self) -> List[AgentStepDispatch]:
-        """Steps dispatched to this agent but not yet claimed.
+        """Steps dispatched to this runner but not yet claimed.
 
         Called on (re)connect: pushes missed while offline are recovered
         here.
         """
         # fmt: off
         query = """
             query PendingSteps {
-                steps: agentPendingSteps {
+                steps: runnerPendingSteps {
                     %s
                 }
             }
@@ -163,7 +163,7 @@ async def complete_step(
     def initial_metrics(self, metrics):
         # fmt: off
         mutate = """
-            mutation updateAgent(
+            mutation updateRunner(
                 $id: ID!
                 $version: String
                 $os: String
@@ -177,7 +177,7 @@ def initial_metrics(self, metrics):
                 $cpuCoresPhysical: Int
                 $cpuCoresLogical: Int
             ) {
-                updateAgent(id: $id, input: {
+                updateRunner(id: $id, input: {
                     version: $version
                     os: $os
                     osRelease: $osRelease
@@ -201,8 +201,8 @@ def initial_metrics(self, metrics):
     def metrics(self, metrics):
         # fmt: off
         mutate = """
-            mutation updateAgent(
-                $agentId: ID!
+            mutation createRunnerMetric(
+                $runnerId: ID!
                 $sampledAt: DateTime!
                 $avgLoad1: Float
                 $avgLoad5: Float
@@ -211,8 +211,8 @@ def metrics(self, metrics):
                 $diskUsage: Float
                 $memoryUsage: Float
             ) {
-                createAgentMetric(input: {
-                    agentId: $agentId
+                createRunnerMetric(input: {
+                    runnerId: $runnerId
                     sampledAt: $sampledAt
                     avgLoad1: $avgLoad1
                     avgLoad5: $avgLoad5
@@ -226,7 +226,7 @@ def metrics(self, metrics):
             }
         """
         # fmt: on
-        params = {"agentId": agent_directory.settings.agent_id, **metrics}
+        params = {"runnerId": agent_directory.settings.agent_id, **metrics}
         return self.execute(mutate, params=params)
 
     def upload_step_logs(self, step_id: str, logs: List[Log]):

datatorch/agent/pipelines/runner/python.py

@@ -1,7 +1,7 @@
 import os
 import sys
 import json
-import base64
+import shlex
 
 from .runner import Runner
 
@@ -18,13 +18,15 @@ def __init__(self, *args, **kwargs):
 
     async def execute(self):
         main = self.get("main").strip("/")
-        # Ok this command parsing stuff is to handle Application Support folder in Macs
-        # Because there are spaces in the folder name
-        main_command_spaces = os.path.join(self.action.dir, main)
-        main_command_nospaces = main_command_spaces.replace(" ", "\ ")
+        main_path = os.path.join(self.action.dir, main)
 
-        json_input = json.dumps(self.variables.inputs).replace("'", '\\"')
+        # Inputs reach the action verbatim as one JSON argv argument.
+        # shlex.quote covers every shell metacharacter, apostrophes
+        # included, and the space in macOS "Application Support" paths.
+        json_input = json.dumps(self.variables.inputs)
 
         await self.monitor_cmd(
-            f"{sys.executable} {main_command_nospaces} '{json_input}'"
+            " ".join(
+                shlex.quote(part) for part in (sys.executable, main_path, json_input)
+            )
         )

datatorch/agent/pipelines/runner/shell.py

@@ -1,4 +1,5 @@
 import os
+import shlex
 
 from .runner import Runner
 
@@ -18,6 +19,12 @@ async def execute(self):
         # ${{ input.* }} — injection policy). Inputs reach the script as
         # $INPUT_<NAME> environment variables.
         script = self.get_command("script").strip("/")
-        script_command = os.path.join(self.action.dir, script)
-        await self.run_cmd("chmod +x {}".format(script_command.split(" ", 1)[0]))
-        await self.monitor_cmd(script_command, env=self.input_env())
+        # `script` may carry arguments after the path ("run.sh --flag");
+        # only the path is joined to the action dir and quoted, since the
+        # dir can contain a space (macOS "Application Support").
+        script_path, _, script_args = script.partition(" ")
+        quoted_path = shlex.quote(os.path.join(self.action.dir, script_path))
+        await self.run_cmd(f"chmod +x {quoted_path}")
+        await self.monitor_cmd(
+            f"{quoted_path} {script_args}".rstrip(), env=self.input_env()
+        )

datatorch/api/client.py

@@ -20,7 +20,7 @@
 __all__ = "Client"
 
 
-AGENT_TOKEN_HEADER = "datatorch-agent-token"
+AGENT_TOKEN_HEADER = "datatorch-runner-token"
 API_KEY_HEADER = "datatorch-api-key"
 
 

datatorch/cli/agent/create.py

@@ -14,9 +14,9 @@ def create_agent(name: str) -> dict:
     # fmt: off
     results = api.execute(
         """
-        mutation CreateAgent($name: String!, $version: String!) {
-            createAgent(name: $name, version: $version) {
-                agent {
+        mutation CreateRunner($name: String!, $version: String!) {
+            createRunner(name: $name, version: $version) {
+                runner {
                     id
                 }
                 token
@@ -54,8 +54,8 @@ def create(ctx):
     spinner = Spinner("Creating runner")
 
     try:
-        agent = create_agent(name)["createAgent"]
-        agent_settings.agent_id = agent["agent"]["id"]
+        agent = create_agent(name)["createRunner"]
+        agent_settings.agent_id = agent["runner"]["id"]
         agent_settings.agent_token = agent["token"]
         agent_settings.api_url = user_settings.api_url
     except Exception as ex:

setup.py

@@ -33,7 +33,7 @@
 
 setup(
     name="datatorch",
-    version="0.7.2",
+    version="0.7.3",
     description="A CLI and library for interacting with DataTorch.",
     author="DataTorch",
     author_email="support@datatorch.io",

test/agents/pipelines/test_runner_quoting.py

@@ -0,0 +1,92 @@
+"""
+Inputs reach a python action as one JSON argv argument, and actions live
+under the agent directory, which on macOS is inside "Application Support".
+Both must survive the shell untouched: the runner used to hand-escape
+them, which turned every apostrophe in an input into a double quote and
+broke script paths containing a space.
+"""
+
+import asyncio
+import os
+import tempfile
+
+from datatorch.agent.pipelines.runner.python import PythonRunner
+from datatorch.agent.pipelines.runner.shell import ShellRunner
+
+
+class _FakeVariables:
+    def __init__(self, inputs):
+        self.inputs = inputs
+
+    def render(self, value):
+        return value
+
+    def render_command(self, value):
+        return value
+
+    def env_inputs(self):
+        return {f"INPUT_{k.upper()}": str(v) for k, v in self.inputs.items()}
+
+
+class _FakeAction:
+    step = None
+
+    def __init__(self, directory):
+        self.dir = directory
+
+
+def _action_dir() -> str:
+    # A space in the path, like ~/Library/Application Support/DataTorch.
+    path = os.path.join(tempfile.mkdtemp(prefix="datatorch-test-"), "App Support")
+    os.makedirs(path)
+    return path
+
+
+AWKWARD_INPUTS = {
+    "message": "what's up? it's \"quoted\", $HOME, `whoami`, $(id), a\\b, 100%\nline two\ttab",
+    "unicode": "café — 日本語 'single' \"double\"",
+    "empty": "",
+    "count": 3,
+    "items": ["it's", "fine"],
+}
+
+
+def test_python_runner_passes_inputs_verbatim():
+    directory = _action_dir()
+    with open(os.path.join(directory, "main.py"), "w") as f:
+        f.write(
+            "import json, sys\n"
+            "print('::echo::' + json.dumps(json.loads(sys.argv[-1])))\n"
+        )
+
+    runner = PythonRunner({"main": "main.py"}, _FakeAction(directory))
+    outputs = asyncio.run(runner.run(_FakeVariables(dict(AWKWARD_INPUTS))))
+
+    assert outputs["echo"] == AWKWARD_INPUTS
+
+
+def test_shell_runner_script_path_with_space():
+    directory = _action_dir()
+    with open(os.path.join(directory, "run.sh"), "w") as f:
+        f.write('#!/bin/sh\necho "::args::$#"\necho "::first::\\"$1\\""\n')
+
+    runner = ShellRunner({"script": "run.sh hello"}, _FakeAction(directory))
+    outputs = asyncio.run(runner.run(_FakeVariables({})))
+
+    assert outputs == {"args": 1, "first": "hello"}
+
+
+def test_shell_runner_inputs_arrive_as_env():
+    directory = _action_dir()
+    with open(os.path.join(directory, "run.sh"), "w") as f:
+        f.write(
+            "#!/bin/sh\n"
+            'printf \'::message::%s\\n\' "$(printf %s "$INPUT_MESSAGE" | '
+            "python3 -c 'import json,sys; print(json.dumps(sys.stdin.read()))')\"\n"
+        )
+
+    runner = ShellRunner({"script": "run.sh"}, _FakeAction(directory))
+    message = 'it\'s "fine" $HOME'
+    outputs = asyncio.run(runner.run(_FakeVariables({"message": message})))
+
+    assert outputs["message"] == message