README

def abort_start(self, error: BaseException) -> None:
    """Retain an incomplete stage when source or startup recording fails."""
    self._close_logging()
    if (
        self.store is not None
        and (self.store.root / "pending" / f"{self.run_id}.json").exists()
    ):
        self.store.append_event(
            "recording-incomplete", self.run_id, {"reason": str(error)}
        )
    self._assets.active = False
    self.active = False

def end(self, exc: BaseException | None = None) -> None:
    """Persist required local evidence; recording failures propagate."""
    if not self.active or self.store is None:
        msg = "no active Cherries run"
        raise RuntimeError(msg)
    if isinstance(exc, SystemExit) and exc.code in (None, 0):
        # ``sys.exit(0)`` has the same process outcome as returning from a
        # normal Python experiment.  Seal it before main() re-raises the
        # exception so the interpreter can still exit successfully.
        exc = None
    self.log_other("cherries/end_time", datetime.now().astimezone())
    try:
        self._join_writers()
        if exc is not None:
            diagnostic = "".join(traceback.format_exception(exc))[-65536:]
            self.log_other("cherries/exception", diagnostic)
            self.plugins.delegate("end", exc=exc)
            self._close_logging()
            failure = {
                "exception": diagnostic,
                "execution": "failed",
                "name": self.run_name,
                "source": self._source_evidence,
            }
            if (
                self._settings.get("execution", {}).get(
                    "failure_payload", "discard"
                )
                == "discard"
            ):
                self.store.cancel_failed_work(self.run_id, failure)
            else:
                self.store.append_event("execution-failed", self.run_id, failure)
            return
        self._assets.end()
        end_source = capture_source(
            self.project_dir,
            self.entrypoint,
            self.working_dir / "source-end",
            self._settings,
        )
        stable = self._source_evidence.get("fingerprint") == end_source.get(
            "fingerprint"
        )
        config = self.working_dir / "config"
        config.mkdir(parents=True, exist_ok=True)
        (config / "resolved.json").write_text(
            json.dumps(
                _json_safe(self.get_params()),
                default=str,
                sort_keys=True,
                indent=2,
                allow_nan=False,
            )
            + "\n"
        )
        (config / "bindings.json").write_text(
            json.dumps(self._assets.bindings, default=str, sort_keys=True, indent=2)
            + "\n"
        )
        metrics = self.get_metrics().to_dicts() if self._metrics.metrics else []
        (self.working_dir / "logs/metrics.json").write_text(
            json.dumps(
                _json_safe(metrics),
                default=str,
                sort_keys=True,
                indent=2,
                allow_nan=False,
            )
            + "\n"
        )
        (self.working_dir / "RUN.md").write_text(
            f"# {self.run_name}\n\nRun: `{self.run_id}`\n\nExecution: succeeded. Review: unreviewed.\n\nSource stable across execution: {stable}. Replay has not been verified.\n"
        )
        self.plugins.delegate("end", exc=None)
        self._close_logging()
        self.record_result = self.store.seal(
            self.run_id,
            {
                "kind": "experiment",
                "name": self.run_name,
                "tags": self.tags,
                "command": shlex.join(sys.orig_argv),
                "argv": sys.argv[1:],
                "entrypoint": str(
                    relative_or_absolute(self.entrypoint, self.project_dir)
                ),
                "execution": {"status": "succeeded", "exit_code": 0},
                "validation": {"status": "not_evaluated"},
                "params": _json_safe(self.get_params()),
                "others": json.loads(
                    json.dumps(
                        _json_safe(self.get_others()), default=str, allow_nan=False
                    )
                ),
                "input_bindings": self._assets.bindings,
                "bundles": self._assets.retained_bundles,
                "source": self._source_evidence,
                "source_stability": stable,
                "replay_verified": False,
            },
            self.working_dir,
        )
        logger.info("Saved Cherries run %s", self.run_id)
    except BaseException as failure:
        self.store.append_event(
            "recording-incomplete",
            self.run_id,
            {"reason": str(failure), "work": str(self.working_dir)},
        )
        raise
    finally:
        self._close_logging()
        self._assets.active = False
        self.active = False

