fix: CVE vulnerability patching

feature: e2e intro pipline surge pricing
chore: store sting
2026-07-15 17:43:36 +00:00 · 2025-12-06 17:43:33 +01:00 · 2025-12-06 16:30:28 +01:00 · 2025-12-05 18:04:23 +01:00 · 2025-12-05 18:03:36 +01:00 · 2025-12-05 17:59:37 +01:00
39 changed files with 250 additions and 1997 deletions
--- a/.gitignore
+++ b/.gitignore
@@ -11,6 +11,3 @@ paper/src/bib/auto
 experiments/airflow/logs/*
 experiments/airflow/logs/scheduler/
 experiments/airflow/logs/dag_processor_manager/
-tests/e2e/node_modules/**
-**/auto/*.el
-*.old
--- a/52
+++ b/52
@@ -11,72 +11,46 @@ PYTEST    := $(VENV)/bin/pytest

 .DEFAULT_GOAL := help

-.PHONY: help
-help:
-	@echo "pdf.build pdf.watch pdf.clean | test.backend test.e2e test.all | web.dev | install | stats.lines"
+all: pdf
+
+run.webapp:
+	@cd web && npm install && npm run dev

 $(BUILDDIR):
 	mkdir -p paper/$(BUILDDIR)

-.PHONY: pdf.build
-pdf.build: $(BUILDDIR)
+pdf: $(BUILDDIR)
+	@echo "Concatenating source code..."
 	@bash paper/concat_code.sh
 	@cd $(SRCDIR) && \
 	$(LATEXMK) -pdf -jobname=$(JOBNAME) \
 		-interaction=nonstopmode -file-line-error \
 		-outdir=../$(BUILDDIR) $(TEX)

-.PHONY: pdf.watch
-pdf.watch: $(BUILDDIR)
+watch: $(BUILDDIR)
 	@cd $(SRCDIR) && \
 	$(LATEXMK) -pvc -pdf -jobname=$(JOBNAME) \
 		-interaction=nonstopmode -file-line-error \
 		-r ../.latexmkrc \
 		-outdir=../$(BUILDDIR) $(TEX)

-.PHONY: pdf.clean
-pdf.clean:
+clean:
 	@cd $(SRCDIR) && \
 	$(LATEXMK) -C -jobname=$(JOBNAME) -outdir=../$(BUILDDIR) || true
 	rm -rf paper/$(BUILDDIR)/*

-.PHONY: test.backend
-test.backend: $(VENV)
-	$(PYTEST) -v
-
-.PHONY: test.e2e
-test.e2e:
-	@cd tests/e2e && npm install
-	@cd tests/e2e && npx playwright install chromium
-	@timeout 30 bash -c 'until curl -sf http://localhost:5000/health > /dev/null 2>&1; do sleep 1; done' || (echo "Backend not ready" && exit 1)
-	@timeout 30 bash -c 'until curl -sf http://localhost:3000 > /dev/null 2>&1; do sleep 1; done' || (echo "Web app not ready" && exit 1)
-	@cd tests/e2e && npm test
-
-.PHONY: test.all
-test.all: test.backend test.e2e
-
-.PHONY: web.dev
-web.dev:
-	@cd web && npm install && npm run dev
-
 $(VENV):
 	python3 -m venv $(VENV)
 	$(PIP) install --upgrade pip

-.PHONY: install
 install: $(VENV)
 	$(PIP) install -r requirements.txt

-.PHONY: stats.lines
-stats.lines:
+test: $(VENV)
+	$(PYTEST) -v
+
+count-lines:
 	@find . \( -path '*/node_modules' -o -path '*/.venv' -o -path '*/venv' \) -prune -o \
 	\( -name "*.ts" -o -name "*.py" \) -type f -print0 | xargs -0 cat | wc -l

-.PHONY: pdf clean watch run.webapp test count-lines all
-pdf: pdf.build
-clean: pdf.clean
-watch: pdf.watch
-run.webapp: web.dev
-test: test.backend
-count-lines: stats.lines
-all: pdf.build
+.PHONY: all pdf clean watch run.webapp install test
--- a/README.md
+++ b/README.md
@@ -1,12 +1,8 @@
-<img width="200" align="left" src="https://github.com/user-attachments/assets/d148b00d-e9f9-4280-89cc-0cc866e17251" />

