@@ -7,3 +7,12 @@
 RUNNER_TOKEN = "DATATORCH_RUNNER_TOKEN"
 API_URL = "DATATORCH_API_URL"
 CONFIG_DIR = "DATATORCH_DIR"
+# The step's place in the record, handed to every action a runner runs
+# (0.8.1): the project the run belongs to, the run, the pipeline and the
+# step. An action that works "for this project" reads PROJECT_ID instead
+# of carrying a project input, so a pipeline forked into another project
+# works there unchanged.
+PROJECT_ID = "DATATORCH_PROJECT_ID"
+RUN_ID = "DATATORCH_RUN_ID"
+PIPELINE_ID = "DATATORCH_PIPELINE_ID"
+STEP_ID = "DATATORCH_STEP_ID"
@@ -91,7 +91,9 @@ def get_command(self, key: str, default=None):
     def runner_env(self) -> dict:
         """What an action needs to reach the server this runner is on:
         the API url and the runner's own token (as DATATORCH_API_URL and
-        DATATORCH_RUNNER_TOKEN). The SDK's client picks them up, so an
+        DATATORCH_RUNNER_TOKEN), and where the step sits in the record
+        (DATATORCH_PROJECT_ID, DATATORCH_RUN_ID, DATATORCH_PIPELINE_ID,
+        DATATORCH_STEP_ID). The SDK's client picks them up, so an
         action talks to the same server as the runner, as the runner,
         instead of the machine's stored login. Anything already set in
         the environment wins (DATATORCH_API_KEY included, since the SDK
@@ -105,6 +107,21 @@ def runner_env(self) -> dict:
             values[env_names.API_URL] = settings.api_url
         if settings.runner_token and not os.environ.get(env_names.RUNNER_TOKEN):
             values[env_names.RUNNER_TOKEN] = settings.runner_token
+        # The step's place in the record (0.8.1), so an action can work
+        # for "this project" without a project input. Always set by the
+        # runner, never inherited: a value left in the environment by an
+        # outer process must not describe another run.
+        sections = getattr(getattr(self, "variables", None), "variables", None) or {}
+        run = sections.get("run") or {}
+        step = sections.get("step") or {}
+        for name, value in (
+            (env_names.PROJECT_ID, run.get("projectId")),
+            (env_names.RUN_ID, run.get("id")),
+            (env_names.PIPELINE_ID, run.get("pipelineId")),
+            (env_names.STEP_ID, step.get("id")),
+        ):
+            if value:
+                values[name] = str(value)
         return values
 
     def input_env(self) -> dict:
 
@@ -32,6 +32,10 @@ def from_dispatch(cls, dispatch: "AgentStepDispatch", api: "AgentApiClient" = No
             name=dispatch.get("stepName") or "",
             inputs=dispatch.get("input") or {},
             run_id=dispatch.get("runId"),
+            run_number=dispatch.get("runNumber"),
+            project_id=dispatch.get("projectId"),
+            pipeline_id=dispatch.get("pipelineId"),
+            job_id=dispatch.get("jobId"),
             api=api,
         )
 
@@ -54,6 +58,10 @@ def __init__(
         cacheable: Union[bool, None] = None,
         inputs: dict = {},
         run_id: str = None,
+        run_number: int = None,
+        project_id: str = None,
+        pipeline_id: str = None,
+        job_id: str = None,
         api: "AgentApiClient" = None,
     ):
         self._action = ActionConfig(action)
@@ -63,6 +71,10 @@ def __init__(
         self.inputs = inputs
         self.cacheable = cacheable
         self.run_id = run_id
+        self.run_number = run_number
+        self.project_id = project_id
+        self.pipeline_id = pipeline_id
+        self.job_id = job_id
         self.api = api
         self.rendered_inputs: Optional[dict] = None
         self.logger = logging.getLogger(f"datatorch.runner.[{self._action.full_name}]")
 
@@ -86,6 +86,8 @@ class Variables(object):
 
     - ``agent``, ``directory``, ``machine``, ``python`` — machine facts
     - ``step``, ``action`` — the step being executed
+    - ``run`` — the run the step belongs to (id, number, projectId,
+      pipelineId, jobId); empty values in local mode
     - ``input`` / ``variable`` — the step's own (already-resolved) inputs,
       for use inside the action's own templates (``runs.command`` etc.).
       ``variable`` is a legacy alias of ``input``; both are scoped to the
@@ -99,6 +101,19 @@ def __init__(self):
 
     def set_step(self, step: "Step"):
         self.set("step", {"id": step.id, "name": step.name})
+        # Where the step sits in the record (server mode; empty when run
+        # locally). `${{ run.projectId }}` in an action's own templates, and
+        # DATATORCH_PROJECT_ID / DATATORCH_RUN_ID in its environment.
+        self.set(
+            "run",
+            {
+                "id": getattr(step, "run_id", None),
+                "number": getattr(step, "run_number", None),
+                "projectId": getattr(step, "project_id", None),
+                "pipelineId": getattr(step, "pipeline_id", None),
+                "jobId": getattr(step, "job_id", None),
+            },
+        )
 
     def set_action(self, action: "Action"):
         self.set(
 
@@ -33,7 +33,7 @@
 
 setup(
     name="datatorch",
-    version="0.8.0",
+    version="0.8.1",
     description="A CLI and library for interacting with DataTorch.",
     author="DataTorch",
     author_email="support@datatorch.io",
 
@@ -0,0 +1,63 @@
+"""A dispatched step tells its action where it sits in the record (0.8.1):
+`${{ run.* }}` in the action's templates and DATATORCH_PROJECT_ID /
+DATATORCH_RUN_ID / DATATORCH_PIPELINE_ID / DATATORCH_STEP_ID in the
+process environment of whatever the action runs."""
+import os
+
+from datatorch.core import env as env_names
+from datatorch.runner.pipelines.step.step import Step
+from datatorch.runner.pipelines.template import Variables
+from datatorch.runner.pipelines.runner.runner import Runner
+
+
+DISPATCH = {
+    "stepId": "step-1",
+    "runnerId": "runner-1",
+    "action": "datatorch-actions/claude-chat@v4",
+    "stepName": "Turn",
+    "stepIndex": 0,
+    "input": {"message": "hi"},
+    "jobId": "job-1",
+    "jobName": "chat",
+    "runId": "run-1",
+    "runNumber": 7,
+    "pipelineId": "pipe-1",
+    "projectId": "proj-1",
+}
+
+
+def test_run_namespace_from_dispatch():
+    step = Step.from_dispatch(DISPATCH)
+    variables = Variables()
+    variables.set_step(step)
+    assert variables.render("${{ run.projectId }}") == "proj-1"
+    assert variables.render("${{ run.id }}/${{ run.number }}") == "run-1/7"
+    assert variables.render("${{ run.pipelineId }}") == "pipe-1"
+    assert variables.render("${{ step.id }}") == "step-1"
+
+
+def test_run_namespace_empty_in_local_mode():
+    step = Step.from_dict({"id": "local", "action": "x/y@v1", "name": "n"})
+    variables = Variables()
+    variables.set_step(step)
+    assert variables.variables["run"]["projectId"] is None
+
+
+def test_runner_env_carries_identity(monkeypatch):
+    for name in (env_names.PROJECT_ID, env_names.RUN_ID, env_names.PIPELINE_ID, env_names.STEP_ID):
+        monkeypatch.setenv(name, "stale")  # never inherited from the outer process
+    step = Step.from_dispatch(DISPATCH)
+    variables = Variables()
+    variables.set_step(step)
+    runner = Runner({}, action=None)
+    runner.variables = variables
+    env = runner.runner_env()
+    assert env[env_names.PROJECT_ID] == "proj-1"
+    assert env[env_names.RUN_ID] == "run-1"
+    assert env[env_names.PIPELINE_ID] == "pipe-1"
+    assert env[env_names.STEP_ID] == "step-1"
+
+
+def test_runner_env_without_variables_sets_nothing():
+    env = Runner({}, action=None).runner_env()
+    assert env_names.PROJECT_ID not in env