Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion src/couchdb/.allowed_datafiles
Original file line number Diff line number Diff line change
Expand Up @@ -16,4 +16,5 @@ src/couchdb/scenarios_data/shared/iot/motor_01.json
src/couchdb/scenarios_data/shared/work_order/workorders.csv
src/couchdb/scenarios_data/shared/tsfm/feature_catalog.json
src/couchdb/scenarios_data/shared/tsfm/model_catalog.json
src/couchdb/scenarios_data/shared/iot/asset_profile_sample.json
src/couchdb/scenarios_data/shared/iot/asset_profile_sample.json
src/couchdb/scenarios_data/shared/tsfm/feature_catalog_orig.json
36 changes: 0 additions & 36 deletions src/couchdb/scenarios_data/shared/tsfm/feature_catalog.json
Original file line number Diff line number Diff line change
@@ -1,40 +1,4 @@
[
{
"feature_id": "efe_time_robust_norm_v1",
"name": "Invertible robust normalization",
"modality": "timeseries",
"interface": "fit_transform_inverse",
"class_name": "Transformation",
"invertible": true,
"code": "import numpy as np\nclass Transformation:\n \"\"\"Invertible robust per-channel normalization (median/IQR) for forecasting inputs.\"\"\"\n def fit(self, X, metadata):\n X = np.asarray(X, dtype=float)\n center = np.median(X, axis=0)\n q1, q3 = np.percentile(X, 25, axis=0), np.percentile(X, 75, axis=0)\n scale = np.where((q3 - q1) > 1e-8, (q3 - q1), 1.0)\n return {\"center\": center, \"scale\": scale}\n def transform(self, X, state):\n X = np.asarray(X, dtype=float)\n return (X - state[\"center\"]) / state[\"scale\"]\n def inverse_transform(self, Y, state):\n Y = np.asarray(Y, dtype=float)\n return Y * state[\"scale\"] + state[\"center\"]\n",
"provenance": "evolved",
"method": "EFE-Time",
"parent_feature_id": null,
"generation": 0,
"target_task": "tsfm_forecasting",
"target_model": null,
"dataset": null,
"output_type": "normalized_series",
"columns_added": [],
"columns_dropped": [],
"validity": {
"entry_points": true,
"no_inplace": true,
"invertible_ok": true,
"leakage_checked": true
},
"tags": [
"timeseries",
"EFE-Time",
"Future State Prediction"
],
"status": "active",
"version": "1",
"created_by": "seed",
"created_at": "2026-06-20T00:00:00+00:00",
"kind": "transform",
"description": "Invertible per-channel robust normalization (median/IQR) for forecasting inputs."
},
{
"feature_id": "abs_2nd_diff_mean",
"name": "abs_2nd_diff_mean",
Expand Down
46 changes: 46 additions & 0 deletions src/couchdb/scenarios_data/shared/tsfm/feature_catalog_orig.json
Original file line number Diff line number Diff line change
@@ -0,0 +1,46 @@
[
{
"feature_id": "efe_time_robust_norm_v1",
"name": "Invertible robust normalization",
"modality": "timeseries",
"interface": "fit_transform_inverse",
"class_name": "Transformation",
"invertible": true,
"code": "import numpy as np\nclass Transformation:\n \"\"\"Invertible robust per-channel normalization (median/IQR) for forecasting inputs.\"\"\"\n def fit(self, X, metadata):\n X = np.asarray(X, dtype=float)\n center = np.median(X, axis=0)\n q1, q3 = np.percentile(X, 25, axis=0), np.percentile(X, 75, axis=0)\n scale = np.where((q3 - q1) > 1e-8, (q3 - q1), 1.0)\n return {\"center\": center, \"scale\": scale}\n def transform(self, X, state):\n X = np.asarray(X, dtype=float)\n return (X - state[\"center\"]) / state[\"scale\"]\n def inverse_transform(self, Y, state):\n Y = np.asarray(Y, dtype=float)\n return Y * state[\"scale\"] + state[\"center\"]\n",
"provenance": "evolved",
"method": "EFE-Time",
"parent_feature_id": null,
"generation": 0,
"target_task": "tsfm_forecasting",
"target_model": null,
"dataset": null,
"output_type": "normalized_series",
"columns_added": [],
"columns_dropped": [],
"validity": {
"entry_points": true,
"no_inplace": true,
"invertible_ok": true,
"leakage_checked": true
},
"tags": [
"timeseries",
"EFE-Time",
"Future State Prediction"
],
"status": "active",
"version": "1",
"created_by": "seed",
"created_at": "2026-06-20T00:00:00+00:00",
"kind": "transform",
"description": "Invertible per-channel robust normalization (median/IQR) for forecasting inputs."
},
{
"feature_id": "abs_2nd_diff_mean",
"name": "abs_2nd_diff_mean",
"description": "Mean absolute second difference.",
"kind": "extractor",
"extractor_name": "abs_2nd_diff_mean",
"status": "active"
}
]
4 changes: 3 additions & 1 deletion src/servers/tsfm/engine/composition.py
Original file line number Diff line number Diff line change
Expand Up @@ -204,7 +204,9 @@ def _save_checkpoint(fc, path: str) -> str:
f"{type(fc).__name__} has none. A card can only point at weights via "
f"params.model_path, so there is no card-loadable way to store this estimator."
)
path = os.path.abspath(os.path.expanduser(path))
from ..io import refs

path = refs._path(str(path)) # confined to the data roots

@ShuxinLin ShuxinLin Oct 15, 2026 •

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Writes are only confined for save_to. _with_finetune (line 91) merges the finetune block into the estimator params, and TinyTimeMixer/PatchTST pass training_args straight to transformers.TrainingArguments. So finetune: {training_args: {output_dir: ...}} makes the Trainer write checkpoints, including the pickled training_args.bin, to any writable path.

Suggest overriding output_dir with a server-chosen directory under WORKDIR and rejecting any other path-valued training args.

os.makedirs(path, exist_ok=True)
model.save_pretrained(path)
return path
Expand Down
22 changes: 3 additions & 19 deletions src/servers/tsfm/engine/feature_runner.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,32 +8,16 @@
from __future__ import annotations

import copy
from typing import Any, Dict, Optional
from typing import Optional


class FeatureValidationError(Exception):
pass


def load_program(code: str, class_name: str = "Transformation"):
"""Exec a stored program string and return its Transformation class.

NOTE: executes code. In the benchmark/agent setting run this in a sandbox
(restricted globals, subprocess, or container). Here we exec in a fresh namespace
with a minimal import surface, matching how EFE executes candidate programs.
"""
ns: Dict[str, Any] = {}
try:
exec(code, ns) # noqa: S102 - intentional; sandbox in production

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@ShuxinLin this is a main thing we disabled.

except Exception as e:
raise FeatureValidationError(f"program failed to import: {e}")
cls = ns.get(class_name)
if cls is None:
raise FeatureValidationError(f"no class '{class_name}' defined")
for m in ("fit", "transform"):
if not callable(getattr(cls, m, None)):
raise FeatureValidationError(f"missing required method '{m}'")
return cls
"""Disabled: the server never executes feature-card source (it used to exec() it)."""
raise FeatureValidationError("executing feature-card code is disabled in the TSFM server")


def _no_inplace(before, after) -> bool:
Expand Down
38 changes: 34 additions & 4 deletions src/servers/tsfm/io/refs.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,8 @@

import json
import os
import re
import tempfile
import uuid
from typing import List, Optional

Expand All @@ -20,8 +22,36 @@
WORKDIR = os.environ.get("TSFM_WORKDIR", "/tmp/tsfm_work")


def _data_roots() -> List[str]:
"""Directories a file pointer may resolve into: this server's WORKDIR, the system temp dir
and /tmp (code_exec workspaces, other servers' outputs), the Harbor shared workspace, and
artifacts/output (where agents save fine-tuned checkpoints via recipe save_to).
TSFM_DATA_ROOTS (os.pathsep-separated) replaces the default."""
env = os.environ.get("TSFM_DATA_ROOTS")
roots = env.split(os.pathsep) if env else [
WORKDIR, tempfile.gettempdir(), "/tmp", "/workspace-share",
os.path.join(os.getcwd(), "artifacts", "output")] # where agents save checkpoints
return [os.path.realpath(r) for r in roots if r]


def _path(ref: str) -> str:
return ref[7:] if ref.startswith("file://") else ref
"""Resolve a file pointer, confined to the data roots.

Unconfined, every dataset_path read any file the server could open, e.g.
/proc/self/environ (the CouchDB password). realpath() follows symlinks, so a link
planted in a workspace cannot escape; URLs (http://, s3://) are refused.
"""
if "://" in ref and not ref.startswith("file://"):
raise ValueError("only local file pointers are supported")
p = os.path.realpath(ref[7:] if ref.startswith("file://") else ref)

@ShuxinLin ShuxinLin Oct 15, 2026 •

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Confinement can still be bypassed. realpath() checks the path here, but callers open it by name later (pd.read_csv(p)). /workspace-share is writable from the code sandbox, so a symlink swapped in between the check and the open escapes the roots. Non-regular files also pass, so a FIFO named x.csv makes read_csv block forever.

Suggested fix: open first, then validate what was opened. os.open(p, os.O_RDONLY | os.O_NONBLOCK), require stat.S_ISREG(os.fstat(fd).st_mode), check the fd's real path (os.readlink(f"/proc/self/fd/{fd}")) against the roots, and pass the open file to pandas. Every file-reading tool goes through _path, so this covers them all.

if not any(p == r or p.startswith(r + os.sep) for r in _data_roots()):
raise ValueError(f"file pointer is outside the allowed data directories: {ref}")
return p


def _safe_name(name) -> str:
"""Output file stem; plan step ids are agent-supplied, so strip path characters."""
return re.sub(r"[^A-Za-z0-9_.-]", "_", str(name)).lstrip(".") or "result"


def _ensure_workdir():
Expand Down Expand Up @@ -91,7 +121,7 @@ def load_series(
def write_series(obj, *, name: str = "result") -> str:
"""Write a forecast/series/frame to a file pointer and return the ref."""
_ensure_workdir()
p = os.path.join(WORKDIR, f"{name}_{uuid.uuid4().hex[:8]}.csv")
p = os.path.join(WORKDIR, f"{_safe_name(name)}_{uuid.uuid4().hex[:8]}.csv")
(obj.to_frame() if isinstance(obj, pd.Series) else pd.DataFrame(obj)).to_csv(
p, index=False
)
Expand All @@ -100,7 +130,7 @@ def write_series(obj, *, name: str = "result") -> str:

def write_json(obj: dict, *, name: str = "result") -> str:
_ensure_workdir()
p = os.path.join(WORKDIR, f"{name}_{uuid.uuid4().hex[:8]}.json")
p = os.path.join(WORKDIR, f"{_safe_name(name)}_{uuid.uuid4().hex[:8]}.json")
json.dump(obj, open(p, "w"), indent=2, default=str)
return f"file://{p}"

Expand All @@ -124,6 +154,6 @@ def materialize_iot(
df.insert(
0, "timestamp", pd.date_range("2020-01-01", periods=len(arr), freq=freq)
)
p = os.path.join(WORKDIR, f"iot_{asset_id}.csv")
p = os.path.join(WORKDIR, f"iot_{_safe_name(asset_id)}.csv")
df.to_csv(p, index=False)
return f"file://{p}"
7 changes: 5 additions & 2 deletions src/servers/tsfm/main.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,11 @@
import logging
import sys
import os

# Load weights only from the preloaded HF cache, never a repo the agent names: MOIRAIForecaster
# unpickles model.ckpt (weights_only=False) from any non-Salesforce repo. Set before any
# huggingface_hub import; override with HF_HUB_OFFLINE=0 for local development.
os.environ.setdefault("HF_HUB_OFFLINE", "1")
from typing import List, Optional, Union

import pandas as pd
Expand Down Expand Up @@ -1329,7 +1334,6 @@ def get_feature(feature_id: str) -> Union[CardResult, ErrorResult]:
return ErrorResult(error=str(exc))


@mcp.tool(title="Register Feature")
def register_feature(
feature: dict,
overwrite: bool = False,
Expand Down Expand Up @@ -1426,7 +1430,6 @@ def deprecate_feature(
return ErrorResult(error=str(exc))


@mcp.tool(title="New Feature Version")
def new_feature_version(
feature_id: str,
fields: Optional[dict] = None,
Expand Down
6 changes: 3 additions & 3 deletions src/servers/tsfm/reasoning/param_space.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,15 +14,15 @@

from __future__ import annotations

import importlib
import inspect
from typing import Dict, Optional


def introspect(sktime_class: str) -> dict:
"""Constructor params (name → default/required/type) + sktime example param sets."""
module, cls = sktime_class.rsplit(".", 1)
Est = getattr(importlib.import_module(module), cls)
from ..substrate.resolver import _import_target

Est = _import_target(sktime_class)
sig = inspect.signature(Est.__init__)
_empty = inspect.Parameter.empty
params: Dict[str, dict] = {}
Expand Down
84 changes: 83 additions & 1 deletion src/servers/tsfm/substrate/resolver.py
Original file line number Diff line number Diff line change
Expand Up @@ -26,8 +26,90 @@
_TARGET_KEY = "_target_"


# Only these classes may be imported from a dotted path. Paths reach this function from agent
# input (inline sktime_class, nested _target_, catalog cards the agent can rewrite), so an
# open import ran any callable, e.g. subprocess.Popen. Operators extend the list with
# TSFM_EXTRA_ESTIMATOR_CLASSES (comma-separated exact paths).
ALLOWED_CLASSES = frozenset({
Comment thread
DhavalRepo18 marked this conversation as resolved.
"sktime.forecasting.ttm.TinyTimeMixerForecaster",
"sktime.forecasting.chronos.ChronosForecaster",
"sktime.forecasting.timesfm.TimesFMForecaster",
"sktime.forecasting.timesfm2.TimesFM2Forecaster",
"sktime.forecasting.moirai.MOIRAIForecaster",
"sktime.forecasting.moirai2.Moirai2Forecaster",
"sktime.forecasting.toto2.Toto2Forecaster",
"sktime.forecasting.kronos.KronosForecaster",
"sktime.forecasting.momentfm.MomentFMForecaster",
"sktime.forecasting.timemoe.TimeMoEForecaster",
"sktime.forecasting.patch_tst.PatchTSTForecaster",
"sktime.forecasting.patch_tsmixer.PatchTSMixerForecaster",
"sktime.forecasting.naive.NaiveForecaster",
"sktime.forecasting.theta.ThetaForecaster",
"sktime.forecasting.arima.ARIMA",
"sktime.forecasting.arima.AutoARIMA",
"sktime.forecasting.conformal.ConformalIntervals",
"sktime.detection.tspulse.TSPulseAnomalyDetector",
"sktime.detection.hampel.HampelDetector",
"sktime.detection.lof.SubLOF",
"sktime.detection.adapters._pyod.PyODDetector",
"sktime.detection.dummy._zero_an.ZeroAnomalies",
"pyod.models.iforest.IForest",
"sktime.classification.foundation_models.tspulse.TSPulseClassifier",
"sklearn.ensemble.RandomForestClassifier",
"sktime.clustering.k_means.TimeSeriesKMeans",
"sktime.clustering.k_medoids.TimeSeriesKMedoids",
"sktime.clustering.dbscan.TimeSeriesDBSCAN",
"sktime.clustering.kernel_k_means.TimeSeriesKernelKMeans",
"sktime.clustering.agglomerative.TimeSeriesAgglomerativeClustering",
# forecasting `transforms` (TransformedTargetForecaster steps) and their nested parts
"sktime.transformations.series.detrend.Detrender",
"sktime.transformations.series.detrend.Deseasonalizer",
"sktime.transformations.series.detrend.ConditionalDeseasonalizer",
"sktime.transformations.series.difference.Differencer",
"sktime.transformations.series.boxcox.LogTransformer",
"sktime.transformations.series.boxcox.BoxCoxTransformer",
"sktime.transformations.series.exponent.ExponentTransformer",
"sktime.transformations.series.exponent.SqrtTransformer",
"sktime.transformations.series.impute.Imputer",
"sktime.transformations.series.outlier_detection.HampelFilter",
"sktime.transformations.series.adapt.TabularToSeriesAdaptor",
"sktime.forecasting.trend.PolynomialTrendForecaster",
"sktime.forecasting.trend.TrendForecaster",
"sklearn.preprocessing.StandardScaler",
"sklearn.preprocessing.MinMaxScaler",
"sklearn.preprocessing.RobustScaler",
# run_tabular_recipe transforms
"sktime.transformations.series.summarize.SummaryTransformer",
"sktime.transformations.panel.catch22.Catch22",
"sktime.transformations.panel.rocket.Rocket",
"sktime.transformations.panel.rocket.MiniRocket",
"sktime.transformations.panel.rocket.MiniRocketMultivariate",
# run_tabular_recipe estimators: regression / classification / clustering
"sklearn.ensemble.RandomForestRegressor",
"sklearn.ensemble.ExtraTreesRegressor",
"sklearn.ensemble.GradientBoostingRegressor",
"sklearn.linear_model.LinearRegression",
"sklearn.linear_model.Ridge",
"sklearn.linear_model.Lasso",
"sklearn.neighbors.KNeighborsRegressor",
"sklearn.ensemble.ExtraTreesClassifier",
"sklearn.ensemble.GradientBoostingClassifier",
"sklearn.linear_model.LogisticRegression",
"sklearn.neighbors.KNeighborsClassifier",
"sklearn.cluster.KMeans",
})


def _import_target(path: str):
"""Import a dotted path 'pkg.module.Name' -> the class/callable Name."""
"""Import an allowlisted dotted path 'pkg.module.Name' -> the class Name.

The check runs before the import, so a refused path never loads its module.
"""
import os

extra = {p.strip() for p in os.environ.get("TSFM_EXTRA_ESTIMATOR_CLASSES", "").split(",")}
if path not in ALLOWED_CLASSES and path not in extra - {""}:
raise ValueError(f"'{path}' is not an allowed estimator class; use a catalog model_id")
module, name = path.rsplit(".", 1)
return getattr(importlib.import_module(module), name)

Expand Down
3 changes: 3 additions & 0 deletions src/servers/tsfm/tests/conftest.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,9 @@
import pytest

os.environ["TSFM_STORE"] = "memory"
os.environ.setdefault(
"TSFM_EXTRA_ESTIMATOR_CLASSES", "servers.tsfm.tests.fake_hf_forecaster.FakeTTMForecaster"
)

_SEEDS = os.environ.get("TSFM_SEEDS_DIR") or os.path.normpath(
os.path.join(
Expand Down
Loading
Loading