Function init_run_env

src/harness/paths.py:129–175

Populate ``AGENTSPEX_RUN_*`` env vars and return ``(run_id, host_run_dir)``. This is the single point where per-run paths are decided. All entry points (``agentspex run``, ``agentspex cli``, ``run_agent.sh`` via Python) call this before YAML parsing so workflow authors can use ``${A

(
    task_name: str,
    override_run_id: Optional[str] = None,
    resume: bool = False,
)

def init_run_env(
    task_name: str,
    override_run_id: Optional[str] = None,
    resume: bool = False,
) -> tuple[str, Path]:
    """Populate ``AGENTSPEX_RUN_*`` env vars and return ``(run_id, host_run_dir)``.

    This is the single point where per-run paths are decided. All entry
    points (``agentspex run``, ``agentspex cli``, ``run_agent.sh`` via
    Python) call this before YAML parsing so workflow authors can use
    ``${AGENTSPEX_RUN_SANDBOX_DIR_ABS}`` in their YAML and it resolves to
    a run-scoped directory on both the host and the sandbox.

    Resolution order for the run id:
      1. ``override_run_id`` (caller-provided — used by tests or external schedulers).
      2. Existing ``AGENTSPEX_RUN_ID`` in the environment (idempotent when
         called multiple times in one process, or when a parent shell has
         already decided the id).
      3. On ``resume=True``, the dirname of the most recent matching run dir.
      4. Otherwise, a fresh ``<task_name>_<YYYYMMDD-HHMMSS>`` id.
    """
    ensure_runtime_dirs()

    existing = os.environ.get("AGENTSPEX_RUN_ID")
    if override_run_id:
        run_id = override_run_id
    elif existing:
        run_id = existing
    elif resume:
        latest = find_latest_run_dir(task_name)
        run_id = latest.name if latest is not None else (
            f"{task_name}_{datetime.now().strftime(RUN_TIMESTAMP_FMT)}"
        )
    else:
        run_id = f"{task_name}_{datetime.now().strftime(RUN_TIMESTAMP_FMT)}"

    sandbox_rel = f"outputs/{run_id}"
    sandbox_abs = f"/workspace/{sandbox_rel}"
    host_run = outputs_root() / run_id
    host_run.mkdir(parents=True, exist_ok=True)

    os.environ["AGENTSPEX_RUN_ID"] = run_id
    os.environ["AGENTSPEX_RUN_SANDBOX_DIR"] = sandbox_rel
    os.environ["AGENTSPEX_RUN_SANDBOX_DIR_ABS"] = sandbox_abs
    os.environ["AGENTSPEX_RUN_HOST_DIR"] = str(host_run)

    return run_id, host_run


def ensure_runtime_dirs() -> None: