feat(workflows): expose workflow source directory to steps

docs/reference/workflows.md

@@ -282,6 +282,8 @@ Steps can reference inputs and previous step outputs using `{{ expression }}` sy
 | `inputs.spec`                  | Workflow input values                |
 | `steps.specify.output.file`    | Output from a previous step          |
 | `item`                         | Current item in a fan-out iteration  |
+| `context.run_id`               | Current workflow run ID              |
+| `context.workflow_dir`         | Resolved absolute path to the workflow source directory. Empty string for string-loaded workflows. |
 
 Available filters: `default`, `join`, `contains`, `map`, `from_json`.
 
@@ -293,6 +295,14 @@ args: "{{ inputs.spec }}"
 message: "{{ status | default('pending') }}"
 ```
 
+## Shell Step Environment Variables
+
+Shell steps automatically receive the following environment variables:
+
+| Variable | Description |
+| -------- | ----------- |
+| `SPECKIT_WORKFLOW_DIR` | Resolved absolute path to the workflow source directory (same value as `{{ context.workflow_dir }}`). Not set when the workflow has no source path. |
+
 ## Input Types
 
 | Type      | Coercion                                          |

src/specify_cli/workflows/base.py

@@ -74,6 +74,9 @@ class StepContext:
     #: Current run ID.
     run_id: str | None = None
 
+    #: Source directory of the workflow definition file.
+    workflow_dir: str | None = None
+
 
 @dataclass
 class StepResult:

src/specify_cli/workflows/engine.py

@@ -455,6 +455,7 @@ def __init__(
         # append_log is never called while _lock is held, the two never nest.
         self._log_lock = threading.Lock()
         self.inputs: dict[str, Any] = {}
+        self.workflow_dir: str | None = None
         self.created_at = datetime.now(timezone.utc).isoformat()
         self.updated_at = self.created_at
         self.log_entries: list[dict[str, Any]] = []
@@ -507,6 +508,7 @@ def save(self) -> None:
                 "current_step_index": self.current_step_index,
                 "current_step_id": self.current_step_id,
                 "step_results": self.step_results,
+                "workflow_dir": self.workflow_dir,
                 "created_at": self.created_at,
                 "updated_at": self.updated_at,
             }
@@ -564,6 +566,7 @@ def load(cls, run_id: str, project_root: Path) -> RunState:
         state.current_step_index = state_data.get("current_step_index", 0)
         state.current_step_id = state_data.get("current_step_id")
         state.step_results = state_data.get("step_results", {})
+        state.workflow_dir = state_data.get("workflow_dir")
         state.created_at = state_data.get("created_at", "")
         state.updated_at = state_data.get("updated_at", "")
 
@@ -697,6 +700,12 @@ def execute(
         # Resolve inputs
         resolved_inputs = self._resolve_inputs(definition, inputs or {})
         state.inputs = resolved_inputs
+        workflow_dir = (
+            str(definition.source_path.resolve().parent)
+            if definition.source_path is not None
+            else None
+        )
+        state.workflow_dir = workflow_dir
         state.status = RunStatus.RUNNING
         state.save()
 
@@ -707,6 +716,7 @@ def execute(
             default_options=definition.default_options,
             project_root=str(self.project_root),
             run_id=state.run_id,
+            workflow_dir=workflow_dir,
         )
 
         # Execute steps
@@ -772,6 +782,7 @@ def resume(
             default_options=definition.default_options,
             project_root=str(self.project_root),
             run_id=state.run_id,
+            workflow_dir=state.workflow_dir,
         )
 
         from . import STEP_REGISTRY

src/specify_cli/workflows/expressions.py

@@ -142,7 +142,8 @@ def _build_namespace(context: Any) -> dict[str, Any]:
     # runs use an 8-character uuid4 hex; operator-supplied ids may be
     # any alphanumeric string with hyphens or underscores.
     run_id = getattr(context, "run_id", None) or ""
-    ns["context"] = {"run_id": run_id}
+    workflow_dir = getattr(context, "workflow_dir", None) or ""
+    ns["context"] = {"run_id": run_id, "workflow_dir": workflow_dir}
     return ns
 
 

src/specify_cli/workflows/steps/shell/__init__.py

@@ -4,6 +4,7 @@
 
 import json
 import math
+import os
 import subprocess
 from typing import Any
 
@@ -40,6 +41,13 @@ def execute(self, config: dict[str, Any], context: StepContext) -> StepResult:
                 error=timeout_error,
                 output={"exit_code": -1, "stdout": "", "stderr": "invalid timeout"},
             )
+
+        env = {**os.environ}
+        if context.workflow_dir:
+            env["SPECKIT_WORKFLOW_DIR"] = context.workflow_dir
+        else:
+            env.pop("SPECKIT_WORKFLOW_DIR", None)
+
         # NOTE: shell=True is required to support pipes, redirects, and
         # multi-command expressions in workflow YAML.  Workflow authors
         # control commands; catalog-installed workflows should be reviewed
@@ -51,6 +59,7 @@ def execute(self, config: dict[str, Any], context: StepContext) -> StepResult:
                 capture_output=True,
                 text=True,
                 cwd=cwd,
+                env=env,
                 timeout=timeout,
             )
             output = {