Method load

src/specify_cli/workflows/engine.py:505–547

Load a run state from disk. Validates ``run_id`` against ``_RUN_ID_PATTERN`` *before* building the lookup path. Without this guard, a caller passing a value like ``../escape`` (e.g. via ``specify workflow resume`` CLI argument) would interpolate path-traversal segmen

(cls, run_id: str, project_root: Path)

    @classmethod
    def load(cls, run_id: str, project_root: Path) -> RunState:
        """Load a run state from disk.

        Validates ``run_id`` against ``_RUN_ID_PATTERN`` *before* building
        the lookup path. Without this guard, a caller passing a value like
        ``../escape`` (e.g. via ``specify workflow resume`` CLI argument)
        would interpolate path-traversal segments into
        ``runs_dir`` below, letting ``state_path.exists()`` probe arbitrary
        paths and ``json.load`` read attacker-planted JSON from outside
        the project's ``runs/`` directory. ``__init__`` already runs this
        check on the stored ``state_data["run_id"]``, but that fires
        *after* the file lookup — too late to prevent the disclosure.
        Mirrors the precedent in ``agents._ensure_within_directory``.
        """
        cls._validate_run_id(run_id)
        runs_dir = project_root / ".specify" / "workflows" / "runs" / run_id
        state_path = runs_dir / "state.json"
        if not state_path.exists():
            msg = f"Run state not found: {state_path}"
            raise FileNotFoundError(msg)

        with open(state_path, encoding="utf-8") as f:
            state_data = json.load(f)

        state = cls(
            run_id=state_data["run_id"],
            workflow_id=state_data["workflow_id"],
            project_root=project_root,
        )
        state.status = RunStatus(state_data["status"])
        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.created_at = state_data.get("created_at", "")
        state.updated_at = state_data.get("updated_at", "")

        inputs_path = runs_dir / "inputs.json"
        if inputs_path.exists():
            with open(inputs_path, encoding="utf-8") as f:
                inputs_data = json.load(f)
            state.inputs = inputs_data.get("inputs", {})

        return state

    def append_log(self, entry: dict[str, Any]) -> None:
        """Append a log entry to the run log.