def get_metric(self, name: str) -> pl.DataFrame:
    """Return one metric series."""
    return self._metrics.get_metric(name)

def get_metrics(self, metrics: Iterator[str] | None = None) -> pl.DataFrame:
    """Return selected metric series concatenated into one dataframe."""
    return self._metrics.get_metrics(metrics)

def get_other(self, name: str) -> Any:
    """Return one flattened metadata value."""
    return self._others.get_other(name)

def get_others(self) -> dict[str, Any]:
    """Return logged metadata as a nested dictionary."""
    return self._others.get_others()

def get_param(self, name: str) -> Any:
    """Return one flattened parameter value."""
    return self._params.get_param(name)

def get_params(self) -> dict[str, Any]:
    """Return logged parameters as a nested dictionary."""
    return self._params.get_params()

def get_step(self) -> int:
    """Return the default metric step."""
    return self.step

def input(
    self,
    path: StrPath,
    *,
    name: StrPath | None = None,
    source_run: str | None = None,
    metadata: Mapping[str, Any] | None = None,
) -> Path:
    """Copy verified input bytes into the active run and record their origin."""
    return self._assets.input(
        path, name=name, source_run=source_run, metadata=metadata
    )

def log_asset(
    self,
    path: StrPath,
    metadata: Mapping[str, Any] | None = None,
    *,
    name: StrPath | None = None,
) -> Path:
    """Retain an explicit artifact as an independent file inside the active run."""
    return self._assets.log_asset(path, metadata=metadata, name=name)

def log_input(
    self,
    path: StrPath,
    metadata: Mapping[str, Any] | None = None,
    *,
    name: StrPath | None = None,
) -> Path:
    """Copy and retain an existing input in the active run."""
    return self._assets.log_input(path, metadata=metadata, name=name)

def log_metric(
    self,
    name: str,
    value: SupportsFloat,
    *,
    step: int | None = None,
    time: datetime | None = None,
) -> None:
    """Log one scalar metric."""
    self._metrics.log_metric(name, value, step=step, time=time)

def log_metrics(
    self,
    metrics: MetricsLike,
    *,
    step: int | None = None,
    time: datetime | None = None,
) -> None:
    """Log multiple scalar metrics, flattening nested mappings with `/`."""
    self._metrics.log_metrics(metrics, step=step, time=time)

def log_other(self, name: str, value: Any) -> None:
    """Log one metadata value."""
    self._others.log_other(name, value)

def log_others(self, others: Mapping[str, Any]) -> None:
    """Log multiple metadata values."""
    self._others.log_others(others)

def log_output(
    self,
    path: StrPath,
    metadata: Mapping[str, Any] | None = None,
    *,
    name: StrPath | None = None,
) -> Path:
    """Copy or register an existing output inside the active run."""
    return self._assets.log_output(path, metadata=metadata, name=name)

def log_param(self, name: str, value: Any) -> None:
    """Log one parameter value."""
    self._params.log_param(name, value)

def log_params(self, params: Mapping[str, Any]) -> None:
    """Log multiple parameter values."""
    self._params.log_params(params)

def log_temp(
    self,
    path: StrPath,
    metadata: Mapping[str, Any] | None = None,
    *,
    name: StrPath | None = None,
) -> Path:
    """Promote a temporary file into retained run artifacts."""
    return self._assets.log_temp(path, metadata=metadata, name=name)

def output(
    self,
    path: StrPath,
    *,
    metadata: Mapping[str, Any] | None = None,
    mkdir: bool = True,
) -> Path:
    """Declare a required output in the active run; missing outputs fail saving."""
    return self._assets.output(path, metadata=metadata, mkdir=mkdir)

def set_step(self, step: int) -> None:
    """Set the default metric step."""
    self.step = step