-### PHANTOM
+<img width="1952" height="2176" alt="nobody_knows" src="https://github.com/user-attachments/assets/d148b00d-e9f9-4280-89cc-0cc866e17251" />

 [![Build PDF](https://github.com/velocitatem/PHANTOM/actions/workflows/latex.yml/badge.svg)](https://github.com/velocitatem/PHANTOM/actions/workflows/latex.yml)
-[![TPU Research Cloud](https://img.shields.io/badge/TPU%20Research%20Cloud-TRC%20supported-4285F4?logo=googlecloud&logoColor=white)](https://sites.research.google/trc/faq/)
-[![Vercel Deploy](https://deploy-badge.vercel.app/?url=https://phantom-hotel.vercel.app&name=Hotel)](https://phantom-hotel.vercel.app)
-[![Vercel Deploy](https://deploy-badge.vercel.app/?url=https://phantom-airline.vercel.app&name=Airline)](https://phantom-airline.vercel.app)
-
-

+- https://phantom-hotel.vercel.app/
+- https://phantom-airline.vercel.app/

--- a/docker-compose.yml
+++ b/docker-compose.yml
@@ -1,15 +1,4 @@
 services:
-
-  tensorboard:
-    image: tensorflow/tensorflow:latest
-    container_name: "PHANTOM-tensorboard"
-    ports:
-      - "6006:6006"
-    volumes:
-      - ./experiments/ml/runs:/logs
-    command: tensorboard --logdir=/logs --host=0.0.0.0 --port=6006
-    restart: unless-stopped
-
  backend:
    container_name: "PHANTOM-backend"
    build:
@@ -114,6 +103,12 @@ services:
      - _AIRFLOW_WWW_USER_PASSWORD=admin
      - REDIS_HOST=redis
      - REDIS_PORT=6379
+    volumes:
+      - ./experiments/airflow/dags:/opt/airflow/dags
+      - ./experiments/airflow/logs:/opt/airflow/logs
+      - ./experiments/airflow/plugins:/opt/airflow/plugins
+      - ./experiments/procesing:/opt/airflow/procesing
+      - ./lib:/opt/airflow/lib
    command: version
    restart: "no"

@@ -134,7 +129,6 @@ services:
      - AIRFLOW__CORE__LOAD_EXAMPLES=false
      - AIRFLOW__CORE__ENABLE_XCOM_PICKLING=true
      - AIRFLOW__WEBSERVER__EXPOSE_CONFIG=true
-      - AIRFLOW__WEBSERVER__SECRET_KEY=${AIRFLOW_SECRET_KEY}
      - KAFKA_HOST=kafka
      - KAFKA_PORT=29092
      - BACKEND_URL=http://backend:5000
@@ -144,6 +138,12 @@ services:
      - REDIS_PORT=6379
    ports:
      - "${AIRFLOW_WEBSERVER_PORT:-8085}:8080"
+    volumes:
+      - ./experiments/airflow/dags:/opt/airflow/dags:ro
+      - ./experiments/airflow/logs:/opt/airflow/logs
+      - ./experiments/airflow/plugins:/opt/airflow/plugins:ro
+      - ./experiments/procesing:/opt/airflow/procesing:ro
+      - ./lib:/opt/airflow/lib:ro
    command: webserver
    restart: unless-stopped
    healthcheck:
@@ -170,7 +170,6 @@ services:
      - AIRFLOW__CORE__DAGS_ARE_PAUSED_AT_CREATION=true
      - AIRFLOW__CORE__LOAD_EXAMPLES=false
      - AIRFLOW__CORE__ENABLE_XCOM_PICKLING=true
-      - AIRFLOW__WEBSERVER__SECRET_KEY=${AIRFLOW_SECRET_KEY}
      - KAFKA_HOST=kafka
      - KAFKA_PORT=29092
      - BACKEND_URL=http://backend:5000
@@ -178,6 +177,12 @@ services:
      - NEXT_PUBLIC_SUPABASE_ANON_KEY=${NEXT_PUBLIC_SUPABASE_ANON_KEY}
      - REDIS_HOST=redis
      - REDIS_PORT=6379
+    volumes:
+      - ./experiments/airflow/dags:/opt/airflow/dags:ro
+      - ./experiments/airflow/logs:/opt/airflow/logs
+      - ./experiments/airflow/plugins:/opt/airflow/plugins:ro
+      - ./experiments/procesing:/opt/airflow/procesing:ro
+      - ./lib:/opt/airflow/lib:ro
    command: scheduler
    restart: unless-stopped
    healthcheck:
--- a/docker/Airflow.dockerfile
+++ b/docker/Airflow.dockerfile
@@ -21,10 +21,3 @@ RUN pip install --no-cache-dir \

 # set airflow home
 ENV AIRFLOW_HOME=/opt/airflow
-
-COPY --chown=airflow:root experiments/airflow/dags ${AIRFLOW_HOME}/dags
-COPY --chown=airflow:root experiments/procesing ${AIRFLOW_HOME}/procesing
-COPY --chown=airflow:root lib ${AIRFLOW_HOME}/lib
-
-# create logs and plugins dirs (airflow expects them)
-RUN mkdir -p ${AIRFLOW_HOME}/logs ${AIRFLOW_HOME}/plugins
--- a/docker/Airflow.railway.dockerfile
+++ b/docker/Airflow.railway.dockerfile
@@ -1,41 +0,0 @@
-FROM apache/airflow:2.7.3-python3.11
-
-USER root
-
-RUN apt-get update && apt-get install -y --no-install-recommends \
-    build-essential \
-    supervisor \
-    && apt-get clean \
-    && rm -rf /var/lib/apt/lists/*
-
-USER airflow
-
-COPY requirements.txt /tmp/requirements.txt
-RUN pip install --no-cache-dir -r /tmp/requirements.txt
-
-RUN pip install --no-cache-dir \
-    psycopg2-binary \
-    apache-airflow-providers-postgres
-
-ENV AIRFLOW_HOME=/opt/airflow
-ENV AIRFLOW__CORE__EXECUTOR=SequentialExecutor
-ENV AIRFLOW__CORE__LOAD_EXAMPLES=false
-ENV AIRFLOW__CORE__ENABLE_XCOM_PICKLING=true
-ENV AIRFLOW__WEBSERVER__EXPOSE_CONFIG=true
-
-# copy all code into image (standalone - no volume mounts needed)
-COPY --chown=airflow:root experiments/airflow/dags ${AIRFLOW_HOME}/dags
-COPY --chown=airflow:root experiments/procesing ${AIRFLOW_HOME}/procesing
-COPY --chown=airflow:root lib ${AIRFLOW_HOME}/lib
-
-RUN mkdir -p ${AIRFLOW_HOME}/logs ${AIRFLOW_HOME}/plugins
-
-# copy entrypoint script
-COPY --chown=airflow:root docker/airflow-railway-entrypoint.sh /entrypoint.sh
-USER root
-RUN chmod +x /entrypoint.sh
-USER airflow
-
-EXPOSE 8080
-
-ENTRYPOINT ["/entrypoint.sh"]
--- a/docker/airflow-railway-entrypoint.sh
+++ b/docker/airflow-railway-entrypoint.sh
@@ -1,20 +0,0 @@
-#!/bin/bash
-set -e
-
-# init db and create admin user on first run
-airflow db migrate
-
-# create admin user if not exists
-airflow users create \
-    --username "${AIRFLOW_ADMIN_USER:-admin}" \
-    --password "${AIRFLOW_ADMIN_PASSWORD:-admin}" \
-    --firstname Admin \
-    --lastname User \
-    --role Admin \
-    --email admin@example.com || true
-
-# start scheduler in background
-airflow scheduler &
-
-# start webserver in foreground (Railway needs one foreground process)
-exec airflow webserver --port ${PORT:-8080}
--- a/experiments/airflow/dags/ml_training_pipeline.py
+++ b/experiments/airflow/dags/ml_training_pipeline.py
@@ -1,115 +0,0 @@
-from airflow import DAG, Dataset
-from airflow.decorators import task
-from airflow.utils.dates import days_ago
-from datetime import timedelta
-import pandas as pd
-import logging
-import sys
-import pickle
-
-sys.path.insert(0, '/opt/airflow')
-
-from procesing.context import PipelineContext
-from procesing.providers import SupabaseProvider, BackendAPIProvider
-from procesing.steps import (
-    FetchInteractionsStep,
-    ValidateDataStep,
-    ExtractSessionFeaturesStep,
-    JoinLabelsStep,
-)
-
-TRAINING_DATASET = Dataset('phantom://ml/training-data')
-
-DEFAULT_ARGS = {
-    'owner': 'phantom-research',
-    'depends_on_past': False,
-    'email_on_failure': False,
-    'email_on_retry': False,
-    'retries': 2,
-    'retry_delay': timedelta(minutes=5),
-}
-
-
-class CompositeProvider(SupabaseProvider, BackendAPIProvider):
-    def __init__(self):
-        SupabaseProvider.__init__(self)
-        BackendAPIProvider.__init__(self)
-
-
-def _get_context(store_mode: str = 'hotel') -> PipelineContext:
-    return PipelineContext(provider=CompositeProvider(), store_mode=store_mode)
-
-
-with DAG(
-    'ml_training_pipeline',
-    default_args=DEFAULT_ARGS,
-    description='ML training data pipeline: fetch -> validate -> extract features -> label -> publish',
-    schedule=None,
-    start_date=days_ago(1),
-    catchup=False,
-    max_active_runs=1,
-    tags=['ml', 'training', 'features', 'research'],
-) as dag:
-
-    @task
-    def fetch_interactions(**kwargs) -> bytes:
-        dag_conf = kwargs.get('dag_run').conf if kwargs.get('dag_run') else {}
-        ctx = _get_context(dag_conf.get('store_mode', 'hotel'))
-        df = FetchInteractionsStep(ctx).transform(None)
-        logging.info(f"Fetched {len(df)} interactions, {df['sessionId'].nunique()} sessions")
-        return pickle.dumps(df)
-
-    @task
-    def validate_data(raw_data: bytes, **kwargs) -> bytes:
-        df = pickle.loads(raw_data)
-        dag_conf = kwargs.get('dag_run').conf if kwargs.get('dag_run') else {}
-        ctx = _get_context(dag_conf.get('store_mode', 'hotel'))
-        validated = ValidateDataStep(ctx).transform(df)
-        report = ctx.get_cached('validation_report') or {}
-        logging.info(f"Validation: {report.get('status')}, {report.get('sessions', 0)} sessions")
-        return pickle.dumps(validated)
-
-    @task
-    def extract_session_features(validated_data: bytes, **kwargs) -> bytes:
-        df = pickle.loads(validated_data)
-        if df.empty:
-            logging.warning("Empty input, skipping feature extraction")
-            return pickle.dumps(pd.DataFrame())
-        dag_conf = kwargs.get('dag_run').conf if kwargs.get('dag_run') else {}
-        ctx = _get_context(dag_conf.get('store_mode', 'hotel'))
-        features = ExtractSessionFeaturesStep(ctx).transform(df)
-        logging.info(f"Extracted {len(features.columns)} features for {len(features)} sessions")
-        return pickle.dumps(features)
-
-    @task
-    def join_labels(features_data: bytes, **kwargs) -> bytes:
-        features_df = pickle.loads(features_data)
-        if features_df.empty:
-            logging.warning("Empty features, skipping label join")
-            return pickle.dumps(pd.DataFrame())
-        dag_conf = kwargs.get('dag_run').conf if kwargs.get('dag_run') else {}
-        ctx = _get_context(dag_conf.get('store_mode', 'hotel'))
-        labeled = JoinLabelsStep(ctx).transform(features_df)
-        n_agents = labeled['is_agent'].sum() if 'is_agent' in labeled.columns else 0
-        logging.info(f"Labeled {len(labeled)} sessions: {n_agents} agents")
-        return pickle.dumps(labeled)
-
-    @task(outlets=[TRAINING_DATASET])
-    def publish_training_data(labeled_data: bytes, **kwargs) -> dict:
-        labeled_df = pickle.loads(labeled_data)
-        if labeled_df.empty:
-            return {'status': 'skipped', 'reason': 'empty_data'}
-        dag_conf = kwargs.get('dag_run').conf if kwargs.get('dag_run') else {}
-        return {
-            'status': 'success',
-            'n_sessions': len(labeled_df),
-            'n_features': len([c for c in labeled_df.columns if c not in ['sessionId', 'experimentId', 'is_agent']]),
-            'store_mode': dag_conf.get('store_mode', 'hotel'),
-            'timestamp': pd.Timestamp.now().isoformat(),
-        }
-
-    raw = fetch_interactions()
-    validated = validate_data(raw)
-    features = extract_session_features(validated)
-    labeled = join_labels(features)
-    publish_training_data(labeled)
--- a/experiments/airflow/dags/surge_pricing_factory.py
+++ b/experiments/airflow/dags/surge_pricing_factory.py
@@ -1,210 +0,0 @@
-from airflow import DAG
-from airflow.operators.python import PythonOperator
-from airflow.utils.dates import days_ago
-from datetime import timedelta
-import pandas as pd
-import logging
-import sys
-import pickle
-
-sys.path.insert(0, '/opt/airflow')
-
-from procesing.context import PipelineContext
-from procesing.providers import SupabaseProvider, BackendAPIProvider
-from procesing.steps import (
-    FetchInteractionsStep,
-    FetchPriceLogsStep,
-    ComputeDemandStep,
-    AggregatePriceLogsStep,
-    JoinProductFeaturesStep,
-)
-from procesing.pricers.simple import SimpleSurgePricer
-
-DEFAULT_ARGS = {
-    'owner': 'phantom-research',
-    'depends_on_past': False,
-    'email_on_failure': False,
-    'email_on_retry': False,
-    'retries': 2,
-    'retry_delay': timedelta(minutes=5),
-}
-
-class CompositeProvider(SupabaseProvider, BackendAPIProvider):
-    def __init__(self):
-        SupabaseProvider.__init__(self)
-        BackendAPIProvider.__init__(self)
-
-def _get_provider():
-    return CompositeProvider()
-
-def _make_task_callables(store_mode: str):
-    """Generate task callables bound to a specific store_mode."""
-
-    def get_context(**kwargs):
-        return PipelineContext(provider=_get_provider(), store_mode=store_mode)
-
-    def fetch_interactions(**kwargs):
-        ctx = get_context(**kwargs)
-        df = FetchInteractionsStep(ctx).transform(None)
-        kwargs['ti'].xcom_push(key='interactions_raw', value=pickle.dumps(df))
-        logging.info(f"[{store_mode}] Fetched {len(df)} interaction records")
-        return len(df)
-
-    def fetch_price_logs(**kwargs):
-        ctx = get_context(**kwargs)
-        df = FetchPriceLogsStep(ctx).transform(None)
-        kwargs['ti'].xcom_push(key='price_logs_raw', value=pickle.dumps(df))
-        logging.info(f"[{store_mode}] Fetched {len(df)} price records")
-        return len(df)
-
-    def compute_demand(**kwargs):
-        ti = kwargs['ti']
-        df = pickle.loads(ti.xcom_pull(key='interactions_raw'))
-        ctx = get_context(**kwargs)
-        demand_df = ComputeDemandStep(ctx).transform(df)
-        ti.xcom_push(key='demand_data', value=pickle.dumps(demand_df))
-        logging.info(f"[{store_mode}] Computed demand for {len(demand_df)} products")
-        return len(demand_df)
-
-    def aggregate_price_logs(**kwargs):
-        ti = kwargs['ti']
-        df = pickle.loads(ti.xcom_pull(key='price_logs_raw'))
-        ctx = get_context(**kwargs)
-        price_df = AggregatePriceLogsStep(ctx).transform(df)
-        ti.xcom_push(key='price_data', value=pickle.dumps(price_df))
-        logging.info(f"[{store_mode}] Aggregated price logs for {len(price_df)} products")
-        return len(price_df)
-
-    def join_product_features(**kwargs):
-        ti = kwargs['ti']
-        demand_df = pickle.loads(ti.xcom_pull(key='demand_data'))
-        price_df = pickle.loads(ti.xcom_pull(key='price_data'))
-        ctx = get_context(**kwargs)
-        joined_df = JoinProductFeaturesStep(ctx).transform((demand_df, price_df))
-        ti.xcom_push(key='product_features', value=pickle.dumps(joined_df))
-        logging.info(f"[{store_mode}] Joined features for {len(joined_df)} products")
-        return len(joined_df)
-
-    def apply_surge_pricing(**kwargs):
-        ti = kwargs['ti']
-        product_features = pickle.loads(ti.xcom_pull(key='product_features'))
-        dag_conf = kwargs.get('dag_run').conf if kwargs.get('dag_run') else {}
-
-        data = product_features.rename(columns={'demand_score': 'demand'})
-        surge_pricer = SimpleSurgePricer(
-            high_threshold=dag_conf.get('high_threshold', 10),
-            low_threshold=dag_conf.get('low_threshold', 2),
-            surge_multiplier=dag_conf.get('surge_multiplier', 1.2),
-            discount_multiplier=dag_conf.get('discount_multiplier', 0.9)
-        )
-        surge_pricer.fit(data)
-        data['optimal_price'] = surge_pricer.predict()
-
-        prices_df = data[['productId', 'price', 'base_price', 'optimal_price', 'demand']].rename(columns={
-            'price': 'current_price', 'demand': 'demand_score'
-        })
-        ti.xcom_push(key='predicted_prices', value=pickle.dumps(prices_df))
-        logging.info(f"[{store_mode}] Applied surge pricing for {len(prices_df)} products")
-        return len(prices_df)
-
-    def publish_results(**kwargs):
-        ti = kwargs['ti']
-        prices_df = pickle.loads(ti.xcom_pull(key='predicted_prices'))
-        from lib.model_registry import ModelRegistry
-
-        registry = ModelRegistry()
-        dag_conf = kwargs.get('dag_run').conf if kwargs.get('dag_run') else {}
-
-        metadata = {
-            'timestamp': pd.Timestamp.now().isoformat(),
-            'store_mode': store_mode,
-            'dag_run_id': kwargs['dag_run'].run_id if kwargs.get('dag_run') else 'manual',
-            'pricing_method': 'surge',
-            'high_threshold': dag_conf.get('high_threshold', 10),
-            'low_threshold': dag_conf.get('low_threshold', 2),
-            'surge_multiplier': dag_conf.get('surge_multiplier', 1.2),
-            'discount_multiplier': dag_conf.get('discount_multiplier', 0.9)
-        }
-        registry.publish_prices(prices_df, model_name=f'{store_mode}_latest', metadata=metadata)
-        logging.info(f"[{store_mode}] Published surge pricing for {len(prices_df)} products")
-
-        return {
-            'n_products': len(prices_df),
-            'registry_status': 'success',
-            'store_mode': store_mode,
-            'mean_demand': float(prices_df['demand_score'].mean()) if 'demand_score' in prices_df.columns else None
-        }
-
-    return {
-        'fetch_interactions': fetch_interactions,
-        'fetch_price_logs': fetch_price_logs,
-        'compute_demand': compute_demand,
-        'aggregate_price_logs': aggregate_price_logs,
-        'join_product_features': join_product_features,
-        'apply_surge_pricing': apply_surge_pricing,
-        'publish_results': publish_results,
-    }
-
-
-def create_surge_pricing_dag(store_mode: str) -> DAG:
-    """Factory: generates a surge pricing DAG for a given store_mode."""
-    callables = _make_task_callables(store_mode)
-
-    dag = DAG(
-        f'surge_pricing_{store_mode}',
-        default_args=DEFAULT_ARGS,
-        description=f'Surge pricing pipeline for {store_mode} store mode',
-        schedule_interval='*/15 * * * *',
-        start_date=days_ago(1),
-        catchup=False,
-        max_active_runs=1,
-        tags=['pricing', 'surge', 'research', store_mode],
-    )
-
-    with dag:
-        t_fetch_interactions = PythonOperator(
-            task_id='fetch_interactions',
-            python_callable=callables['fetch_interactions'],
-            provide_context=True,
-        )
-        t_fetch_price_logs = PythonOperator(
-            task_id='fetch_price_logs',
-            python_callable=callables['fetch_price_logs'],
-            provide_context=True,
-        )
-        t_compute_demand = PythonOperator(
-            task_id='compute_demand',
-            python_callable=callables['compute_demand'],
-            provide_context=True,
-        )
-        t_aggregate_prices = PythonOperator(
-            task_id='aggregate_price_logs',
-            python_callable=callables['aggregate_price_logs'],
-            provide_context=True,
-        )
-        t_join_features = PythonOperator(
-            task_id='join_product_features',
-            python_callable=callables['join_product_features'],
-            provide_context=True,
-        )
-        t_surge_pricing = PythonOperator(
-            task_id='apply_surge_pricing',
-            python_callable=callables['apply_surge_pricing'],
-            provide_context=True,
-        )
-        t_publish = PythonOperator(
-            task_id='publish_results',
-            python_callable=callables['publish_results'],
-            provide_context=True,
-        )
-
-        t_fetch_interactions >> t_compute_demand
-        t_fetch_price_logs >> t_aggregate_prices
-        [t_compute_demand, t_aggregate_prices] >> t_join_features >> t_surge_pricing >> t_publish
-
-    return dag
-
-
-# instantiate DAGs for Airflow to discover
-dag_airline = create_surge_pricing_dag('airline')
-dag_hotel = create_surge_pricing_dag('hotel')
--- a/experiments/ml/init.py
+++ b/experiments/ml/init.py
@@ -1,11 +0,0 @@
-from .evals import evaluate
-from .arch import (
-    XGBoostAgentClassifier,
-    LightGBMAgentClassifier
-)
-
-__all__ =[
-    'evaluate',
-    'XGBoostAgentClassifier',
-    'LightGBMAgentClassifier'
-]
--- a/experiments/ml/arch.py
+++ b/experiments/ml/arch.py
@@ -1,122 +0,0 @@
-# sklearn compatible models for agent detection
-from sklearn.base import BaseEstimator, ClassifierMixin
-from procesing.context import PipelineContext
-from typing import Any, Optional, Tuple
-from abc import ABC, abstractmethod
-import xgboost as xgb
-import lightgbm as lgb
-import numpy as np
-import pandas as pd
-
-TASK = 'classification'
-LABELS = ['human', 'agent']
-
-
-class BaseAgentClassifier(BaseEstimator, ClassifierMixin, ABC):
-    """Base class for tree-based agent detection classifiers with common logic"""
-
-    def __init__(self, context: Optional[PipelineContext] = None, n_estimators: int = 200,
-                 max_depth: int = 6, learning_rate: float = 0.05,
-                 early_stopping_rounds: int = 20):
-        self.context = context
-        self.n_estimators = n_estimators
-        self.max_depth = max_depth
-        self.learning_rate = learning_rate
-        self.early_stopping_rounds = early_stopping_rounds
-        self.model_ = None
-        self.feature_names_ = None
-
-    def _to_array(self, X):
-        """Convert pandas structures to numpy arrays"""
-        return X.values if isinstance(X, (pd.DataFrame, pd.Series)) else X
-
-    def _compute_pos_weight(self, y_arr):
-        """Calculate scale_pos_weight for class imbalance handling"""
-        n_neg, n_pos = (y_arr == 0).sum(), (y_arr == 1).sum()
-        return n_neg / n_pos if n_pos > 0 else 1.0
-
-    def _prepare_eval_set(self, eval_set):
-        """Convert eval_set to numpy arrays if needed"""
-        if not eval_set:
-            return None
-        X_val, y_val = eval_set[0]
-        return [(self._to_array(X_val), self._to_array(y_val))]
-
-    @abstractmethod
-    def _build_model(self, scale_pos: float):
-        """Build the underlying model instance (must be implemented by subclasses)"""
-        pass
-
-    @abstractmethod
-    def _fit_with_eval(self, X_arr, y_arr, eval_arr):
-        """Fit model with evaluation set (must be implemented by subclasses)"""
-        pass
-
-    def fit(self, X, y, eval_set=None):
-        X_arr, y_arr = self._to_array(X), self._to_array(y)
-
-        if isinstance(X, pd.DataFrame):
-            self.feature_names_ = X.columns.tolist()
-
-        scale_pos = self._compute_pos_weight(y_arr)
-        self.model_ = self._build_model(scale_pos)
-
-        eval_arr = self._prepare_eval_set(eval_set)
-        if eval_arr:
-            self._fit_with_eval(X_arr, y_arr, eval_arr)
-        else:
-            self.model_.fit(X_arr, y_arr)
-
-        return self
-
-    def predict(self, X):
-        return self.model_.predict(self._to_array(X))
-
-    def predict_proba(self, X):
-        return self.model_.predict_proba(self._to_array(X))
-
-    @property
-    def feature_importances_(self):
-        return self.model_.feature_importances_ if self.model_ else None
-
-
-class XGBoostAgentClassifier(BaseAgentClassifier):
-    """XGBoost binary classifier for agent detection with class imbalance handling"""
-
-    def _build_model(self, scale_pos: float):
-        return xgb.XGBClassifier(
-            n_estimators=self.n_estimators,
-            max_depth=self.max_depth,
-            learning_rate=self.learning_rate,
-            scale_pos_weight=scale_pos,
-            eval_metric='auc',
-            early_stopping_rounds=self.early_stopping_rounds,
-            random_state=42,
-            tree_method='hist',
-            enable_categorical=False
-        )
-
-    def _fit_with_eval(self, X_arr, y_arr, eval_arr):
-        self.model_.fit(X_arr, y_arr, eval_set=eval_arr, verbose=False)
-
-
-class LightGBMAgentClassifier(BaseAgentClassifier):
-    """LightGBM binary classifier for agent detection with class imbalance handling"""
-
-    def _build_model(self, scale_pos: float):
-        return lgb.LGBMClassifier(
-            n_estimators=self.n_estimators,
-            max_depth=self.max_depth,
-            learning_rate=self.learning_rate,
-            scale_pos_weight=scale_pos,
-            metric='auc',
-            random_state=42,
-            verbosity=-1
-        )
-
-    def _fit_with_eval(self, X_arr, y_arr, eval_arr):
-        self.model_.fit(
-            X_arr, y_arr,
-            eval_set=eval_arr,
-            callbacks=[lgb.early_stopping(self.early_stopping_rounds, verbose=False)]
-        )
--- a/experiments/ml/evals.py
+++ b/experiments/ml/evals.py
@@ -1,103 +0,0 @@
-from sklearn.metrics import (accuracy_score, precision_score, recall_score,
-                             f1_score, roc_auc_score, confusion_matrix, roc_curve)
-from torch.utils.tensorboard import SummaryWriter
-from logging import getLogger
-import numpy as np
-import matplotlib.pyplot as plt
-import io
-from PIL import Image
-
-logger = getLogger(__name__)
-
-
-def log_feature_importance(writer, model, feature_names, epoch):
-    """Visualize and log feature importance to TensorBoard"""
-    if not hasattr(model, 'feature_importances_') or model.feature_importances_ is None:
-        return
-
-    importance = model.feature_importances_
-    indices = np.argsort(importance)[::-1][:20]  # top 20
-    top_features = [feature_names[i] for i in indices]
-    top_importance = importance[indices]
-
-    for i, (feat, imp) in enumerate(zip(top_features, top_importance)):
-        writer.add_scalar(f'FeatureImportance/{feat}', imp, epoch)
-
-    fig, ax = plt.subplots(figsize=(10, 8))
-    ax.barh(range(len(top_features)), top_importance, align='center')
-    ax.set_yticks(range(len(top_features)))
-    ax.set_yticklabels(top_features)
-    ax.invert_yaxis()
-    ax.set_xlabel('Importance')
-    ax.set_title(f'Top 20 Feature Importance (Epoch {epoch})')
-    ax.grid(axis='x', alpha=0.3)
-
-    buf = io.BytesIO()
-    plt.tight_layout()
-    plt.savefig(buf, format='png', dpi=100)
-    buf.seek(0)
-    img = Image.open(buf)
-    img_arr = np.array(img)
-    writer.add_image('FeatureImportance/Chart', img_arr, epoch, dataformats='HWC')
-    plt.close()
-
-def evaluate(perdicted_class, predicted_proba, true_class, writer: SummaryWriter, epoch: int):
-    accuracy = accuracy_score(true_class, perdicted_class)
-    precision = precision_score(true_class, perdicted_class, zero_division=0)
-    recall = recall_score(true_class, perdicted_class, zero_division=0)
-    f1 = f1_score(true_class, perdicted_class, zero_division=0)
-    roc_auc = roc_auc_score(true_class, predicted_proba)
-
-    writer.add_scalar('Eval/Accuracy', accuracy, epoch)
-    writer.add_scalar('Eval/Precision', precision, epoch)
-    writer.add_scalar('Eval/Recall', recall, epoch)
-    writer.add_scalar('Eval/F1_Score', f1, epoch)
-    writer.add_scalar('Eval/ROC_AUC', roc_auc, epoch)
-
-    # confusion matrix
-    cm = confusion_matrix(true_class, perdicted_class)
-    tn, fp, fn, tp = cm.ravel()
-    writer.add_scalar('Eval/TrueNeg', tn, epoch)
-    writer.add_scalar('Eval/FalsePos', fp, epoch)
-    writer.add_scalar('Eval/FalseNeg', fn, epoch)
-    writer.add_scalar('Eval/TruePos', tp, epoch)
-
-    # specificity and sensitivity
-    specificity = tn / (tn + fp) if (tn + fp) > 0 else 0
-    sensitivity = recall  # same as recall/TPR
-    writer.add_scalar('Eval/Specificity', specificity, epoch)
-    writer.add_scalar('Eval/Sensitivity', sensitivity, epoch)
-
-    fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(12, 4))
-    ax1.matshow(cm, cmap='Blues', alpha=0.7)
-    for i in range(2):
-        for j in range(2):
-            ax1.text(j, i, str(cm[i, j]), ha='center', va='center', fontsize=14)
-    ax1.set_xlabel('Predicted')
-    ax1.set_ylabel('True')
-    ax1.set_title(f'Confusion Matrix (Epoch {epoch})')
-    ax1.set_xticks([0, 1])
-    ax1.set_yticks([0, 1])
-    ax1.set_xticklabels(['Human', 'Agent'])
-    ax1.set_yticklabels(['Human', 'Agent'])
-
-    # ROC curve
-    fpr, tpr, _ = roc_curve(true_class, predicted_proba)
-    ax2.plot(fpr, tpr, label=f'AUC={roc_auc:.3f}', linewidth=2)
-    ax2.plot([0, 1], [0, 1], 'k--', label='Random')
-    ax2.set_xlabel('False Positive Rate')
-    ax2.set_ylabel('True Positive Rate')
-    ax2.set_title('ROC Curve')
-    ax2.legend()
-    ax2.grid(alpha=0.3)
-
-    buf = io.BytesIO()
-    plt.tight_layout()
-    plt.savefig(buf, format='png', dpi=100)
-    buf.seek(0)
-    img = Image.open(buf)
-    img_arr = np.array(img)
-    writer.add_image('Eval/Metrics', img_arr, epoch, dataformats='HWC')
-    plt.close()
-
-    logger.info(f"Eval {epoch}: Acc={accuracy:.4f} Prec={precision:.4f} Rec={recall:.4f} F1={f1:.4f} AUC={roc_auc:.4f}")
--- a/experiments/ml/requirements.txt
+++ b/experiments/ml/requirements.txt
@@ -1,6 +0,0 @@
-torch
-tensorboard
-fastparquet
-pyarrow
-xgboost
-lightgbm
--- a/experiments/ml/train.py
+++ b/experiments/ml/train.py
@@ -1,137 +0,0 @@
-from torch.utils.tensorboard import SummaryWriter
-from sklearn.model_selection import train_test_split
-from logging import getLogger
-from pathlib import Path
-import pandas as pd
-import numpy as np
-import joblib
-from datetime import datetime
-from ml.evals import evaluate, log_feature_importance
-from ml.arch import XGBoostAgentClassifier, LightGBMAgentClassifier, LABELS
-
-logger = getLogger(__name__)
-
-FEATURE_COLS_EXCLUDE = ['sessionId', 'experimentId', 'is_agent', 'xp_human_only', 'xp_market_mode', 'browser_family']
-RUNS_DIR = Path('ml/runs')
-CHECKPOINTS_DIR = Path('ml/checkpoints')
-
-
-def prepare_data(df):
-    """
-    Prepare feature matrix and labels from raw dataframe
-    Handles missing labels, feature selection, and categorical encoding
-    Returns: (X, y, feature_cols)
-    """
-    # drop rows with missing labels
-    n_before = len(df)
-    df = df[df['is_agent'].notna()].copy()
-    n_dropped = n_before - len(df)
-    if n_dropped > 0:
-        logger.warning(f"Dropped {n_dropped} sessions with missing labels")
-
-    if len(df) == 0:
-        logger.error("No labeled data available")
-        return None, None, None
-
-    feature_cols = [c for c in df.columns if c not in FEATURE_COLS_EXCLUDE]
-
-    # handle categorical browser_family via one-hot encoding
-    if 'browser_family' in df.columns:
-        browser_dummies = pd.get_dummies(df['browser_family'], prefix='browser', drop_first=True)
-        df = pd.concat([df, browser_dummies], axis=1)
-        feature_cols.extend(browser_dummies.columns.tolist())
-
-    X = df[feature_cols].fillna(0)
-    y = df['is_agent'].astype(int)
-
-    return X, y, feature_cols
-
-
-def train(data_path=None, model_type='xgboost', test_size=0.2, random_state=42,
-          n_estimators=200, max_depth=6, learning_rate=0.05):
-    """
-    Train agent detection classifier
-    Args:
-        data_path: path to labeled feature matrix CSV or parquet
-        model_type: 'xgboost' or 'lightgbm'
-        test_size: fraction for test split
-        random_state: seed for reproducibility
-    """
-    RUNS_DIR.mkdir(exist_ok=True)
-    CHECKPOINTS_DIR.mkdir(exist_ok=True)
-
-    run_name = f"{model_type}_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
-    writer = SummaryWriter(log_dir=RUNS_DIR / run_name)
-    logger.info(f"Starting training run: {run_name}")
-
-    # load data
-    if data_path is None:
-        logger.error("data_path required")
-        return
-    df = pd.read_parquet(data_path)
-    logger.info(f"Loaded {len(df)} sessions from {data_path}")
-
-    # prepare features and labels
-    if 'is_agent' not in df.columns:
-        logger.error("Missing is_agent column")
-        return
-
-    X, y, feature_cols = prepare_data(df)
-    if X is None:
-        return
-
-    # class distribution
-    n_agents = y.sum()
-    n_humans = (y == 0).sum()
-    logger.info(f"Class distribution: {n_humans} humans, {n_agents} agents" + (f" (ratio {n_humans / n_agents:.2f})" if n_agents > 0 else ""))
-
-    # train/test split with stratification
-    X_train, X_test, y_train, y_test = train_test_split(
-        X, y, test_size=test_size, random_state=random_state, stratify=y
-    )
-    logger.info(f"Train: {len(X_train)}, Test: {len(X_test)}")
-
-    # init model
-    if model_type == 'xgboost':
-        model = XGBoostAgentClassifier(
-            n_estimators=n_estimators,
-            max_depth=max_depth,
-            learning_rate=learning_rate
-        )
-    elif model_type == 'lightgbm':
-        model = LightGBMAgentClassifier(
-            n_estimators=n_estimators,
-            max_depth=max_depth,
-            learning_rate=learning_rate
-        )
-    else:
-        logger.error(f"Unknown model type: {model_type}")
-        return
-
-    # train with eval set for early stopping
-    model.fit(X_train, y_train, eval_set=[(X_test, y_test)])
-    logger.info("Training complete")
-
-    # evaluate on test set
-    y_pred = model.predict(X_test)
-    y_prob = model.predict_proba(X_test)[:, 1]
-
-    evaluate(y_pred, y_prob, y_test, writer, epoch=0)
-
-    # log feature importance
-    log_feature_importance(writer, model, X.columns.tolist(), epoch=0)
-
-    # save model
-    model_path = CHECKPOINTS_DIR / f"{run_name}.pkl"
-    joblib.dump({'model': model, 'feature_cols': X.columns.tolist(), 'run_name': run_name}, model_path)
-    logger.info(f"Model saved to {model_path}")
-
-    writer.close()
-    return model, X.columns.tolist()
-
-
-if __name__ == "__main__":
-    import sys
-    data_path = sys.argv[1]
-    model_type = sys.argv[2] if len(sys.argv) > 2 else 'xgboost'
-    train(data_path, model_type=model_type)
--- a/experiments/procesing/pipelines.py
+++ b/experiments/procesing/pipelines.py
@@ -2,7 +2,6 @@ from sklearn.pipeline import Pipeline
 import pandas as pd
 from procesing.context import PipelineContext
 from procesing.providers import SupabaseProvider, BackendAPIProvider
-import os
 from procesing.steps import (
    FetchInteractionsStep,
    FetchPriceLogsStep,
@@ -13,13 +12,11 @@ from procesing.steps import (
    ChunkByTimeWindowStep,
    ComputeDemandForChunksStep,
    AggregatePriceLogsStep,
+    # BuildStateSpaceStep,
    FitPricingFunctionStep,
    PredictPricesStep,
    ComputeDemandStep,
-    JoinProductFeaturesStep,
-    ExtractSessionFeaturesStep,
-    JoinLabelsStep,
-    ValidateDataStep,
+    JoinProductFeaturesStep
 )
 from procesing.pricers import SimpleSurgePricer

@@ -109,66 +106,33 @@ def full_pipeline(context: PipelineContext,
    return product_features_df, optimal_prices_df


-def ml_training_pipeline(context: PipelineContext) -> pd.DataFrame:
-    """
-    Build labeled session-level feature matrix for ML model training.
-    Pipeline: fetch -> validate -> extract features -> join labels
-
-    Returns:
-        DataFrame with ~25 features per session + is_agent label
-        Columns: sessionId, experimentId, temporal/behavioral/product/ua features, is_agent
-    """
-    # fetch raw interactions
-    interactions_df = FetchInteractionsStep(context).transform(None)
-
-    # validate data quality (report cached in context)
-    interactions_df = ValidateDataStep(context).transform(interactions_df)
-    if interactions_df.empty:
-        return pd.DataFrame()
-
-    # extract vectorized session features
-    features_df = ExtractSessionFeaturesStep(context).transform(interactions_df)
-    if features_df.empty:
-        return pd.DataFrame()
-
-    # join experiment labels (is_agent = ~xp_human_only)
-    labeled_df = JoinLabelsStep(context).transform(features_df)
-
-    return labeled_df
-
-


 if __name__ == '__main__':

-    class ExperimentsProvider(SupabaseProvider, BackendAPIProvider):
+    class Provider(SupabaseProvider, BackendAPIProvider):
+        def __init__(self, backend_url: str):
+            SupabaseProvider.__init__(self)
+            BackendAPIProvider.__init__(self, backend_url=backend_url)
+
+
+    class HistoricalProvider(SupabaseProvider, BackendAPIProvider):
        def fetch_kafka_topic(self, topic: str) -> pd.DataFrame:
-            base_path = "/home/velocitatem/Documents/Projects/PHANTOM/experiments/collected_data/" # os.path.join(os.path.dirname(__file__), "collected_data")
-            if not os.path.isdir(base_path):
-                return pd.DataFrame()
+            path = "/home/velocitatem/Documents/Projects/PHANTOM/experiments/collected_data/858c61ab-0a7f-4595-ae49-33f4365517b9/"
+            interactions_file = "messages(2).json"
+            prices_file = "messages(3).json"

-            files = {"user-interactions": "int.json", "price-logs": "price.json"}
-            file_to_read = files.get(topic, files["user-interactions"])
-            frames = []
+            data = pd.read_json(path + (interactions_file if topic == "user-interactions" else prices_file))
+            data = [r['payload'] for r in data['value'].to_list()]
+            data = pd.DataFrame(data)
+            return data

-            for d in os.listdir(base_path):
-                full_path = os.path.join(base_path, d, file_to_read)
-                if not os.path.isfile(full_path):
-                    continue
-                try:
-                    data = pd.read_json(full_path)
-                    payloads = pd.DataFrame([r['payload'] for r in data['value'].to_list()])
-                    frames.append(payloads)
-                except Exception as e:
-                    print(f"Warning: Could not process {full_path}: {e}")

-            return pd.concat(frames, ignore_index=True) if frames else pd.DataFrame()
+    # example run
+    context = PipelineContext(
+        provider=HistoricalProvider(),
+        store_mode='hotel',
+    )

-    # demo: run ML training pipeline
-    context = PipelineContext(provider=ExperimentsProvider(), store_mode='hotel')
-    features = ml_training_pipeline(context)
-    print(f"Feature matrix: {features.shape}")
-    print(features.head())
-    print(features.info())
-
-    features.to_parquet("features.parquet")
+    product_features, prices = full_pipeline(context)
+    print(prices.to_string())
--- a/experiments/procesing/providers/supabase.py
+++ b/experiments/procesing/providers/supabase.py
@@ -18,17 +18,10 @@ class SupabaseProvider(DataProvider):
        self.supabase: Client = create_client(self.supabase_url, self.supabase_key)

    def fetch_products(self, store_mode: str) -> pd.DataFrame:
-        # hotel uses room_type, airline uses flight_type; select all and normalize
-        resp = self.supabase.table(f'{store_mode}_products').select("*").execute()
-        if not resp.data:
-            return pd.DataFrame()
-        df = pd.DataFrame(resp.data)
-        # normalize type column: hotel has room_type, airline has flight_type
-        if 'room_type' in df.columns:
-            df['product_type'] = df['room_type']
-        elif 'flight_type' in df.columns:
-            df['product_type'] = df['flight_type']
-        return df
+        resp = self.supabase.table(f'{store_mode}_products').select(
+            "id, room_type, date_index, metadata, availability"
+        ).execute()
+        return pd.DataFrame(resp.data) if resp.data else pd.DataFrame()

    def fetch_experiments(self, experiment_ids: List[str]) -> pd.DataFrame:
        if not experiment_ids:
--- a/experiments/procesing/steps/init.py
+++ b/experiments/procesing/steps/init.py
@@ -6,11 +6,7 @@ from procesing.steps.chunk import ChunkByTimeWindowStep
 from procesing.steps.demand import ComputeDemandStep, ComputeDemandForChunksStep
 from procesing.steps.elasticity import AggregatePriceLogsStep
 from procesing.steps.pricing import FitPricingFunctionStep, PredictPricesStep
-from procesing.steps.session import (
-    ExtractSessionFeaturesStep, JoinLabelsStep, ValidateDataStep,
-    TemporalFeatureStep, BehavioralFeatureStep, ProductFeatureStep, UserAgentFeatureStep,
-    _extract_features_for_session
-)
+from procesing.steps.session import ExtractSessionFeaturesStep, _extract_features_for_session

 __all__ = [
    'BaseContextStep',
@@ -29,11 +25,5 @@ __all__ = [
    'FitPricingFunctionStep',
    'PredictPricesStep',
    'ExtractSessionFeaturesStep',
-    'JoinLabelsStep',
-    'ValidateDataStep',
-    'TemporalFeatureStep',
-    'BehavioralFeatureStep',
-    'ProductFeatureStep',
-    'UserAgentFeatureStep',
    '_extract_features_for_session',
 ]
--- a/experiments/procesing/steps/base.py
+++ b/experiments/procesing/steps/base.py
@@ -1,7 +1,6 @@
 from abc import ABC, abstractmethod
 from sklearn.base import BaseEstimator, TransformerMixin
 from procesing.context import PipelineContext
-from typing import Any

 class BaseContextStep(BaseEstimator, TransformerMixin, ABC):
    """
@@ -17,7 +16,7 @@ class BaseContextStep(BaseEstimator, TransformerMixin, ABC):
        return self

    @abstractmethod
-    def transform(self, X) -> Any:
+    def transform(self, X):
        """Transform input using context. Must be implemented by subclass."""
        pass

--- a/experiments/procesing/steps/elasticity.py
+++ b/experiments/procesing/steps/elasticity.py
@@ -7,12 +7,12 @@ class AggregatePriceLogsStep(BaseContextStep):
    """
    Aggregate price logs into time windows using VECTORIZED operations.
    Input: price_logs_df
-    Output: DataFrame with columns [productId, price]
+    Output: list of price chunks with [productId, price]
    """

    def transform(self, price_logs_df: pd.DataFrame):
        if price_logs_df.empty:
-            return pd.DataFrame(columns=['productId', 'price'])
+            return []

        df = price_logs_df.copy()
        ts_col = self.context.config.get('ts_col', 'ts')
--- a/experiments/procesing/steps/fetch.py
+++ b/experiments/procesing/steps/fetch.py
@@ -2,7 +2,7 @@ import pandas as pd
 from procesing.steps.base import BaseContextStep

 class FetchInteractionsStep(BaseContextStep):
-    """Fetch raw interaction data from Kafka topic with optional time and store_mode filtering"""
+    """Fetch raw interaction data from Kafka topic with optional time filtering"""

    def __init__(self, context, lookback: str = None):
        super().__init__(context)
@@ -24,10 +24,6 @@ class FetchInteractionsStep(BaseContextStep):
        # drop all where page has /admin/
        df = df[~df['page'].str.contains('/admin/', na=False)]

-        # filter by store_mode from context
-        if 'storeMode' in df.columns:
-            df = df[df['storeMode'] == self.context.store_mode]
-
        # Remap dateIndex if present
        if 'metadata_dateIndex' in df.columns:
            df['dateIndex'] = df['metadata_dateIndex'].astype('Int64')
@@ -42,7 +38,7 @@ class FetchInteractionsStep(BaseContextStep):


 class FetchPriceLogsStep(BaseContextStep):
-    """Fetch price log data from Kafka topic with optional time and store_mode filtering"""
+    """Fetch price log data from Kafka topic with optional time filtering"""

    def __init__(self, context, lookback: str = None):
        super().__init__(context)
@@ -54,10 +50,6 @@ class FetchPriceLogsStep(BaseContextStep):
        if df.empty:
            return df

-        # filter by store_mode from context
-        if 'storeMode' in df.columns:
-            df = df[df['storeMode'] == self.context.store_mode]
-
        # Apply time filtering if lookback specified
        if self.lookback and 'ts' in df.columns:
            df['ts'] = pd.to_datetime(df['ts'])
--- a/experiments/procesing/steps/session.py
+++ b/experiments/procesing/steps/session.py
@@ -1,261 +1,159 @@
 """
-Session feature extraction for ML training pipeline.
+Session feature extraction for S_t component of state space.
+Computes behavioral signals from interaction data already in pipeline.
 """
 import pandas as pd
 import numpy as np
-import re
-from typing import Dict, Any
+from typing import Optional, Dict, Any
+from collections import Counter
 from procesing.steps.base import BaseContextStep

-EVENT_CATS = {
-    'page_view': ['page_view'],
-    'item_view': ['view_item_page', 'learn_more_about_item'],
-    'cart_add': ['add_item_to_cart'],
-    'purchase': ['purchase', 'checkout_complete'],
-    'hover': ['hover_over_title', 'hover_over_paragraph', 'hover_over_link', 'hover_over_button'],
-    # 'filter': ['filter', 'search', 'apply_filter'],
-}
-HEADLESS_RE = re.compile(r'HeadlessChrome|Headless|PhantomJS', re.I)
-AUTOMATION_RE = re.compile(r'Selenium|Playwright|Puppeteer|WebDriver|chromedriver|geckodriver', re.I)
-BROWSER_PATTERNS = [('Chrome', r'Chrome/[\d.]+'), ('Firefox', r'Firefox/[\d.]+'),
-                    ('Safari', r'Safari/[\d.]+'), ('Edge', r'Edg/[\d.]+')]
+def _extract_features_for_session(session_df: pd.DataFrame, session_timeout_sec: float = 900) -> Dict[str, Any]:
+    """Compute features for single session.
+
+    Args:
+        session_df: interaction events for this session
+        session_timeout_sec: max gap between events before resetting duration (default 900s = 15min)
+    """
+    features = {}
+
+    # basic counts
+    features['total_interactions'] = len(session_df)
+
+    event_counts = session_df['eventName'].value_counts().to_dict()
+    features['page_views'] = event_counts.get('page_view', 0) + event_counts.get('view_item_page', 0)
+    features['item_views'] = event_counts.get('view_item_page', 0)
+    features['searches'] = event_counts.get('search', 0)
+    features['cart_adds'] = event_counts.get('add_item_to_cart', 0)
+
+    # hover events
+    hover_events = ['hover_over_title', 'hover_over_paragraph', 'hover_over_link', 'hover_over_button']
+    features['hovers'] = sum(event_counts.get(ev, 0) for ev in hover_events)
+
+    # product-level signals
+    product_ids = session_df['productId'].dropna()
+    features['unique_products_viewed'] = product_ids.nunique()
+
+    if len(product_ids) > 0:
+        product_view_counts = Counter(product_ids)
+        features['product_view_depth'] = max(product_view_counts.values())
+    else:
+        features['product_view_depth'] = 0
+
+    # temporal features with session timeout logic
+    if 'ts' in session_df.columns:
+        timestamps = session_df['ts'].sort_values()
+
+        # compute active duration considering timeout gaps
+        if len(timestamps) > 1:
+            time_diffs = timestamps.diff().dropna().dt.total_seconds()
+            # only count gaps shorter than timeout towards active session duration
+            active_diffs = time_diffs[time_diffs <= session_timeout_sec]
+            features['session_duration_sec'] = active_diffs.sum() if len(active_diffs) > 0 else 0.0
+
+            features['avg_time_between_events'] = time_diffs.mean()
+            features['std_time_between_events'] = time_diffs.std()
+        else:
+            features['session_duration_sec'] = 0.0
+            features['avg_time_between_events'] = 0.0
+            features['std_time_between_events'] = 0.0
+
+        if features['session_duration_sec'] > 0:
+            features['interaction_velocity'] = (features['total_interactions'] / features['session_duration_sec']) * 60
+        else:
+            features['interaction_velocity'] = 0.0
+    else:
+        features['session_duration_sec'] = 0.0
+        features['interaction_velocity'] = 0.0
+        features['avg_time_between_events'] = 0.0
+        features['std_time_between_events'] = 0.0
+
+    # cart/conversion signals
+    features['cart_to_view_ratio'] = features['cart_adds'] / features['item_views'] if features['item_views'] > 0 else 0.0
+
+    return features


-def _get_browser(s: str) -> str:
-    if pd.isna(s): return 'Unknown'
-    for name, pat in BROWSER_PATTERNS:
-        if re.search(pat, s): return name
-    return 'Other'
+def _apply_to_slice(df: pd.DataFrame) -> pd.DataFrame:
+    """Apply feature extraction to sliding window of interactions."""
+    # add columns of all features at each step
+    new_cols = ["total_interactions", "page_views", "item_views", "searches",
+                "cart_adds", "hovers", "unique_products_viewed", "product_view_depth",
+                "session_duration_sec", "interaction_velocity",
+                "avg_time_between_events", "std_time_between_events",
+                "cart_to_view_ratio"]
+    for col in new_cols: df[col] = np.nan
+    for idx in range(1, len(df) + 1):
+        features = _extract_features_for_session(df.iloc[:idx])
+        # fillna kinda meh
+        features = { k: (v if not pd.isna(v) else 0.0) for k, v in features.items() }
+        for col in new_cols:
+            df.at[df.index[idx - 1], col] = features[col]
+        #print(f"Processed {idx}/{len(df)} events for session {df['sessionId'].iloc[0]}")
+    return df
+
+class BuildStateSpaceStep(BaseContextStep):
+    """
+    Build state space representation S_t from session features.
+
+    Input: session_features DataFrame
+    Output: state_space_df DataFrame with S_t vectors
+    """
+
+    def transform(self, rich_dataset: pd.DataFrame) -> pd.DataFrame:
+        # check if features are present
+        required_cols = ["total_interactions", "page_views", "item_views", "searches",
+                         "cart_adds", "hovers", "unique_products_viewed", "product_view_depth",
+                         "session_duration_sec", "interaction_velocity",
+                         "avg_time_between_events", "std_time_between_events",
+                         "cart_to_view_ratio"]
+        if not all(col in rich_dataset.columns for col in required_cols):
+            raise ValueError("Missing required columns for feature extraction.")
+        if rich_dataset.empty:
+            return pd.DataFrame()


-class TemporalFeatureStep(BaseContextStep):
-    """Vectorized time-based features: durations, velocities, gaps."""
-
-    def __init__(self, context, timeout_sec: float = 900, velocity_window: str = '5min'):
-        super().__init__(context)
-        self.timeout_sec = timeout_sec
-        self.velocity_window = velocity_window
-
-    def transform(self, X: pd.DataFrame) -> pd.DataFrame:
-        df = X.copy()
-        if df.empty or 'ts' not in df.columns:
-            return pd.DataFrame(columns=pd.Series(['sessionId']))
-
-        df['ts_dt'] = pd.to_datetime(df['ts'])
-        df = df.sort_values(['sessionId', 'ts_dt'])
-        df['time_diff'] = df.groupby('sessionId')['ts_dt'].diff().dt.total_seconds()
-        df['active_diff'] = df['time_diff'].where(df['time_diff'] <= self.timeout_sec, 0)
-
-        agg = df.groupby('sessionId').agg(
-            session_duration_sec=('active_diff', 'sum'),
-            total_interactions=('sessionId', 'count'),
-            avg_time_between_events=('time_diff', 'mean'),
-            std_time_between_events=('time_diff', 'std'),
-            min_time_between_events=('time_diff', 'min'),
-            session_start_hour=('ts_dt', lambda x: x.min().hour),
-        ).reset_index()
-        agg['std_time_between_events'] = agg['std_time_between_events'].fillna(0)
-        agg['interaction_velocity'] = np.where(
-            agg['session_duration_sec'] > 0,
-            (agg['total_interactions'] / agg['session_duration_sec']) * 60, 0)
-
-        vel = df.set_index('ts_dt').groupby('sessionId').resample(self.velocity_window, include_groups=False).size()
-        max_velocity = vel.groupby('sessionId').max().rename('max_velocity_5min')
-        agg = agg.merge(max_velocity, on='sessionId', how='left')
-        agg['max_velocity_5min'] = agg['max_velocity_5min'].fillna(0)
-        return agg
+        # For simplicity, we return as is
+        return rich_dataset.copy()


-class BehavioralFeatureStep(BaseContextStep):
-    """Vectorized event counts and ratios per session."""
-
-    def transform(self, X: pd.DataFrame) -> pd.DataFrame:
-        df = X.copy()
-        if df.empty or 'eventName' not in df.columns:
-            return pd.DataFrame(columns=pd.Series(['sessionId']))
-
-        for cat, events in EVENT_CATS.items():
-            df[f'is_{cat}'] = df['eventName'].isin(events)
-        df['is_hover'] = df['is_hover'] | df['eventName'].str.startswith('hover_over_')
-
-        agg = df.groupby('sessionId').agg(
-            total_events=('eventName', 'count'), unique_pages=('page', 'nunique'),
-            page_views=('is_page_view', 'sum'), item_views=('is_item_view', 'sum'),
-            cart_adds=('is_cart_add', 'sum'), purchases=('is_purchase', 'sum'),
-            hover_events=('is_hover', 'sum'),
-            # filter_events=('is_filter', 'sum'),
-        ).reset_index()
-        agg['cart_to_view_ratio'] = np.where(agg['item_views'] > 0, agg['cart_adds'] / agg['item_views'], 0)
-        agg['conversion_rate'] = np.where(agg['item_views'] > 0, agg['purchases'] / agg['item_views'], 0)
-        agg['hover_intensity'] = np.where(agg['total_events'] > 0, agg['hover_events'] / agg['total_events'], 0)
-        return agg
-
-
-class ProductFeatureStep(BaseContextStep):
-    """Vectorized product interaction features: diversity, depth, price sensitivity."""
-
-    def transform(self, X: pd.DataFrame) -> pd.DataFrame:
-        df = X.copy()
-        if df.empty:
-            return pd.DataFrame(columns=pd.Series(['sessionId']))
-        price_col = next((c for c in ['metadata_base_price', 'metadata_price', 'base_price'] if c in df.columns), None)
-        df['price_seen'] = pd.to_numeric(df[price_col], errors='coerce') if price_col else np.nan
-
-        prod_df = df[df['productId'].notna()]
-        if prod_df.empty:
-            return pd.DataFrame(columns=pd.Series(['sessionId', 'unique_products_viewed', 'product_view_depth', 'avg_price_seen', 'min_price_seen', 'max_price_seen', 'price_range']))
-
-        agg = prod_df.groupby('sessionId').agg(
-            unique_products_viewed=('productId', 'nunique'),
-            product_view_depth=('productId', lambda x: x.value_counts().iloc[0] if len(x) > 0 else 0),
-            avg_price_seen=('price_seen', 'mean'), min_price_seen=('price_seen', 'min'),
-            max_price_seen=('price_seen', 'max'),
-        ).reset_index()
-        agg['price_range'] = (agg['max_price_seen'] - agg['min_price_seen']).fillna(0)
-        return agg
-
-
-class UserAgentFeatureStep(BaseContextStep):
-    """Parse userAgent into bot-detection signals."""
-
-    def transform(self, X: pd.DataFrame) -> pd.DataFrame|pd.Series:
-        df = X.copy()
-        if df.empty or 'userAgent' not in df.columns:
-            return pd.DataFrame(columns=pd.Series(['sessionId']))
-
-        ua = df.groupby('sessionId')['userAgent'].first().reset_index()
-        ua['is_headless'] = ua['userAgent'].str.contains(HEADLESS_RE, na=False)
-        ua['is_automation'] = ua['userAgent'].str.contains(AUTOMATION_RE, na=False)
-        ua['browser_family'] = ua['userAgent'].apply(_get_browser)
-        return ua[['sessionId', 'is_headless', 'is_automation', 'browser_family']]


 class ExtractSessionFeaturesStep(BaseContextStep):
    """
-    Vectorized session feature extraction - replaces O(n^2) per-row loop.
-    Input: interactions_df
-    Output: session-level feature matrix
+    Extract session-level behavioral features from interaction logs.
+
+    Input: interactions_df (user-interactions from earlier pipeline step)
+    Output: interactions_df with added session feature columns
    """

-    def transform(self, X: pd.DataFrame) -> pd.DataFrame:
-        if X.empty:
+    def transform(self, interactions_df: pd.DataFrame) -> pd.DataFrame:
+        if interactions_df.empty:
            return pd.DataFrame()
-        df = X.copy()

-        # run all feature steps and merge on sessionId
-        temporal = TemporalFeatureStep(self.context).transform(df)
-        behavioral = BehavioralFeatureStep(self.context).transform(df)
-        product = ProductFeatureStep(self.context).transform(df)
-        ua = UserAgentFeatureStep(self.context).transform(df)
+        # ensure timestamp column
+        if 'ts' in interactions_df.columns:
+            interactions_df = interactions_df.copy()
+            interactions_df['ts'] = pd.to_datetime(interactions_df['ts'])

-        result = temporal
-        for other in [behavioral, product, ua]:
-            if not other.empty and 'sessionId' in other.columns:
-                result = result.merge(other, on='sessionId', how='left')
+        # group by session and compute features
+        session_features = []
+        for session_id, session_df in interactions_df.groupby('sessionId'):
+            new_slice = _apply_to_slice(session_df.sort_values('ts'))
+            session_features.append(new_slice)

-        # carry forward experimentId for label joining
-        if 'experimentId' in df.columns:
-            exp_map = df.groupby('sessionId')['experimentId'].first()
-            result = result.merge(exp_map, on='sessionId', how='left')
-
-        return result
+        return pd.concat(session_features, ignore_index=True)


-class JoinLabelsStep(BaseContextStep):
+
+class FilterSessionInteractionsStep(BaseContextStep):
    """
-    Join experiment labels to session features.
-    Input: (features_df, experiments_df) or features_df (fetches experiments)
-    Output: labeled feature matrix with is_agent column
+    Filter interactions DataFrame to specific session.
+
+    Input: (interactions_df, session_id)
+    Output: interactions_df filtered to session_id
    """

-    def transform(self, X : tuple) -> pd.DataFrame:
-        data = X;
-        if isinstance(data, tuple):
-            features_df, experiments_df = data
-        else:
-            features_df = data
-            if 'experimentId' not in features_df.columns:
-                return features_df
-            exp_ids = features_df['experimentId'].dropna().unique().tolist()
-            experiments_df = self.context.provider.fetch_experiments(exp_ids) if exp_ids else pd.DataFrame()
-
-        if features_df.empty:
-            return features_df
-        if experiments_df.empty:
-            features_df['is_agent'] = np.nan
-            return features_df
-
-        exp = experiments_df.copy()
-        if 'id' in exp.columns:
-            exp = exp.rename(columns={'id': 'experimentId'})
-        if 'xp_human_only' in exp.columns:
-            exp['is_agent'] = ~exp['xp_human_only']
-
-        cols = ['experimentId'] + [c for c in ['is_agent', 'xp_human_only', 'xp_market_mode'] if c in exp.columns]
-        return features_df.merge(exp[cols].drop_duplicates(), on='experimentId', how='left')
-
-
-class ValidateDataStep(BaseContextStep):
-    """
-    Data quality checks before training.
-    Input: df
-    Output: df (unchanged, but logs validation report to context)
-    """
-    REQUIRED = ['sessionId', 'eventName', 'ts']
-
-    def transform(self, X: pd.DataFrame) -> pd.DataFrame:
-        df = X.copy()
-        report = {'status': 'valid', 'rows': len(df), 'sessions': 0}
-        if df.empty:
-            report['status'] = 'empty'
-            self.context.cache('validation_report', report)
-            return df
-
-        missing = [c for c in self.REQUIRED if c not in df.columns]
-        if missing:
-            report['status'] = 'invalid'
-            report['missing_cols'] = missing
-
-        report['sessions'] = df['sessionId'].nunique() if 'sessionId' in df.columns else 0
-        report['null_sessions'] = int(df['sessionId'].isna().sum()) if 'sessionId' in df.columns else 0
-        if 'experimentId' in df.columns:
-            report['null_experiments'] = int(df['experimentId'].isna().sum())
-
-        self.context.cache('validation_report', report)
-        return df
-
-
-# legacy compat - kept for backwards compatibility with existing code
-def _extract_features_for_session(session_df: pd.DataFrame, session_timeout_sec: float = 900) -> Dict[str, Any]:
-    """Single-session feature extraction (legacy interface)."""
-    defaults = {k: 0 for k in ['total_interactions', 'page_views', 'item_views', 'searches',
-                               'cart_adds', 'hovers', 'unique_products_viewed', 'product_view_depth',
-                               'session_duration_sec', 'interaction_velocity',
-                               'avg_time_between_events', 'std_time_between_events', 'cart_to_view_ratio']}
-    if session_df.empty:
-        return defaults
-
-    session_df = session_df.copy()
-    if 'sessionId' not in session_df.columns:
-        session_df['sessionId'] = 'tmp'
-
-    # use a dummy context for the steps
-    class DummyCtx: config = {} # should maybe inherit but whatever
-    ctx = DummyCtx()
-
-    t = TemporalFeatureStep(ctx, timeout_sec=session_timeout_sec).transform(session_df)
-    b = BehavioralFeatureStep(ctx).transform(session_df)
-    p = ProductFeatureStep(ctx).transform(session_df)
-
-    result = {}
-    for df in [t, b, p]:
-        if not df.empty:
-            for col in df.columns:
-                if col != 'sessionId':
-                    result[col] = df[col].iloc[0] if len(df) > 0 else 0
-
-    remap = {'hover_events': 'hovers', 'filter_events': 'searches', 'unique_pages': 'unique_pages_visited'}
-    for old, new in remap.items():
-        if old in result:
-            result[new] = result.pop(old)
-    return result
+    def transform(self, data: tuple) -> pd.DataFrame:
+        interactions_df, session_id = data
+        return interactions_df[interactions_df['sessionId'] == session_id].copy()
--- a/experiments/procesing/tests/conftest.py
+++ b/experiments/procesing/tests/conftest.py
@@ -144,7 +144,7 @@ def mock_price_logs_raw_kafka():
                    'price': 162.47,
                    'sessionId': 'd423ce8a-77aa-4c9a-94d4-d1adddcc3472',
                    'experimentId': '53aefd07-f66a-4d7f-ba8b-7ea1fc562d35',
-                    'storeMode': 'hotel',
+                    'storeMode': 'shop',
                    'ts': '2025-11-25T21:05:57.967Z'
                }
            }
@@ -157,7 +157,7 @@ def mock_price_logs_raw_kafka():
                    'price': 743.49,
                    'sessionId': 'd423ce8a-77aa-4c9a-94d4-d1adddcc3472',
                    'experimentId': '53aefd07-f66a-4d7f-ba8b-7ea1fc562d35',
-                    'storeMode': 'hotel',
+                    'storeMode': 'shop',
                    'ts': '2025-11-25T21:05:57.993Z'
                }
            }
@@ -170,7 +170,7 @@ def mock_price_logs_raw_kafka():
                    'price': 163.87,
                    'sessionId': 'd423ce8a-77aa-4c9a-94d4-d1adddcc3472',
                    'experimentId': '53aefd07-f66a-4d7f-ba8b-7ea1fc562d35',
-                    'storeMode': 'hotel',
+                    'storeMode': 'shop',
                    'ts': '2025-11-25T21:05:58.009Z'
                }
            }
@@ -183,7 +183,7 @@ def mock_price_logs_raw_kafka():
                    'price': 397.46,
                    'sessionId': 'd423ce8a-77aa-4c9a-94d4-d1adddcc3472',
                    'experimentId': '53aefd07-f66a-4d7f-ba8b-7ea1fc562d35',
-                    'storeMode': 'hotel',
+                    'storeMode': 'shop',
                    'ts': '2025-11-25T21:05:58.049Z'
                }
            }
@@ -196,7 +196,7 @@ def mock_price_logs_raw_kafka():
                    'price': 401.66,
                    'sessionId': 'd423ce8a-77aa-4c9a-94d4-d1adddcc3472',
                    'experimentId': '53aefd07-f66a-4d7f-ba8b-7ea1fc562d35',
-                    'storeMode': 'hotel',
+                    'storeMode': 'shop',
                    'ts': '2025-11-25T21:06:08.864Z'
                }
            }
@@ -222,7 +222,7 @@ def mock_experiments():
        'created_at': pd.to_datetime(['2025-11-25T20:00:00Z', '2025-11-26T10:00:00Z']),
        'subject_name': ['Session A', 'Session B'],
        'xp_human_only': [True, False],
-        'xp_market_mode': ['hotel', 'airline'],
+        'xp_market_mode': ['hotel', 'shop'],
        'xp_task_id': [None, None]
    })

@@ -269,13 +269,3 @@ def empty_context(empty_provider):
        store_mode='hotel',
        window_size='30s'
    )
-
-
-@pytest.fixture
-def session_interactions(mock_interactions):
-    """Enriched interaction data for session feature extraction tests"""
-    df = mock_interactions.copy()
-    df['userAgent'] = ['Mozilla/5.0 Chrome/120', 'Mozilla/5.0 Chrome/120',
-                       'HeadlessChrome/120', 'HeadlessChrome/120', 'HeadlessChrome/120']
-    df['metadata_base_price'] = [None, None, 150.0, 150.0, 200.0]
-    return df
--- a/tests/e2e/init.py
+++ b/tests/e2e/init.py
@@ -1 +0,0 @@
-"""E2E test suite for PHANTOM dynamic pricing pipeline."""
--- a/tests/e2e/fixtures.ts
+++ b/tests/e2e/fixtures.ts
@@ -1,17 +0,0 @@
-import { test as base } from '@playwright/test';
-
-type TestFixtures = {
-  backendUrl: string;
-  pricingUrl: string;
-};
-
-export const test = base.extend<TestFixtures>({
-  backendUrl: async ({}, use) => {
-    await use(process.env.BACKEND_URL || 'http://localhost:5000');
-  },
-  pricingUrl: async ({}, use) => {
-    await use(process.env.PRICING_PROVIDER_URL || 'http://localhost:5001');
-  },
-});
-
-export { expect } from '@playwright/test';
--- a/tests/e2e/helpers/api.ts
+++ b/tests/e2e/helpers/api.ts
@@ -1,69 +0,0 @@
-interface PriceResponse {
-  price: number;
-  base_price: number;
-  markup: number;
-  model_version?: string;
-}
-
-export async function fetchPrice(
-  baseUrl: string,
-  productId: string,
-  mode: string = 'simple_surge',
-  sessionId?: string
-): Promise<PriceResponse> {
-  const params = new URLSearchParams();
-  if (sessionId) params.set('sessionId', sessionId);
-
-  const url = `${baseUrl}/api/pricing?mode=${mode}&productId=${productId}&${params}`;
-  const resp = await fetch(url);
-
-  if (!resp.ok) {
-    throw new Error(`Price fetch failed: ${resp.status}`);
-  }
-
-  return resp.json();
-}
-
-export async function waitForPriceChange(
-  baseUrl: string,
-  productId: string,
-  baselinePrice: number,
-  mode: string,
-  sessionId?: string,
-  maxRetries: number = 10,
-  pollInterval: number = 500
-): Promise<PriceResponse> {
-  for (let i = 0; i < maxRetries; i++) {
-    const priceResp = await fetchPrice(baseUrl, productId, mode, sessionId);
-    if (Math.abs(priceResp.price - baselinePrice) > 0.01) {
-      return priceResp;
-    }
-    await new Promise(r => setTimeout(r, pollInterval));
-  }
-
-  throw new Error(`Price did not change after ${maxRetries} retries`);
-}
-
-export async function ingestEvent(
-  baseUrl: string,
-  sessionId: string,
-  event: string,
-  productId?: string,
-  metadata?: Record<string, any>
-): Promise<void> {
-  const resp = await fetch(`${baseUrl}/api/ingest`, {
-    method: 'POST',
-    headers: { 'Content-Type': 'application/json' },
-    body: JSON.stringify({
-      sessionId,
-      event,
-      productId,
-      timestamp: new Date().toISOString(),
-      metadata,
-    }),
-  });
-
-  if (!resp.ok) {
-    throw new Error(`Event ingest failed: ${resp.status}`);
-  }
-}
--- a/tests/e2e/helpers/interactions.ts
+++ b/tests/e2e/helpers/interactions.ts
@@ -1,219 +0,0 @@
-import { Page } from '@playwright/test';
-
-export async function getSessionId(page: Page): Promise<string | null> {
-  const cookies = await page.context().cookies();
-  const sessionCookie = cookies.find(c => c.name === 'phantom_session_id');
-  return sessionCookie?.value || null;
-}
-
-export async function verifySessionConsistency(page: Page, expectedSessionId: string): Promise<boolean> {
-  const currentSessionId = await getSessionId(page);
-  return currentSessionId === expectedSessionId;
-}
-
-export async function createFreshSession(page: Page, storeType: 'hotel' | 'airline' = 'hotel'): Promise<string> {
-  await page.context().clearCookies();
-  await page.goto('/');
-  await page.waitForLoadState('networkidle');
-  await page.waitForTimeout(500);
-
-  const sid = await getSessionId(page);
-  if (!sid) throw new Error('Session not created');
-  return sid;
-}
-
-interface SearchParams {
-  destination?: string;
-  checkIn?: string;
-  guests?: number;
-  rooms?: number;
-  origin?: string;
-  departure?: string;
-  adults?: number;
-}
-
-export async function performSearch(page: Page, params: SearchParams, storeType: 'hotel' | 'airline' = 'hotel' ): Promise<void> {
-  await page.waitForLoadState('networkidle');
-
-  if (storeType === 'hotel') {
-    const destInput = page.locator('input#destination');
-    await destInput.fill(params.destination || 'New York');
-
-    const checkInInput = page.locator('input#checkIn');
-    const checkInDate = params.checkIn || new Date(Date.now() + 7 * 86400000).toISOString().split('T')[0];
-    await checkInInput.fill(checkInDate);
-
-    const searchBtn = page.locator('button:has-text("Search Rooms")');
-    await searchBtn.click();
-  } else {
-    const originDropdown = page.locator('button:has-text("Select origin")').or(
-      page.locator('[id="origin"]').locator('button').first()
-    );
-    await originDropdown.click();
-    await page.waitForTimeout(200);
-    const originOption = page.locator(`button:has-text("${params.origin || 'JFK'}")`).first();
-    await originOption.click();
-    await page.waitForTimeout(200);
-
-    const destDropdown = page.locator('button:has-text("Select destination")').or(
-      page.locator('[id="destination"]').locator('button').first()
-    );
-    await destDropdown.click();
-    await page.waitForTimeout(200);
-    const destOption = page.locator(`button:has-text("${params.destination || 'LAX'}")`).first();
-    await destOption.click();
-    await page.waitForTimeout(200);
-
-    const departInput = page.locator('input#departDate');
-    const departDate = params.departure || new Date(Date.now() + 7 * 86400000).toISOString().split('T')[0];
-    await departInput.fill(departDate);
-
-    const searchBtn = page.locator('button:has-text("Search Flights")');
-    await searchBtn.click();
-  }
-
-  await page.waitForLoadState('networkidle');
-}
-
-export async function selectRandomProduct(page: Page, storeType: 'hotel' | 'airline' = 'hotel'): Promise<string> {
-  await page.waitForLoadState('networkidle');
-
-  const cardClass = storeType === 'hotel' ? '.hotel-card' : '.flight-card';
-  const productCards = page.locator(cardClass);
-
-  const count = await productCards.count();
-  if (count === 0) throw new Error('No products found on listing page');
-
-  const randomIdx = Math.floor(Math.random() * count);
-  return randomIdx.toString();
-}
-
-export async function openProductFromListing(page: Page, productId?: string): Promise<string> {
-  await page.waitForLoadState('networkidle');
-
-  const hotelCards = page.locator('.hotel-card');
-  const flightCards = page.locator('.flight-card');
-
-  const hotelCount = await hotelCards.count();
-  const flightCount = await flightCards.count();
-
-  let productCards;
-  if (hotelCount > 0) {
-    productCards = hotelCards;
-  } else if (flightCount > 0) {
-    productCards = flightCards;
-  } else {
-    throw new Error('No products found on listing page');
-  }
-
-  const count = await productCards.count();
-  const randomIdx = productId ? 0 : Math.floor(Math.random() * count);
-  await productCards.nth(randomIdx).click();
-
-  await page.waitForLoadState('networkidle');
-
-  const url = page.url();
-  const match = url.match(/\/products\/([^/?]+)/);
-  if (!match) throw new Error('Cannot parse product ID from URL after navigation');
-
-  return match[1];
-}
-
-export async function getPriceFromDOM(page: Page): Promise<number> {
-  await page.waitForLoadState('networkidle');
-
-  await page.waitForSelector('.price-amount', { timeout: 15000 }).catch(() => null);
-
-  const priceSelectors = [
-    '.price-amount',
-    '.price-display',
-    '[data-testid="price"]',
-    '[data-price]',
-  ];
-
-  for (const selector of priceSelectors) {
-    const priceEl = page.locator(selector).first();
-    if (await priceEl.count() > 0) {
-      const text = await priceEl.textContent();
-      if (!text) continue;
-
-      const match = text.match(/[\$]?\s*([\d,]+(?:\.\d{2})?)/);
-      if (match) {
-        const priceStr = match[1].replace(/,/g, '');
-        return parseFloat(priceStr);
-      }
-    }
-  }
-
-  const dataPrice = await page.locator('[data-price]').first().getAttribute('data-price').catch(() => null);
-  if (dataPrice) return parseFloat(dataPrice);
-
-  throw new Error('Cannot extract price from DOM');
-}
-
-export async function navigateToProduct(page: Page,productId: string,storeType: 'hotel' | 'airline' = 'hotel'): Promise<void> {
-  await page.goto(`/products/${productId}`);
-  await page.waitForLoadState('networkidle');
-}
-
-export async function viewProductViaFlow(page: Page, storeType: 'hotel' | 'airline' = 'hotel', searchParams?: SearchParams): Promise<string> {
-  const params = new URLSearchParams();
-  params.set('dateIndex', '7');
-
-  if (storeType === 'hotel') {
-    params.set('destination', searchParams?.destination || 'New York');
-    params.set('adults', '2');
-    params.set('rooms', '1');
-  } else {
-    params.set('origin', searchParams?.origin || 'JFK');
-    params.set('destination', searchParams?.destination || 'LAX');
-    params.set('adults', '1');
-    params.set('children', '0');
-    params.set('infants', '0');
-  }
-
-  await page.goto(`/products?${params.toString()}`);
-  await page.waitForLoadState('networkidle');
-
-  const productId = await openProductFromListing(page);
-  await page.waitForTimeout(500);
-  return productId;
-}
-
-export async function rapidViewProductViaFlow(page: Page, count: number, delayMs: number = 100, storeType: 'hotel' | 'airline' = 'hotel'): Promise<string[]> {
-  const productIds: string[] = [];
-
-  for (let i = 0; i < count; i++) {
-    const productId = await viewProductViaFlow(page, storeType);
-    productIds.push(productId);
-
-    await page.waitForTimeout(delayMs);
-  }
-
-  return productIds;
-}
-
-export async function humanLikeViewProduct(page: Page, storeType: 'hotel' | 'airline' = 'hotel'
-): Promise<string> {
-  const productId = await viewProductViaFlow(page, storeType);
-
-  await page.hover('h1');
-  await page.waitForTimeout(800 + Math.random() * 400);
-
-  await page.mouse.wheel(0, 200);
-  await page.waitForTimeout(500 + Math.random() * 300);
-
-  const paragraphs = await page.locator('p').all();
-  if (paragraphs.length > 0) {
-    await paragraphs[0].hover();
-    await page.waitForTimeout(600 + Math.random() * 400);
-  }
-
-  return productId;
-}
-
-export async function addToCart(page: Page): Promise<void> {
-  const addBtn = page.locator('button:has-text("Add to Cart")');
-  await addBtn.click();
-  await page.waitForTimeout(500);
-}
--- a/tests/e2e/helpers/kafka.ts
+++ b/tests/e2e/helpers/kafka.ts
@@ -1,39 +0,0 @@
-interface InteractionEvent {
-  sessionId: string;
-  event: string;
-  productId?: string;
-  timestamp: string;
-  metadata?: Record<string, any>;
-}
-
-const dumpKafkaTopic = async (backendUrl: string, topic: string) => {
-  const resp = await fetch(`${backendUrl}/api/kafka/dump?topic=${topic}`);
-  if (!resp.ok) throw new Error(`Kafka dump failed: ${resp.status}`);
-  const { messages = [] } = await resp.json();
-  return messages as any[];
-};
-
-export const waitForInteractionEvent = async (
-  backendUrl: string,
-  sessionId: string,
-  eventType: string,
-  maxRetries = 10,
-  pollInterval = 500
-): Promise<InteractionEvent | null> => {
-  for (let i = 0; i < maxRetries; i++) {
-    const msgs = await dumpKafkaTopic(backendUrl, "user-interactions");
-    const hit = msgs.find(m => m.sessionId === sessionId && m.event === eventType);
-    if (hit) return hit as InteractionEvent;
-    await new Promise<void>(r => setTimeout(r, pollInterval));
-  }
-  return null;
-};
-
-export const countProductViews = async (backendUrl: string, productId: string) =>
-  (await dumpKafkaTopic(backendUrl, "user-interactions")).reduce(
-    (n, m) => n + (m.productId === productId && m.event === "view_item_page" ? 1 : 0),
-    0
-  );
-
-export const getSessionEvents = async (backendUrl: string, sessionId: string) =>
-  (await dumpKafkaTopic(backendUrl, "user-interactions")).filter(m => m.sessionId === sessionId);
--- a/tests/e2e/package.json
+++ b/tests/e2e/package.json
@@ -1,19 +0,0 @@
-{
-  "name": "e2e",
-  "version": "1.0.0",
-  "main": "index.js",
-  "scripts": {
-    "test": "playwright test",
-    "test:ui": "playwright test --ui",
-    "test:debug": "playwright test --debug"
-  },
-  "keywords": [],
-  "author": "",
-  "license": "ISC",
-  "description": "",
-  "devDependencies": {
-    "@playwright/test": "^1.57.0",
-    "@types/node": "^25.0.6",
-    "typescript": "^5.9.3"
-  }
-}
--- a/tests/e2e/playwright.config.ts
+++ b/tests/e2e/playwright.config.ts
@@ -1,25 +0,0 @@
-import { defineConfig, devices } from '@playwright/test';
-
-export default defineConfig({
-  testDir: './scenarios',
-  fullyParallel: true,
-  forbidOnly: !!process.env.CI,
-  retries: 0,
-  workers: 5,
-  reporter: 'list',
-  use: {
-    baseURL: process.env.WEB_URL || 'http://localhost:3000',
-    trace: 'retain-on-failure',
-    screenshot: 'only-on-failure',
-  },
-  timeout: 60000,
-  expect: {
-    timeout: 10000,
-  },
-  projects: [
-    {
-      name: 'chromium',
-      use: { ...devices['Desktop Chrome'] },
-    },
-  ],
-});
--- a/tests/e2e/scenarios/session-aware.spec.ts
+++ b/tests/e2e/scenarios/session-aware.spec.ts
@@ -1,156 +0,0 @@
-import { test, expect } from '../fixtures';
-import {
-  createFreshSession,
-  viewProductViaFlow,
-  rapidViewProductViaFlow,
-  humanLikeViewProduct,
-  getPriceFromDOM,
-  verifySessionConsistency,
-  addToCart,
-} from '../helpers/interactions';
-import { getSessionEvents } from '../helpers/kafka';
-
-test.describe('SessionAwarePricer E2E', () => {
-  const STORE_TYPE = 'hotel';
-
-  test('baseline: human-like behavior maintains base price', async ({ page, backendUrl }) => {
-    const sessionId = await createFreshSession(page, STORE_TYPE);
-
-    const productId1 = await humanLikeViewProduct(page, STORE_TYPE);
-    const baselinePrice = await getPriceFromDOM(page);
-    expect(await verifySessionConsistency(page, sessionId)).toBeTruthy();
-
-    await page.waitForTimeout(1500);
-
-    const productId2 = await humanLikeViewProduct(page, STORE_TYPE);
-    const secondPrice = await getPriceFromDOM(page);
-    expect(await verifySessionConsistency(page, sessionId)).toBeTruthy();
-
-    expect(Math.abs(secondPrice - baselinePrice) / baselinePrice).toBeLessThan(0.1);
-  });
-
-  test('agent detection: rapid robot-like behavior increases price', async ({ page, backendUrl }) => {
-    const sessionId = await createFreshSession(page, STORE_TYPE);
-
-    const productId = await viewProductViaFlow(page, STORE_TYPE);
-    const baselinePrice = await getPriceFromDOM(page);
-
-    await page.waitForTimeout(500);
-
-    await rapidViewProductViaFlow(page, 8, 100, STORE_TYPE);
-    expect(await verifySessionConsistency(page, sessionId)).toBeTruthy();
-
-    await page.waitForTimeout(2500);
-
-    const events = await getSessionEvents(backendUrl, sessionId);
-    expect(events.length).toBeGreaterThanOrEqual(8);
-
-    await page.goto(`/products/${productId}`);
-    await page.waitForLoadState('networkidle');
-    const agentPrice = await getPriceFromDOM(page);
-
-    expect(agentPrice).toBeGreaterThan(baselinePrice);
-    expect((agentPrice - baselinePrice) / baselinePrice).toBeGreaterThan(0.01);
-  });
-
-  test('velocity threshold: high event rate triggers detection', async ({ page, backendUrl }) => {
-    const sessionId = await createFreshSession(page, STORE_TYPE);
-
-    const productId = await viewProductViaFlow(page, STORE_TYPE);
-    const baselinePrice = await getPriceFromDOM(page);
-
-    const startTime = Date.now();
-    await rapidViewProductViaFlow(page, 10, 80, STORE_TYPE);
-    const duration = (Date.now() - startTime) / 1000;
-
-    const eventsPerSec = 10 / duration;
-    expect(eventsPerSec).toBeGreaterThan(2.0);
-
-    await page.waitForTimeout(2000);
-
-    await page.goto(`/products/${productId}`);
-    await page.waitForLoadState('networkidle');
-    const agentPrice = await getPriceFromDOM(page);
-
-    expect(agentPrice).toBeGreaterThan(baselinePrice);
-    expect(await verifySessionConsistency(page, sessionId)).toBeTruthy();
-  });
-
-  test('cart ratio: high cart/view ratio signals intent', async ({ page, backendUrl }) => {
-    const sessionId = await createFreshSession(page, STORE_TYPE);
-
-    const productId = await viewProductViaFlow(page, STORE_TYPE);
-    const baselinePrice = await getPriceFromDOM(page);
-
-    await page.waitForTimeout(500);
-    await addToCart(page);
-
-    await page.waitForTimeout(2000);
-
-    await page.goto(`/products/${productId}`);
-    await page.waitForLoadState('networkidle');
-    const cartPrice = await getPriceFromDOM(page);
-
-    expect(cartPrice).toBeGreaterThanOrEqual(baselinePrice);
-    expect(await verifySessionConsistency(page, sessionId)).toBeTruthy();
-  });
-
-  test('mixed behavior: occasional fast actions tolerated', async ({ page, backendUrl }) => {
-    const sessionId = await createFreshSession(page, STORE_TYPE);
-
-    const productId1 = await humanLikeViewProduct(page, STORE_TYPE);
-    const baselinePrice = await getPriceFromDOM(page);
-
-    await page.waitForTimeout(1200);
-
-    await rapidViewProductViaFlow(page, 2, 150, STORE_TYPE);
-
-    await page.waitForTimeout(1500);
-    await humanLikeViewProduct(page, STORE_TYPE);
-    const finalPrice = await getPriceFromDOM(page);
-
-    expect(Math.abs(finalPrice - baselinePrice) / baselinePrice).toBeLessThan(0.3);
-    expect(await verifySessionConsistency(page, sessionId)).toBeTruthy();
-  });
-
-  test('session isolation: agent behavior in one session does not affect others', async ({
-    page,
-    context,
-    backendUrl,
-  }) => {
-    const sessionIdA = await createFreshSession(page, STORE_TYPE);
-    const productId = await viewProductViaFlow(page, STORE_TYPE);
-    const basePrice = await getPriceFromDOM(page);
-
-    await rapidViewProductViaFlow(page, 10, 100, STORE_TYPE);
-    await page.waitForTimeout(2000);
-
-    await page.goto(`/products/${productId}`);
-    await page.waitForLoadState('networkidle');
-    const agentPrice = await getPriceFromDOM(page);
-    expect(agentPrice).toBeGreaterThan(basePrice * 0.99);
-
-    const page2 = await context.newPage();
-    const sessionIdB = await createFreshSession(page2, STORE_TYPE);
-
-    await page2.goto(`/products/${productId}`);
-    await page2.waitForLoadState('networkidle');
-    const cleanPrice = await getPriceFromDOM(page2);
-
-    expect(Math.abs(cleanPrice - basePrice) / basePrice).toBeLessThan(0.1);
-    expect(sessionIdA).not.toBe(sessionIdB);
-  });
-
-  test('session persistence: session ID maintained across views', async ({ page }) => {
-    const sessionId = await createFreshSession(page, STORE_TYPE);
-
-    await viewProductViaFlow(page, STORE_TYPE);
-    expect(await verifySessionConsistency(page, sessionId)).toBeTruthy();
-
-    await viewProductViaFlow(page, STORE_TYPE);
-    expect(await verifySessionConsistency(page, sessionId)).toBeTruthy();
-
-    await viewProductViaFlow(page, STORE_TYPE);
-    expect(await verifySessionConsistency(page, sessionId)).toBeTruthy();
-  });
-});
--- a/tests/e2e/scenarios/surge-pricing.spec.ts
+++ b/tests/e2e/scenarios/surge-pricing.spec.ts
@@ -1,111 +0,0 @@
-import { test, expect } from '../fixtures';
-import {
-  createFreshSession,
-  viewProductViaFlow,
-  rapidViewProductViaFlow,
-  getPriceFromDOM,
-  verifySessionConsistency,
-} from '../helpers/interactions';
-import { waitForInteractionEvent, countProductViews } from '../helpers/kafka';
-
-test.describe('SimpleSurgePricer E2E', () => {
-  const STORE_TYPE = 'hotel';
-
-  test('baseline: initial price equals base price', async ({ page, backendUrl }) => {
-    const sessionId = await createFreshSession(page, STORE_TYPE);
-
-    const productId = await viewProductViaFlow(page, STORE_TYPE);
-    const price = await getPriceFromDOM(page);
-
-    expect(price).toBeGreaterThan(0);
-    expect(await verifySessionConsistency(page, sessionId)).toBeTruthy();
-  });
-
-  test('surge: rapid views trigger price increase', async ({ page, backendUrl }) => {
-    const sessionId = await createFreshSession(page, STORE_TYPE);
-
-    const productId = await viewProductViaFlow(page, STORE_TYPE);
-    const baselinePrice = await getPriceFromDOM(page);
-
-    await rapidViewProductViaFlow(page, 5, 200, STORE_TYPE);
-
-    await page.waitForTimeout(2000);
-
-    const evt = await waitForInteractionEvent(backendUrl, sessionId, 'view_item_page');
-    expect(evt).not.toBeNull();
-
-    const viewCount = await countProductViews(backendUrl, productId);
-    expect(viewCount).toBeGreaterThanOrEqual(5);
-
-    await page.goto(`/products/${productId}`);
-    await page.waitForLoadState('networkidle');
-    const surgedPrice = await getPriceFromDOM(page);
-
-    expect(surgedPrice).toBeGreaterThan(baselinePrice);
-    expect((surgedPrice - baselinePrice) / baselinePrice).toBeGreaterThan(0.01);
-    expect(await verifySessionConsistency(page, sessionId)).toBeTruthy();
-  });
-
-  test('threshold: price unchanged below threshold', async ({ page, backendUrl }) => {
-    const sessionId = await createFreshSession(page, STORE_TYPE);
-
-    const productId = await viewProductViaFlow(page, STORE_TYPE);
-    const baselinePrice = await getPriceFromDOM(page);
-
-    await rapidViewProductViaFlow(page, 2, 300, STORE_TYPE);
-
-    await page.waitForTimeout(1500);
-
-    await page.goto(`/products/${productId}`);
-    await page.waitForLoadState('networkidle');
-    const currentPrice = await getPriceFromDOM(page);
-
-    expect(Math.abs(currentPrice - baselinePrice) / baselinePrice).toBeLessThan(0.05);
-    expect(await verifySessionConsistency(page, sessionId)).toBeTruthy();
-  });
-
-  test('window: surge decays after window expires', async ({ page, backendUrl }) => {
-    const sessionId = await createFreshSession(page, STORE_TYPE);
-
-    const productId = await viewProductViaFlow(page, STORE_TYPE);
-    const baselinePrice = await getPriceFromDOM(page);
-
-    await rapidViewProductViaFlow(page, 5, 150, STORE_TYPE);
-
-    await page.waitForTimeout(1500);
-
-    await page.goto(`/products/${productId}`);
-    await page.waitForLoadState('networkidle');
-    const surgedPrice = await getPriceFromDOM(page);
-    expect(surgedPrice).toBeGreaterThan(baselinePrice);
-
-    await page.waitForTimeout(12000);
-
-    await page.goto(`/products/${productId}`);
-    await page.waitForLoadState('networkidle');
-    const decayedPrice = await getPriceFromDOM(page);
-    expect(decayedPrice).toBeLessThan(surgedPrice);
-    expect(await verifySessionConsistency(page, sessionId)).toBeTruthy();
-  });
-
-  test('isolation: different products have independent surge', async ({ page, backendUrl }) => {
-    const sessionId = await createFreshSession(page, STORE_TYPE);
-
-    const productIdA = await viewProductViaFlow(page, STORE_TYPE);
-    const basePriceA = await getPriceFromDOM(page);
-
-    await rapidViewProductViaFlow(page, 5, 200, STORE_TYPE);
-    await page.waitForTimeout(2000);
-
-    await page.goto(`/products/${productIdA}`);
-    await page.waitForLoadState('networkidle');
-    const surgedPriceA = await getPriceFromDOM(page);
-
-    const productIdB = await viewProductViaFlow(page, STORE_TYPE);
-    const priceB = await getPriceFromDOM(page);
-
-    expect(surgedPriceA).toBeGreaterThan(basePriceA * 0.99);
-    expect(productIdA).not.toBe(productIdB);
-    expect(await verifySessionConsistency(page, sessionId)).toBeTruthy();
-  });
-});
--- a/tests/e2e/tsconfig.json
+++ b/tests/e2e/tsconfig.json
@@ -1,15 +0,0 @@
-{
-  "compilerOptions": {
-    "target": "ES2022",
-    "module": "commonjs",
-    "lib": ["ES2022"],
-    "strict": true,
-    "esModuleInterop": true,
-    "skipLibCheck": true,
-    "forceConsistentCasingInFileNames": true,
-    "resolveJsonModule": true,
-    "types": ["node", "@playwright/test"]
-  },
-  "include": ["**/*.ts"],
-  "exclude": ["node_modules"]
-}
--- a/web/src/app/api/ingest/route.ts
+++ b/web/src/app/api/ingest/route.ts
@@ -7,7 +7,7 @@ export async function POST(req: NextRequest) {
  try {
    const body = await req.json();

-    const storeMode = process.env.NEXT_PUBLIC_STORE_MODE || process.env.STORE_MODE || 'hotel';
+    const storeMode = process.env.STORE_MODE || 'hotel';
    const userAgent = req.headers.get('user-agent') || undefined;

    const event: EventBase = {
--- a/web/src/app/api/pricing/route.ts
+++ b/web/src/app/api/pricing/route.ts
@@ -11,7 +11,7 @@ export async function GET(req: NextRequest) {
    const productId = searchParams.get('productId');
    const sessionId = searchParams.get('sessionId');
    const experimentId = searchParams.get('experimentId');
-    const storeMode = process.env.NEXT_PUBLIC_STORE_MODE || process.env.STORE_MODE || 'hotel';
+    const storeMode = process.env.NEXT_PUBLIC_STORE_MODE || 'shop';

    if (!productId) {
        return NextResponse.json(
--- a/web/src/components/feats/airline/AirlineHero.tsx
+++ b/web/src/components/feats/airline/AirlineHero.tsx
@@ -2,20 +2,10 @@

 import { useState, FormEvent } from 'react';
 import { useRouter } from 'next/navigation';
-import { Button, Label, DateInput, Dropdown, DropdownCounter, SelectDropdown, SelectOption } from '@/components/ui';
+import { Button, Label, Input, DateInput, RadioGroup, Dropdown, DropdownCounter } from '@/components/ui';
 import { dateToDaysFromToday } from '@/lib/airline-utils';

-const CITIES: SelectOption[] = [
-  { value: 'JFK', label: 'New York (JFK)', sublabel: 'John F. Kennedy International' },
-  { value: 'LAX', label: 'Los Angeles (LAX)', sublabel: 'Los Angeles International' },
-  { value: 'ORD', label: 'Chicago (ORD)', sublabel: "O'Hare International" },
-  { value: 'MIA', label: 'Miami (MIA)', sublabel: 'Miami International' },
-  { value: 'SFO', label: 'San Francisco (SFO)', sublabel: 'San Francisco International' },
-  { value: 'SEA', label: 'Seattle (SEA)', sublabel: 'Seattle-Tacoma International' },
-  { value: 'ATL', label: 'Atlanta (ATL)', sublabel: 'Hartsfield-Jackson International' },
-  { value: 'DFW', label: 'Dallas (DFW)', sublabel: 'Dallas/Fort Worth International' },
-];
-
+type TripType = 'roundtrip' | 'oneway' | 'multicity';

 const PlaneIcon = () => (
  <svg className="w-5 h-5" fill="none" stroke="currentColor" viewBox="0 0 24 24">
@@ -32,9 +22,11 @@ const LocationIcon = () => (

 export default function AirlineHero() {
  const router = useRouter();
+  const [tripType, setTripType] = useState<TripType>('roundtrip');
  const [origin, setOrigin] = useState('');
  const [destination, setDestination] = useState('');
  const [departDate, setDepartDate] = useState('');
+  const [returnDate, setReturnDate] = useState('');
  const [passengers, setPassengers] = useState({ adults: 1, children: 0, infants: 0 });

  const handleSearch = (e: FormEvent) => {
@@ -48,6 +40,8 @@ export default function AirlineHero() {

    if (origin) params.set('origin', origin);
    if (destination) params.set('destination', destination);
+    if (tripType !== 'roundtrip') params.set('tripType', tripType);
+    if (returnDate && tripType === 'roundtrip') params.set('returnDate', returnDate);

    params.set('adults', passengers.adults.toString());
    params.set('children', passengers.children.toString());
@@ -72,15 +66,28 @@ export default function AirlineHero() {

        <div className="search-form">
          <form onSubmit={handleSearch}>
-            <div className="grid grid-cols-1 sm:grid-cols-2 lg:grid-cols-3 gap-4">
+            <div className="mb-6">
+              <RadioGroup
+                name="tripType"
+                value={tripType}
+                onChange={setTripType}
+                options={[
+                  { value: 'roundtrip', label: 'Round-trip' },
+                  { value: 'oneway', label: 'One-way' },
+                  { value: 'multicity', label: 'Multi-city' },
+                ]}
+              />
+            </div>
+
+            <div className="grid grid-cols-1 sm:grid-cols-2 lg:grid-cols-4 gap-4">
              <div>
                <Label htmlFor="origin">From</Label>
-                <SelectDropdown
+                <Input
+                  type="text"
                  id="origin"
                  value={origin}
-                  onChange={setOrigin}
-                  options={CITIES}
-                  placeholder="Select origin"
+                  onChange={(e) => setOrigin(e.target.value)}
+                  placeholder="Airport or city"
                  icon={<PlaneIcon />}
                  required
                />
@@ -88,12 +95,12 @@ export default function AirlineHero() {

              <div>
                <Label htmlFor="destination">To</Label>
-                <SelectDropdown
+                <Input
+                  type="text"
                  id="destination"
                  value={destination}
-                  onChange={setDestination}
-                  options={CITIES}
-                  placeholder="Select destination"
+                  onChange={(e) => setDestination(e.target.value)}
+                  placeholder="Airport or city"
                  icon={<LocationIcon />}
                  required
                />
@@ -108,6 +115,20 @@ export default function AirlineHero() {
                  required
                />
              </div>
+
+              <div>
+                <Label htmlFor="returnDate">Return</Label>
+                {tripType === 'roundtrip' ? (
+                  <DateInput
+                    id="returnDate"
+                    value={returnDate}
+                    onChange={(e) => setReturnDate(e.target.value)}
+                    required
+                  />
+                ) : (
+                  <DateInput id="returnDate" disabled />
+                )}
+              </div>
            </div>

            <div className="grid grid-cols-4 sm:grid-cols-3 lg:grid-cols-4 gap-4 mt-4">
--- a/web/src/components/ui/SelectDropdown.tsx
+++ b/web/src/components/ui/SelectDropdown.tsx
@@ -1,119 +0,0 @@
-'use client';
-
-import { useState, useRef, useEffect, ReactNode } from 'react';
-
-export interface SelectOption {
-  value: string;
-  label: string;
-  sublabel?: string;
-}
-
-interface SelectDropdownProps {
-  value: string;
-  onChange: (value: string) => void;
-  options: SelectOption[];
-  placeholder?: string;
-  icon?: ReactNode;
-  required?: boolean;
-  id?: string;
-}
-
-export default function SelectDropdown({
-  value,
-  onChange,
-  options,
-  placeholder = 'Select...',
-  icon,
-  required,
-  id,
-}: SelectDropdownProps) {
-  const [open, setOpen] = useState(false);
-  const [filter, setFilter] = useState('');
-  const ref = useRef<HTMLDivElement>(null);
-  const inputRef = useRef<HTMLInputElement>(null);
-
-  useEffect(() => {
-    const handleClick = (e: MouseEvent) => {
-      if (ref.current && !ref.current.contains(e.target as Node)) {
-        setOpen(false);
-        setFilter('');
-      }
-    };
-    document.addEventListener('mousedown', handleClick);
-    return () => document.removeEventListener('mousedown', handleClick);
-  }, []);
-
-  const selectedOption = options.find((o) => o.value === value);
-  const filtered = options.filter(
-    (o) =>
-      o.label.toLowerCase().includes(filter.toLowerCase()) ||
-      o.value.toLowerCase().includes(filter.toLowerCase()) ||
-      o.sublabel?.toLowerCase().includes(filter.toLowerCase())
-  );
-
-  const handleSelect = (opt: SelectOption) => {
-    onChange(opt.value);
-    setOpen(false);
-    setFilter('');
-  };
-
-  return (
-    <div className="relative" ref={ref}>
-      <div
-        className="input-field flex items-center gap-2 cursor-pointer box-border"
-        onClick={() => {
-          setOpen(true);
-          setTimeout(() => inputRef.current?.focus(), 0);
-        }}
-      >
-        {icon && <span className="text-[var(--text-secondary)]">{icon}</span>}
-        {open ? (
-          <input
-            ref={inputRef}
-            type="text"
-            id={id}
-            value={filter}
-            onChange={(e) => setFilter(e.target.value)}
-            placeholder={placeholder}
-            className="flex-1 bg-transparent outline-none text-sm text-[var(--text-primary)]"
-          />
-        ) : (
-          <span className={`flex-1 text-sm ${value ? 'text-[var(--text-primary)]' : 'text-[var(--text-secondary)]'}`}>
-            {selectedOption ? selectedOption.label : placeholder}
-          </span>
-        )}
-        <svg
-          className={`w-4 h-4 text-[var(--text-secondary)] transition-transform ${open ? 'rotate-180' : ''}`}
-          fill="none"
-          stroke="currentColor"
-          viewBox="0 0 24 24"
-        >
-          <path strokeLinecap="round" strokeLinejoin="round" strokeWidth={2} d="M19 9l-7 7-7-7" />
-        </svg>
-      </div>
-      {open && (
-        <div className="absolute z-20 mt-1 w-full bg-[var(--bg-primary)] border-2 border-[var(--accent-primary)] rounded-md shadow-lg max-h-60 overflow-y-auto">
-          {filtered.length === 0 ? (
-            <div className="px-4 py-3 text-sm text-[var(--text-secondary)]">No results</div>
-          ) : (
-            filtered.map((opt) => (
-              <div
-                key={opt.value}
-                onClick={() => handleSelect(opt)}
-                className={`px-4 py-2 cursor-pointer transition-colors hover:bg-[var(--accent-primary-light)] ${
-                  opt.value === value ? 'bg-[var(--accent-primary-light)]' : ''
-                }`}
-              >
-                <div className="text-sm font-medium text-[var(--text-primary)]">{opt.label}</div>
-                {opt.sublabel && <div className="text-xs text-[var(--text-secondary)]">{opt.sublabel}</div>}
-              </div>
-            ))
-          )}
-        </div>
-      )}
-      {required && !value && (
-        <input type="text" required className="sr-only" tabIndex={-1} value="" onChange={() => {}} />
-      )}
-    </div>
-  );
-}
--- a/web/src/components/ui/index.ts
+++ b/web/src/components/ui/index.ts
@@ -5,5 +5,3 @@ export { default as DateInput } from './DateInput';
 export { default as RadioGroup } from './RadioGroup';
 export { default as Dropdown, DropdownCounter } from './Dropdown';
 export { default as Navigation } from './Navigation';
-export { default as SelectDropdown } from './SelectDropdown';
-export type { SelectOption } from './SelectDropdown';
--- a/web/src/lib/config.ts
+++ b/web/src/lib/config.ts
@@ -16,7 +16,7 @@ const envSchema = z.object({
 // parse and validate env at module load, fail fast with descriptive errors
 const parseEnv = (): Env => {
    const result = envSchema.safeParse({
-        STORE_MODE: process.env.NEXT_PUBLIC_STORE_MODE || process.env.STORE_MODE,
+        STORE_MODE: process.env.STORE_MODE,
        NEXT_PUBLIC_API_BASE: process.env.NEXT_PUBLIC_API_BASE,
        NEXT_PUBLIC_APP_ENV: process.env.NEXT_PUBLIC_APP_ENV,
    });
--- a/web/src/styles/airline.css
+++ b/web/src/styles/airline.css
@@ -278,8 +278,6 @@
  padding: 12px;
  transition: border-color 0.2s ease;
  width: 100%;
-  min-height: 48px;
-  box-sizing: border-box;
 }

 [data-mode="airline"] .input-field:focus {
Author	SHA1	Message	Date
Daniel Rosel	b2288d7f4f	fix: CVE vulnerability patching	2025-12-06 17:43:33 +01:00
Daniel Rosel	e6a5b95875	feature: e2e intro pipline surge pricing	2025-12-06 16:30:28 +01:00
Daniel Rosel	503c5e182d	chore: store sting	2025-12-05 18:04:23 +01:00
Daniel Rosel	2adfee5791	chore: cleaning elasticity references	2025-12-05 18:03:36 +01:00
Daniel Rosel	9f0d8b4532	test: we wont be using elasticity anymore so its okay	2025-12-05 17:59:37 +01:00
Daniel Rosel	6bcef33fdf	fix: fixing stale pacakge	2025-12-05 13:03:37 +01:00
Daniel Rosel	42fc78e402	adding a checkout page to both sites	2025-12-05 12:45:02 +01:00
Daniel Rosel	47cd52a0ac	created new pricing pipeline	2025-12-05 12:44:48 +01:00
Daniel Rosel	a38fac9d2b	chore: simple surge pricer	2025-12-05 12:44:21 +01:00
Daniel Rosel	951b08d65e	feature: cleaning up pipeline	2025-12-05 12:43:36 +01:00
Daniel Rosel	a351af1dbe	adding loader with historical data loading	2025-12-05 11:54:58 +01:00
Daniel Rosel	9041af2979	chore: clean up product display in hotel and cleaner interfacing	2025-12-05 11:45:16 +01:00
Daniel Rosel	93fb465cbb	fixing hotel information with image placeholders	2025-12-05 11:23:03 +01:00
Daniel Rosel	2a702a6907	fix: fixes of backwords	2025-12-04 18:39:43 +01:00
				`@@ -1 +0,0 @@`
				`"""E2E test suite for PHANTOM dynamic pricing pipeline."""`