Files
github-spec-kit/src/specify_cli/workflows/expressions.py
Huy Do c6afe4cde1 feat(workflows): expose {{ context.run_id }} template variable (#2664)
* feat(workflows): expose `{{ context.run_id }}` template variable

Closes #2590.

Surfaces the engine-assigned run id (the same 8-character hex
string Spec Kit prints as `Run ID:` at the end of
`workflow run`) as a workflow template variable so YAML
authors can reference it from shell `run:`, command
`input.args:`, switch `expression:`, and any other field that
already evaluates `{{ ... }}` templates.

### Why

The run id is the natural join key between a Spec Kit workflow
run and downstream artifacts, telemetry, or per-run scratch
state. Today the operator sees it in stdout but workflows
themselves cannot reference it — there was no way to stamp a
log line, name a scratch directory, or tag an artifact with
the same id Spec Kit assigned.

The three motivating use cases from the issue:

1. Telemetry / observability — stamp logs and events with the
   run id so external systems can join workflow runs to
   downstream artifacts.
2. Per-run scratch / isolation — interactive operator commands
   that need their own state directory under
   `/tmp/run-<id>/`.
3. Run-id in artifact metadata — stable join key from artifact
   back to the producing run.

### Implementation

`StepContext.run_id` is already populated by `WorkflowEngine`
in both `execute()` and `resume()`. The only gap was the
template namespace builder.

`_build_namespace` (in `workflows/expressions.py`) now adds a
`context` key alongside the existing `inputs`, `steps`,
`item`, and `fan_in` namespaces:

```python
ns["context"] = {"run_id": run_id}
```

The value is always present (even outside a run) and falls
back to an empty string when no run is active. Workflows
referencing `{{ context.run_id }}` therefore never error — a
hard requirement from the issue's acceptance criteria for
dry-run, validation, and ad-hoc evaluator usage.

### Default behaviour preserved

Workflows that do not reference `{{ context.run_id }}` are
byte-equivalent to before this change. The `context`
namespace is added unconditionally to keep template
resolution branch-free, but its presence has no observable
effect when nothing references it.

### Tests

`TestExpressions` (unit-level) gains three tests:

- `test_context_run_id_resolves` — direct lookup against a
  `StepContext(run_id=...)`.
- `test_context_run_id_defaults_to_empty_when_unset` —
  graceful default outside a run context.
- `test_context_run_id_string_interpolation` — mixed
  template (e.g. `"RUN_ID={{ context.run_id }}"`).

`TestContextRunId` (end-to-end) covers the three step types
the acceptance criteria called out:

- `test_shell_run_resolves_run_id` — `run:` field
  substitution, verified via captured stdout.
- `test_command_input_args_resolves_run_id` — `input.args:`
  resolution, captured in step output even when CLI dispatch
  is unavailable (the artifact-metadata use case).
- `test_switch_expression_matches_on_run_id` — switch
  matches against the resolved value, proving the run id is a
  first-class value in the expression engine, not just an
  interpolation token.
- `test_workflow_without_context_reference_unchanged` —
  locks the byte-equivalent default required by the issue.

### Docs

`workflows/README.md` gains a "Runtime Context" subsection
under "Expressions" documenting the new namespace and the
three canonical use patterns (telemetry, per-run scratch,
artifact metadata).

* test(workflows): drop inline double-quotes in run_id shell tests

`test_shell_run_resolves_run_id` and
`test_switch_expression_matches_on_run_id` used
`run: 'echo "RUN_ID={{ context.run_id }}"'` with inner double-quotes
around the echo argument. Bash/sh strips those quotes before invoking
echo, but cmd.exe (used on Windows when `shell=True`) treats them
as literal characters and emits `"RUN_ID=abc12345"` — failing the
exact-match assertion. Linux passed; all three Windows-latest matrix
entries failed with `assert '"RUN_ID=abc12345"' == 'RUN_ID=abc12345'`.

Resolve by dropping the inner double-quotes (the value has no spaces
or shell metacharacters) and wrapping the YAML scalar in plain
double-quotes the same way other shell-step tests in this file do
(e.g. `run: "echo b-saw-..."`). Behaviour-equivalent on POSIX,
portable to cmd.exe.
2026-05-27 13:00:58 -05:00

310 lines
10 KiB
Python

"""Sandboxed expression evaluator for workflow templates.
Provides a safe Jinja2 subset for evaluating expressions in workflow YAML.
No file I/O, no imports, no arbitrary code execution.
"""
from __future__ import annotations
import re
from typing import Any
# -- Custom filters -------------------------------------------------------
def _filter_default(value: Any, default_value: Any = "") -> Any:
"""Return *default_value* when *value* is ``None`` or empty string."""
if value is None or value == "":
return default_value
return value
def _filter_join(value: Any, separator: str = ", ") -> str:
"""Join a list into a string with *separator*."""
if isinstance(value, list):
return separator.join(str(v) for v in value)
return str(value)
def _filter_map(value: Any, attr: str) -> list[Any]:
"""Map a list of dicts to a specific attribute."""
if isinstance(value, list):
result = []
for item in value:
if isinstance(item, dict):
# Support dot notation: "result.status" → item["result"]["status"]
parts = attr.split(".")
v = item
for part in parts:
if isinstance(v, dict):
v = v.get(part)
else:
v = None
break
result.append(v)
else:
result.append(item)
return result
return []
def _filter_contains(value: Any, substring: str) -> bool:
"""Check if a string or list contains *substring*."""
if isinstance(value, str):
return substring in value
if isinstance(value, list):
return substring in value
return False
# -- Expression resolution ------------------------------------------------
_EXPR_PATTERN = re.compile(r"\{\{(.+?)\}\}")
def _resolve_dot_path(obj: Any, path: str) -> Any:
"""Resolve a dotted path like ``steps.specify.output.file`` against *obj*.
Supports dict key access and list indexing (e.g., ``task_list[0]``).
"""
parts = path.split(".")
current = obj
for part in parts:
# Handle list indexing: name[0]
idx_match = re.match(r"^([\w-]+)\[(\d+)\]$", part)
if idx_match:
key, idx = idx_match.group(1), int(idx_match.group(2))
if isinstance(current, dict):
current = current.get(key)
else:
return None
if isinstance(current, list) and 0 <= idx < len(current):
current = current[idx]
else:
return None
elif isinstance(current, dict):
current = current.get(part)
else:
return None
if current is None:
return None
return current
def _build_namespace(context: Any) -> dict[str, Any]:
"""Build the variable namespace from a StepContext."""
ns: dict[str, Any] = {}
if hasattr(context, "inputs"):
ns["inputs"] = context.inputs or {}
if hasattr(context, "steps"):
ns["steps"] = context.steps or {}
if hasattr(context, "item"):
ns["item"] = context.item
if hasattr(context, "fan_in"):
ns["fan_in"] = context.fan_in or {}
# Engine-managed runtime metadata. Always present (even outside a
# run) so templates referencing it never error: `run_id` falls back
# to an empty string when no run is active (dry-run, validation,
# ad-hoc evaluator usage). The value is the same one Spec Kit
# prints as `Run ID:` at the end of `workflow run` — auto-generated
# 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}
return ns
def _evaluate_simple_expression(expr: str, namespace: dict[str, Any]) -> Any:
"""Evaluate a simple expression against the namespace.
Supports:
- Dot-path access: ``steps.specify.output.file``
- Comparisons: ``==``, ``!=``, ``>``, ``<``, ``>=``, ``<=``
- Boolean operators: ``and``, ``or``, ``not``
- ``in``, ``not in``
- Pipe filters: ``| default('...')``, ``| join(', ')``, ``| contains('...')``, ``| map('...')``
- String and numeric literals
"""
expr = expr.strip()
# String literal — check before pipes and operators so quoted strings
# containing | or operator keywords are not mis-parsed.
if (expr.startswith("'") and expr.endswith("'")) or (
expr.startswith('"') and expr.endswith('"')
):
return expr[1:-1]
# Handle pipe filters
if "|" in expr:
parts = expr.split("|", 1)
value = _evaluate_simple_expression(parts[0].strip(), namespace)
filter_expr = parts[1].strip()
# Parse filter name and argument
filter_match = re.match(r"(\w+)\((.+)\)", filter_expr)
if filter_match:
fname = filter_match.group(1)
farg = _evaluate_simple_expression(filter_match.group(2).strip(), namespace)
if fname == "default":
return _filter_default(value, farg)
if fname == "join":
return _filter_join(value, farg)
if fname == "map":
return _filter_map(value, farg)
if fname == "contains":
return _filter_contains(value, farg)
# Filter without args
filter_name = filter_expr.strip()
if filter_name == "default":
return _filter_default(value)
return value
# Boolean operators — parse 'or' first (lower precedence) so that
# 'a or b and c' is evaluated as 'a or (b and c)'.
if " or " in expr:
parts = expr.split(" or ", 1)
left = _evaluate_simple_expression(parts[0].strip(), namespace)
right = _evaluate_simple_expression(parts[1].strip(), namespace)
return bool(left) or bool(right)
if " and " in expr:
parts = expr.split(" and ", 1)
left = _evaluate_simple_expression(parts[0].strip(), namespace)
right = _evaluate_simple_expression(parts[1].strip(), namespace)
return bool(left) and bool(right)
if expr.startswith("not "):
inner = _evaluate_simple_expression(expr[4:].strip(), namespace)
return not bool(inner)
# Comparison operators (order matters — check multi-char ops first)
for op in ("!=", "==", ">=", "<=", ">", "<", " not in ", " in "):
if op in expr:
parts = expr.split(op, 1)
left = _evaluate_simple_expression(parts[0].strip(), namespace)
right = _evaluate_simple_expression(parts[1].strip(), namespace)
if op == "==":
return left == right
if op == "!=":
return left != right
if op == ">":
return _safe_compare(left, right, ">")
if op == "<":
return _safe_compare(left, right, "<")
if op == ">=":
return _safe_compare(left, right, ">=")
if op == "<=":
return _safe_compare(left, right, "<=")
if op == " in ":
return left in right if right is not None else False
if op == " not in ":
return left not in right if right is not None else True
# Numeric literal
try:
if "." in expr:
return float(expr)
return int(expr)
except (ValueError, TypeError):
pass
# Boolean literal
if expr.lower() == "true":
return True
if expr.lower() == "false":
return False
# Null
if expr.lower() in ("none", "null"):
return None
# List literal (simple)
if expr.startswith("[") and expr.endswith("]"):
inner = expr[1:-1].strip()
if not inner:
return []
items = [_evaluate_simple_expression(i.strip(), namespace) for i in inner.split(",")]
return items
# Variable reference (dot-path)
return _resolve_dot_path(namespace, expr)
def _safe_compare(left: Any, right: Any, op: str) -> bool:
"""Safely compare two values, coercing types when possible."""
try:
if isinstance(left, str):
left = float(left) if "." in left else int(left)
if isinstance(right, str):
right = float(right) if "." in right else int(right)
except (ValueError, TypeError):
return False
try:
if op == ">":
return left > right # type: ignore[operator]
if op == "<":
return left < right # type: ignore[operator]
if op == ">=":
return left >= right # type: ignore[operator]
if op == "<=":
return left <= right # type: ignore[operator]
except TypeError:
return False
return False
def evaluate_expression(template: str, context: Any) -> Any:
"""Evaluate a template string with ``{{ ... }}`` expressions.
If the entire string is a single expression, returns the raw value
(preserving type). Otherwise, substitutes each expression inline
and returns a string.
Parameters
----------
template:
The template string (e.g., ``"{{ steps.plan.output.task_count }}"``
or ``"Processed {{ inputs.spec }}"``.
context:
A ``StepContext`` or compatible object.
Returns
-------
The resolved value (any type for single-expression templates,
string for multi-expression or mixed templates).
"""
if not isinstance(template, str):
return template
namespace = _build_namespace(context)
# Single expression: return typed value
match = _EXPR_PATTERN.fullmatch(template.strip())
if match:
return _evaluate_simple_expression(match.group(1).strip(), namespace)
# Multi-expression: string interpolation
def _replacer(m: re.Match[str]) -> str:
val = _evaluate_simple_expression(m.group(1).strip(), namespace)
return str(val) if val is not None else ""
return _EXPR_PATTERN.sub(_replacer, template)
def evaluate_condition(condition: str, context: Any) -> bool:
"""Evaluate a condition expression and return a boolean.
Convenience wrapper around ``evaluate_expression`` that coerces
the result to bool.
"""
result = evaluate_expression(condition, context)
# Treat plain "false"/"true" strings as booleans so that
# condition: "false" (without {{ }}) behaves as expected.
if isinstance(result, str):
lower = result.lower()
if lower == "false":
return False
if lower == "true":
return True
return bool(result)