def start(self) -> None:
    """Allocate local work and capture evidence before calling user code."""
    if self.active:
        msg = "a Cherries run is already active"
        raise RuntimeError(msg)
    self.run_id = str(uuid.uuid4())
    self.start_time = datetime.now().astimezone()
    self._settings = load_settings(self.project_dir)
    self.store = Store(self.store_root or storage_root(self.project_dir))
    self.store.ensure_initialized(self._settings.get("collection", {}).get("id"))
    self.working_dir = self.store.start_work(
        self.run_id,
        {
            "pid": os.getpid(),
            "name": self.run_name,
            "machine_id": self.store.machine_id,
        },
    )
    parent = os.environ.get("CHERRIES_PARENT_RUN")
    if parent:
        self.store.register_parent(self.run_id, self.store.resolve_id(parent))
    self._assets = AssetsManager(
        working_dir=self.working_dir,
        plugins=cast("AssetPluginProtocol", self.plugins),
        active=True,
        store=self.store,
        run_id=self.run_id,
    )
    self._metrics = self._default_metrics()
    self._params = self._default_params()
    self._others = self._default_others()
    self.record_result = None
    self.active = True
    logs = self.working_dir / "logs"
    logs.mkdir(parents=True)
    handler = logging.FileHandler(logs / "run.log", encoding="utf-8")
    handler.setFormatter(
        logging.Formatter("%(asctime)s %(levelname)s %(name)s: %(message)s")
    )
    logging.getLogger().addHandler(handler)
    self._file_handler = handler
    self._source_evidence = capture_source(
        self.project_dir,
        self.entrypoint,
        self.working_dir / "source",
        self._settings,
    )
    capture_environment(self.project_dir, self.working_dir / "environment")
    self.plugins.delegate("start")
    self._threads = tuple(threading.enumerate())
    self._child_pids = self._children()
    self.log_other("cherries/cmd", shlex.join(sys.orig_argv))
    self.log_other(
        "cherries/entrypoint",
        relative_or_absolute(self.entrypoint, self.project_dir),
    )
    self.log_other("cherries/exp_dir", self.working_dir)
    self.log_other("cherries/start_time", self.start_time)
    self.log_other("cherries/run_id", self.run_id)

def summary(self, prefix: StrPath | None = None) -> dict[str, Any]:
    """Build a JSON/YAML-friendly run summary.

    Args:
        prefix: Optional directory to strip from artifact paths.

    Returns:
        Run metadata, parameters, artifact paths, and user metadata.
    """
    summary: dict[str, Any] = {"name": self.run_name}
    if self.tags:
        summary["tags"] = self.tags
    others: dict[str, Any] = self.get_others()
    summary.update(others.pop("cherries"))
    summary["params"] = self.get_params()
    summary.update(self._assets.summary.to_dict(prefix=prefix))
    summary["others"] = others
    return summary

def temp(
    self,
    path: StrPath,
    *,
    metadata: Mapping[str, Any] | None = None,
    mkdir: bool = True,
) -> Path:
    """Return a disposable scratch path in the active run."""
    return self._assets.temp(path, metadata=metadata, mkdir=mkdir)

def __enter__(self) -> Self:
    return self

def __exit__(self, *_args) -> None:
    self.close()

def close(self) -> None:
    if not self.closed:
        self.store.release_hold(self.run_id, self.reason)
    self.closed = True

def path(self, relative: str) -> Path:
    if self.closed:
        msg = "record accessor is closed"
        raise RuntimeError(msg)
    if not relative or Path(relative).is_absolute() or ".." in Path(relative).parts:
        msg = "asset path must be relative and contained"
        raise ValueError(msg)
    files = self.store.read_manifest(self.run_id)["files"]
    selected = [
        entry
        for entry in files
        if entry["path"] == relative
        or entry["path"].startswith(relative.rstrip("/") + "/")
    ]
    bundle_path = self._bundle_path(relative)
    if bundle_path is not None:
        return bundle_path
    if not selected:
        raise FileNotFoundError(relative)
    from .core.assets.bundle import bundles

    for entry in selected:
        local = self._materialize(entry["path"])
        for companion_name, optional in bundles.ls_files(local):
            companion = Path(companion_name)
            if not companion.resolve().is_relative_to(local.parent.resolve()):
                msg = "asset companion escapes its bundle"
                raise ValueError(msg)
            logical = companion.relative_to(
                self.store.root / "runs" / self.run_id
            ).as_posix()
            try:
                self._materialize(logical)
            except (FileNotFoundError, NotFoundError):
                if not optional:
                    raise
    return self.store.root / "runs" / self.run_id / relative