release: v1.1.3 — 13 fixes across build, frontend, pipeline, and data handling

Build: fix incremental package (update_info.json SHA256, __delete__, test exclude, files/ prefix), fix bump_version for 3-segment, PYTHONUTF8 in update.bat
Frontend: upload→manual redirect, LLM→cluster_overview redirect, filter UI before clustering, 地球→态势 rename, fix globe drag direction+inertia
Pipeline: LLM→auto clustering+PCA, column union handling (keep extra cols, no dropping), FFT spectral analysis tool
Quality: fix 24 bare except:pass patterns, enrich LLM context with TlsDB metadata, auto-save workflows with pinning, TlsDB-based column types

analysis/data_loader.py

@@ -117,12 +117,17 @@ def _merge_schema_entries(
     return merged, conflicts
 
 
-def validate_schemas(schemas: list[dict]) -> None:
-    """Compare schemas across files and raise :class:`ValueError` on mismatch.
+def validate_schemas(schemas: list[dict], strict: bool = True) -> None:
+    """Compare schemas across files and raise or warn on column mismatch.
 
     Checks that every file has the same set of column names.  Type
     differences across files are **not** considered an error (Polars can
     cast), but differences in *column sets* are.
+
+    Args:
+        schemas: List of ``{column_name: dtype_string}`` dicts, one per file.
+        strict: If ``True`` (default), raise :class:`ValueError` on any
+            column mismatch.  If ``False``, log a warning instead.
     """
     if not schemas:
         return
@@ -139,7 +144,11 @@ def validate_schemas(schemas: list[dict]) -> None:
                 msg_parts.append(f"File #{i} missing columns: {sorted(only_in_ref)}")
             if only_in_current:
                 msg_parts.append(f"File #{i} has extra columns: {sorted(only_in_current)}")
-            raise ValueError(f"Schema mismatch: {'; '.join(msg_parts)}")
+            full_msg = f"Schema mismatch: {'; '.join(msg_parts)}"
+            if strict:
+                raise ValueError(full_msg)
+            else:
+                logger.warning(full_msg)
 
 
 # ── dtype inference from schema ─────────────────────────────────────────

analysis/tool_registry.py

@@ -696,6 +696,33 @@ def get_tools_meta() -> list[Tool]:
                 "required": ["dataset_id"],
             },
         ),
+        Tool(
+            name="analyze_fft",
+            description=(
+                "FFT频谱分析——从时间戳列提取周期模式和频率特征。适用于检测时间周期性"
+                "（如每日/每周流量模式）。返回频率、幅度、top周期、频谱质心、周期性评分。"
+                "只读，无副作用。"
+            ),
+            inputSchema={
+                "type": "object",
+                "properties": {
+                    "dataset_id": {
+                        "type": "string",
+                        "description": "Dataset ID from load_data",
+                    },
+                    "timestamp_column": {
+                        "type": "string",
+                        "description": "Column name containing timestamps or numeric time values",
+                    },
+                    "top_n": {
+                        "type": "integer",
+                        "description": "Number of top frequency peaks to return",
+                        "default": 5,
+                    },
+                },
+                "required": ["dataset_id", "timestamp_column"],
+            },
+        ),
         Tool(
             name="analyze_tls_health",
             description=(
@@ -820,9 +847,10 @@ async def handle_call(name: str, arguments: dict) -> dict:
         'compare_datasets': _handle_compare_datasets,
         'export_debug_sample': _handle_export_debug_sample,
         'repair_schema': _handle_repair_schema,
-        'analyze_patterns': _handle_analyze_patterns,
-        'analyze_temporal': _handle_analyze_temporal,
-        'analyze_tls_health': _handle_analyze_tls_health,
+    'analyze_patterns': _handle_analyze_patterns,
+                'analyze_temporal': _handle_analyze_temporal,
+                'analyze_fft': _handle_analyze_fft,
+                'analyze_tls_health': _handle_analyze_tls_health,
         'analyze_geo_distribution': _handle_analyze_geo_distribution,
         'analyze_entity_detail': _handle_analyze_entity_detail,
         'compute_distance_matrix': _handle_compute_distance_matrix,
@@ -2492,8 +2520,7 @@ async def _handle_validate_data(dataset_id: str, fix_issues: bool = False) -> di
                 if unique <= 1 and null_count == 0:
                     issues.append(f'{col}: constant value (no variation)')
             except Exception:
-                pass
-        total_rows = lf.select(pl.len()).collect(streaming=True).item()
+                pass  # n_unique not supported for this column type
     except Exception as exc:
         return {'error': f'Validation failed: {exc}', 'truncated': False}
     return {
@@ -2581,7 +2608,7 @@ async def _handle_find_outliers(dataset_id: str, columns: list[str] = None,
             if n_out > 0:
                 outlier_counts[col] = int(n_out)
         except Exception:
-            pass
+            pass  # quantile not supported for this column type
     return {'outlier_counts': outlier_counts, 'total_sampled': len(df), 'threshold_iqr': threshold}
 
 
@@ -2630,7 +2657,7 @@ async def _handle_diagnose_clustering(dataset_id: str, cluster_result_id: str) -
         if var_explained[0] > 0.95:
             diagnosis.append(f'First PCA component explains {var_explained[0]:.0%} variance — data nearly 1D')
     except Exception:
-        pass
+        pass  # PCA fit failed — likely constant or all-null columns
 
     return {
         'diagnosis': diagnosis,
@@ -2764,7 +2791,7 @@ async def _handle_analyze_patterns(dataset_id: str, top_n: int = 10) -> dict:
                     'unique': int(df[col_name].n_unique()),
                 }
             except Exception:
-                pass
+                pass  # value_counts not supported for column, skip
 
     # Port distribution
     for port_col in [':prs', ':prd']:
@@ -2777,7 +2804,7 @@ async def _handle_analyze_patterns(dataset_id: str, top_n: int = 10) -> dict:
                     'unique': int(df[port_col].n_unique()),
                 }
             except Exception:
-                pass
+                pass  # port distribution extraction failed — skip
 
     # TLS version breakdown
     for tls_col in ['0ver']:
@@ -2787,7 +2814,7 @@ async def _handle_analyze_patterns(dataset_id: str, top_n: int = 10) -> dict:
                 result['tls_versions'] = [{'version': str(r[tls_col]), 'count': int(r['count'])}
                                           for r in vc.iter_rows(named=True)]
             except Exception:
-                pass
+                pass  # TLS version value_counts not supported — skip
 
     return result
 
@@ -2816,6 +2843,154 @@ async def _handle_analyze_temporal(dataset_id: str, timestamp_column: str = None
             'note': 'Temporal analysis requires parsed timestamps; raw values shown in explore_distributions'}
 
 
+# ── 24b. analyze_fft ─────────────────────────────────────────────────
+
+async def _handle_analyze_fft(
+    dataset_id: str,
+    timestamp_column: str,
+    top_n: int = 5,
+) -> dict:
+    """FFT spectrum analysis on a timestamp column to detect periodic patterns.
+
+    Computes the Fast Fourier Transform of sorted timestamp-derived values,
+    returning dominant frequencies, spectral centroid, and a periodicity score.
+    """
+    store = SessionStore()
+    entry = store.get_dataset(dataset_id)
+    if entry is None:
+        return {'error': f'Dataset not found: {dataset_id}', 'truncated': False}
+
+    lf: pl.LazyFrame = entry['lazyframe']
+    schema = entry.get('schema', {})
+
+    # ── Resolve timestamp column ──────────────────────────────────────
+    col = None
+    # Try exact match first
+    if timestamp_column in schema:
+        col = timestamp_column
+    else:
+        # Try case-insensitive match
+        col_lower = timestamp_column.lower()
+        for c in schema:
+            if c.lower() == col_lower:
+                col = c
+                break
+    if col is None:
+        return {'error': f"Timestamp column '{timestamp_column}' not found in dataset. Available: {list(schema.keys())[:20]}",
+                'truncated': False}
+
+    # ── Extract and sort values ───────────────────────────────────────
+    try:
+        df = lf.select(pl.col(col).alias('__ts__')).collect(streaming=True)
+    except Exception as e:
+        return {'error': f'Failed to collect timestamp column: {e}', 'truncated': False}
+
+    n_raw = len(df)
+    if n_raw < 4:
+        return {'error': f'Need at least 4 samples for FFT; got {n_raw}', 'truncated': False}
+
+    # Determine dtype and convert to numeric values
+    ts_series = df['__ts__']
+    dtype = ts_series.dtype
+    str_dtypes = (pl.Utf8, pl.String)
+
+    if dtype in str_dtypes:
+        # Try parsing as datetime string, then convert to epoch seconds
+        try:
+            ts_parsed = ts_series.str.strptime(pl.Datetime, format=None, strict=False)
+            ts_numeric = ts_parsed.cast(pl.Int64) / 1_000_000_000.0  # ns → seconds
+        except Exception:
+            # Fallback: try to cast to float directly
+            try:
+                ts_numeric = ts_series.cast(pl.Float64)
+            except Exception as e:
+                return {'error': f'Cannot convert string column to numeric or datetime: {e}',
+                        'truncated': False}
+    elif dtype in (pl.Datetime, pl.Date):
+        ts_numeric = ts_series.cast(pl.Int64) / 1_000_000_000.0
+    else:
+        # Float, Int types — cast to float
+        try:
+            ts_numeric = ts_series.cast(pl.Float64)
+        except Exception as e:
+            return {'error': f'Cannot cast column to float: {e}', 'truncated': False}
+
+    # Build DataFrame with values, drop nulls, sort
+    df_sorted = pl.DataFrame({'__val__': ts_numeric}).drop_nulls().sort('__val__')
+    n = len(df_sorted)
+    if n < 4:
+        return {'error': f'After removing nulls, only {n} samples remain (need ≥4)', 'truncated': False}
+
+    values = df_sorted['__val__'].to_numpy(dtype=np.float64)
+
+    # ── FFT computation ────────────────────────────────────────────────
+    # Center by subtracting mean (DC offset removal)
+    values_centered = values - np.mean(values)
+    fft_complex = np.fft.fft(values_centered)
+    fft_magnitudes = np.abs(fft_complex)
+    freqs = np.fft.fftfreq(n)
+
+    # Take positive half (excluding DC at index 0)
+    pos_mask = freqs > 0
+    freqs_pos = freqs[pos_mask]
+    mags_pos = fft_magnitudes[pos_mask]
+
+    if len(freqs_pos) == 0:
+        return {'error': 'Insufficient positive frequencies for FFT analysis', 'truncated': False}
+
+    # ── Find top-N peaks ──────────────────────────────────────────────
+    peak_indices = np.argsort(mags_pos)[::-1][:min(top_n, len(mags_pos))]
+    top_periods = []
+    for idx in peak_indices:
+        freq = float(freqs_pos[idx])
+        mag = float(mags_pos[idx])
+        # Convert frequency to period (in sample-index units)
+        period = float(1.0 / freq) if freq > 1e-10 else float('inf')
+        # Map to time-domain period if data is temporal
+        if n > 1:
+            time_span = float(values[-1] - values[0])
+            sample_period = time_span / max(n - 1, 1)
+            time_period = period * sample_period
+        else:
+            time_period = float('inf')
+        top_periods.append({
+            'period': round(period, 2),
+            'magnitude': round(mag, 4),
+            'frequency': round(freq, 6),
+            'time_domain_period_seconds': round(time_period, 2) if time_period != float('inf') else None,
+        })
+
+    # ── Spectral centroid (weighted mean frequency) ────────────────────
+    mag_sum = float(np.sum(mags_pos))
+    if mag_sum > 0:
+        spectral_centroid = float(np.sum(freqs_pos * mags_pos) / mag_sum)
+    else:
+        spectral_centroid = 0.0
+
+    # ── Periodicity score: ratio of top peak to mean magnitude ────────
+    mean_mag = float(np.mean(mags_pos)) if len(mags_pos) > 0 else 1.0
+    max_mag = float(mags_pos[peak_indices[0]]) if len(peak_indices) > 0 else 0.0
+    periodicity_score = round(max_mag / max(mean_mag, 1e-10), 4)
+
+    # ── Return decimated data for LLM context window ──────────────────
+    # Decimate to at most 200 frequency-magnitude pairs
+    decimate_step = max(1, len(freqs_pos) // 200)
+    freqs_decimated = freqs_pos[::decimate_step].tolist()
+    mags_decimated = mags_pos[::decimate_step].tolist()
+
+    return {
+        'frequencies': [round(float(f), 6) for f in freqs_decimated],
+        'magnitudes': [round(float(m), 4) for m in mags_decimated],
+        'top_periods': top_periods,
+        'spectral_centroid': round(spectral_centroid, 6),
+        'periodicity_score': periodicity_score,
+        'n_samples': n,
+        'n_raw': n_raw,
+        'nan_dropped': n_raw - n,
+        'timestamp_column': col,
+    }
+
+
 # ── 25. analyze_tls_health ────────────────────────────────────────────
 
 async def _handle_analyze_tls_health(dataset_id: str) -> dict:
@@ -2861,7 +3036,7 @@ async def _handle_analyze_tls_health(dataset_id: str) -> dict:
             try:
                 weak_count += df['cipher-suite'].str.to_lowercase().str.contains(w).sum()
             except Exception:
-                pass
+                pass  # str.contains not supported for cipher-suite column — skip
         result['weak_cipher_pct'] = round(float(weak_count) / max(total_ciphers, 1) * 100, 1) if total_ciphers else 0
 
     if result.get('tls_legacy_pct', 0) > 20:
@@ -2894,7 +3069,7 @@ async def _handle_analyze_geo_distribution(dataset_id: str) -> dict:
                                  for r in vc.iter_rows(named=True)]
                 result[f'{label}_unique'] = int(df[geo_col].n_unique())
             except Exception:
-                pass
+                pass  # country distribution not computable — skip
 
     if 'scnt' in df.columns and 'dcnt' in df.columns:
         try:
@@ -2902,7 +3077,7 @@ async def _handle_analyze_geo_distribution(dataset_id: str) -> dict:
             result['domestic_pct'] = round(float(same) / max(len(df), 1) * 100, 1)
             result['international_pct'] = round(100.0 - result['domestic_pct'], 1)
         except Exception:
-            pass
+                pass  # domestic/international comparison failed — skip
 
     return result
 
@@ -2955,7 +3130,7 @@ async def _handle_analyze_entity_detail(dataset_id: str, entity_value: str) -> d
                     'max': float(df[c].max()) if df[c].max() is not None else None,
                 }
             except Exception:
-                pass
+                pass  # numeric stats not computable for this column — skip
 
     return result
 

analysis/type_classifier.py

@@ -20,6 +20,8 @@ from enum import Enum
 
 import polars as pl
 
+from analysis.tls_ref import get_tls_ref_dict
+
 # ---------------------------------------------------------------------------
 #  TLS version hex-to-label mapping
 #  (Duplicated in views.py for globe visualization — keep both in sync.)
@@ -118,6 +120,30 @@ _CONFIG_TYPE_MAP: dict[str, DataType] = {
     'string': DataType.STRING,
 }
 
+# Mapping from tls_ref_data ``data_type`` column values to DataType
+TLSDB_TYPE_MAP: dict[str, DataType] = {
+    'IPv4': DataType.IPv4,
+    '浮点': DataType.FLOAT,
+    '字符串': DataType.STRING,
+    '整数': DataType.INT,
+    '2小写字母缩写': DataType.ENUM,
+    'XX:YY.Z': DataType.STRING,
+    '2字节': DataType.HEX,
+    '2大写字母': DataType.STRING,
+    '枚举': DataType.ENUM,
+    '布尔': DataType.BOOL_ENUM,
+    '字符串或整数': DataType.STRING,
+}
+
+
+def _tlsdb_type_to_datatype(tlsdb_type_str: str) -> DataType | None:
+    """Convert a TlsDB ``data_type`` string to a :class:`DataType`.
+
+    Returns ``None`` if the type string is unknown, allowing the
+    caller to fall through to value-based detection.
+    """
+    return TLSDB_TYPE_MAP.get(tlsdb_type_str)
+
 
 # ---------------------------------------------------------------------------
 #  Column classification (single column)
@@ -243,6 +269,11 @@ def _values_to_type(series: pl.Series, col_name: str | None = None) -> DataType:
     # ── 8. ENUM detection (few unique values) ──────────────────────────
     unique_count = non_null.n_unique()
     if unique_count < 20:
+        # Safety: if >80 % of the sample values are unique strings,
+        # the column is high-cardinality — classify as STRING, not ENUM.
+        sample_unique_count = len(set(sample_strs))
+        if sample_unique_count > 0.8 * len(sample_strs):
+            return DataType.STRING
         return DataType.ENUM
 
     # ── 9. TLS version hex detection (before STRING fallback) ──────────
@@ -296,16 +327,20 @@ def classify_column(
     1. **config_type** — user override from ``config.yaml``.
        If provided (e.g. ``'ipv4'``), returns the corresponding
        :class:`DataType` immediately.
-    2. **Value-based** pattern matching — samples up to 100 non-null
+    2. **tls_ref_data** — authoritative type from ``TlsDB.csv`` /
+       ``tls_ref_data`` table.  Looked up by column name; if found,
+       the declared type is used directly (no value-based guessing).
+    3. **Value-based** pattern matching — samples up to 100 non-null
        values and tests against IPv4 → HEX → URL → PORT → BOOL_ENUM
        → LAT_LON → ENUM → FLOAT in order.
-    3. **Name-based** heuristics — only consulted when value-based
+    4. **Name-based** heuristics — only consulted when value-based
        detection returned ``STRING`` (i.e. no value pattern matched).
 
     Parameters
     ----------
     name:
-        Column name (used for name-based heuristics).
+        Column name (used for name-based heuristics and tls_ref_data
+        lookup).
     series:
         Polars :class:`Series` containing the column's data.
     config_type:
@@ -322,10 +357,17 @@ def classify_column(
         if mapped is not None:
             return mapped
 
-    # 2. Value-based sampling
+    # 2. tls_ref_data lookup (authoritative type from TlsDB)
+    tls_ref = get_tls_ref_dict()
+    if name in tls_ref:
+        dtype = _tlsdb_type_to_datatype(tls_ref[name]['data_type'])
+        if dtype is not None:
+            return dtype
+
+    # 3. Value-based sampling
     result = _values_to_type(series, name)
 
-    # 3. Name-based heuristics (only when values returned STRING)
+    # 4. Name-based heuristics (only when values returned STRING)
     if result == DataType.STRING:
         name_match = _name_to_type(name)
         if name_match is not None:

analysis/urls.py

@@ -7,7 +7,7 @@ urlpatterns = [
     path('upload/', views.upload_page, name='upload'),
     path('upload/csv/', views.upload_csv, name='upload_csv'),
     path('analyze/manual/', views.manual_page, name='manual'),
-    path('analyze/run/', views.run_analysis, name='run_analysis'),
+    path('analyze/run/', views.start_analysis, name='run_analysis'),
     path('analyze/auto/', views.auto_page, name='auto'),
     path('analyze/llm/', views.run_llm_analysis_view, name='run_llm_analysis'),
     path('runs/', views.run_list, name='run_list'),
@@ -24,4 +24,5 @@ urlpatterns = [
     path('globe/', views.globe_view, name='globe'),
     path('tools/run/', views.tool_lab_run, name='tool_lab_run'),
     path('tools/plan/', views.tool_plan, name='tool_plan'),
+    path('tools/filter/', views.apply_filter, name='apply_filter'),
 ]

analysis/views.py

@@ -431,7 +431,7 @@ def _background_process(run_id, upload_dir):
         run.progress_msg = f'正在流式加载 {total_files} 个文件（共 {total_batches} 批）...'
         run.save(update_fields=['status', 'progress_pct', 'progress_msg'])
 
-        from analysis.data_loader import save_to_db, drop_sqlite_table, load_from_db
+        from analysis.data_loader import save_to_db, drop_sqlite_table, load_from_db, _dtype_to_sqlite
 
         table_name = f'_data_{run_id}'
 
@@ -442,7 +442,8 @@ def _background_process(run_id, upload_dir):
                                schema_overrides=lat_lon_overrides)
         first_schema = dict(first_lf.collect_schema())
         schema = {name: str(dtype) for name, dtype in first_schema.items()}
-        reference_columns = list(schema.keys())  # canonical column order & set
+        # Dynamic column set: starts with first batch, unions new columns from later batches
+        reference_columns = list(schema.keys())
 
         # ── Phase 2: Batched collection → SQLite write ──
         table_created = False
@@ -454,13 +455,16 @@ def _background_process(run_id, upload_dir):
                                    schema_overrides=lat_lon_overrides)
             df_batch = batch_lf.collect(streaming=True)
 
-            # Schema alignment: ensure columns match reference order
+            # Schema alignment: union of all columns across batches
+            # 1) Fill NULL for columns in reference_columns but missing in this batch
             for col in reference_columns:
                 if col not in df_batch.columns:
                     df_batch = df_batch.with_columns(pl.lit(None).alias(col))
-            extra = [c for c in df_batch.columns if c not in reference_columns]
-            if extra:
-                df_batch = df_batch.drop(extra)
+            # 2) Append new columns found in this batch to reference_columns
+            new_cols = [c for c in df_batch.columns if c not in reference_columns]
+            if new_cols:
+                reference_columns = list(dict.fromkeys(reference_columns + new_cols))
+            # 3) Select all reference_columns in canonical order (NO dropping)
             df_batch = df_batch.select(reference_columns)
 
             # Dtype alignment: cast each column to the reference dtype
@@ -485,6 +489,18 @@ def _background_process(run_id, upload_dir):
                 df_batch = df_batch.with_columns(list(casts.values()))
 
             mode = 'replace' if not table_created else 'append'
+            # If new columns appeared in a later batch, ALTER TABLE to add them
+            if table_created and new_cols:
+                import sqlite3 as _sqlite3
+                from django.conf import settings as _dj_settings
+                _conn = _sqlite3.connect(_dj_settings.DATABASES['default']['NAME'])
+                try:
+                    for _col in new_cols:
+                        _sql_dtype = _dtype_to_sqlite(df_batch[_col].dtype)
+                        _conn.execute(f'ALTER TABLE "{table_name}" ADD COLUMN "{_col}" {_sql_dtype}')
+                    _conn.commit()
+                finally:
+                    _conn.close()
             save_to_db(run.id, df_batch.lazy(), mode=mode)
             table_created = True
 
@@ -539,7 +555,7 @@ def _background_process(run_id, upload_dir):
             from analysis.data_loader import drop_sqlite_table  # fmt: skip
             drop_sqlite_table(f'_data_{run_id}')
         except Exception:
-            pass
+            pass  # Expected: drop_sqlite_table may fail if table already gone
         run.status = 'failed'
         run.error_message = tb
         run.run_log += f'\n[ERROR] {tb}'
@@ -813,14 +829,8 @@ def _run_clustering_pipeline(run, store, entity_ds_id, feature_columns, algorith
             run.progress_msg = f'失败: {tb}'
             run.run_log += f'\n[ERROR] {tb}'
             run.save(update_fields=['status', 'error_message', 'progress_msg', 'run_log'])
-        except Exception:
-            pass
-
-
-# ── Manual analysis page ───────────────────────────────────────────────
-
-@csrf_exempt
-def run_analysis(request):
+        except Exception as e:
+            logger.warning('save failed after pipeline error: %s', e)
     """AJAX POST: run clustering pipeline on the upload dataset."""
     if request.method != 'POST':
         return JsonResponse({'error': 'POST required'}, status=405)
@@ -896,6 +906,73 @@ def run_analysis(request):
 
 
 
+
+
+
+# ── Filter endpoint for manual analysis page ──────────────────────────
+
+@csrf_exempt
+def apply_filter(request):
+    """POST {dataset_id, filters, logic} → apply filter_data tool → return filtered_X id."""
+    if request.method != 'POST':
+        return JsonResponse({'error': 'POST required'}, status=405)
+    import json as _json
+    import asyncio
+    body = _json.loads(request.body)
+    dataset_id = body.get('dataset_id')
+    filters = body.get('filters', [])
+    logic = body.get('logic', 'and')
+
+    if not dataset_id:
+        return JsonResponse({'error': 'dataset_id is required'})
+    if not filters:
+        return JsonResponse({'error': 'filters is required'})
+
+    try:
+        from analysis.tool_registry import _handle_filter_data
+        result = asyncio.run(_handle_filter_data(
+            dataset_id=dataset_id,
+            filters=filters,
+            logic=logic,
+        ))
+        if 'error' in result:
+            return JsonResponse({'error': result['error']})
+
+        filtered_id = result.get('filtered_dataset_id', '')
+        row_count = result.get('row_count', 0)
+
+        # Re-store with filtered_X ID pattern
+        if dataset_id.startswith('upload_'):
+            display_suffix = dataset_id.split('_', 1)[1]
+            new_id = f'filtered_{display_suffix}'
+        else:
+            new_id = filtered_id
+
+        from analysis.session_store import SessionStore
+        store = SessionStore()
+        entry = store.get_dataset(filtered_id)
+        if entry and new_id != filtered_id:
+            store.store_dataset(
+                dataset_id=new_id,
+                lazyframe=entry['lazyframe'],
+                schema=entry.get('schema', {}),
+                metadata=dict(entry.get('metadata', {}), parent_dataset_id=dataset_id),
+                parent_id=dataset_id,
+            )
+            new_id_to_use = new_id
+        else:
+            new_id_to_use = filtered_id
+
+        return JsonResponse({
+            'dataset_id': new_id_to_use,
+            'row_count': row_count,
+        })
+    except Exception as e:
+        import traceback
+        logger.error(f'[apply_filter] {traceback.format_exc()}')
+        return JsonResponse({'error': str(e)})
+
+
 # ── LLM auto analysis page ─────────────────────────────────────────────
 
 
@@ -978,8 +1055,8 @@ def run_llm_analysis_view(request):
                     save_fields = ['progress_pct', 'progress_msg', 'run_log', 'tool_calls_json']
 
                 run.save(update_fields=save_fields)
-            except Exception:
-                pass
+            except Exception as e:
+                logger.warning('progress_callback error: %s', e)
 
         try:
             from config import get_config
@@ -997,12 +1074,94 @@ def run_llm_analysis_view(request):
                 run.status = 'failed'
                 run.progress_msg = f'失败: {result["error"]}'
                 run.run_log += f'\n[ERROR] {result["error"]}'
+                run.save(update_fields=['status', 'progress_pct', 'progress_msg', 'run_log'])
             else:
-                run.status = 'completed'
-                run.progress_pct = 100
-                run.progress_msg = 'LLM 分析完成'
-                run.run_log += f'\n[完成] {result.get("result", "")}'
-            run.save(update_fields=['status', 'progress_pct', 'progress_msg', 'run_log'])
+                run.progress_pct = 55
+                run.progress_msg = f'LLM分析完成: {result.get("result", "")[:200]}'
+                run.run_log += f'\n[完成] {result.get("result", "")}\n[INFO] 开始自动聚类流程...'
+                run.save(update_fields=['progress_pct', 'progress_msg', 'run_log'])
+
+                # ── Auto-run clustering pipeline after LLM analysis ──
+                entity_ds_id = None
+                tool_calls = run.tool_calls_json or []
+                for tc in tool_calls:
+                    tool_name = tc.get('name', '')
+                    tool_input = tc.get('input', {})
+                    if tool_name in ('build_entity_profiles', 'run_clustering'):
+                        entity_ds_id = tool_input.get('dataset_id', '')
+                        if entity_ds_id and store.get_dataset(entity_ds_id):
+                            break
+                        entity_ds_id = None
+
+                # Fallback: try common dataset ID patterns
+                if not entity_ds_id:
+                    candidates = [
+                        ctx.get('dataset_id', ''),
+                        f'entity_{pk}',
+                        f'entity_{run.display_id}',
+                        f'upload_{run.display_id}',
+                    ]
+                    for cid in candidates:
+                        if cid and store.get_dataset(cid):
+                            entity_ds_id = cid
+                            break
+
+                if entity_ds_id:
+                    _run_clustering_pipeline(
+                        run=run,
+                        store=store,
+                        entity_ds_id=entity_ds_id,
+                        feature_columns=None,
+                        algorithm='hdbscan',
+                        min_cluster_size=5,
+                        run_pca=True,
+                        head=ctx.get('head'),
+                    )
+                else:
+                    logger.warning(
+                        f'[LLM auto] No entity dataset found for clustering. '
+                        f'Tool calls: {[tc.get("name") for tc in tool_calls]}. '
+                        f'Marking run as completed without clustering.'
+                    )
+                    run.status = 'completed'
+                    run.progress_pct = 100
+                    run.progress_msg = 'LLM分析完成（无实体数据，跳过聚类）'
+                    run.save(update_fields=['status', 'progress_pct', 'progress_msg'])
+
+                # ── Auto-save workflow as loadable plan ──
+                try:
+                    tool_calls_data = run.tool_calls_json or []
+                    steps = []
+                    for tc in tool_calls_data:
+                        tc_name = tc.get('name', '')
+                        if tc_name == 'thinking':
+                            continue
+                        steps.append({
+                            'tool_name': tc_name,
+                            'params': tc.get('input', {}),
+                        })
+                    if steps:
+                        import json as _json
+                        plan_data = {
+                            'name': f'_auto_{run.display_id}',
+                            'steps': steps,
+                            'auto': True,
+                            'display_id': run.display_id,
+                            'created_at': str(run.created_at) if run.created_at else '',
+                        }
+                        plan_path = _PLANS_DIR / f'_auto_{run.display_id}.json'
+                        plan_path.write_text(
+                            _json.dumps(plan_data, ensure_ascii=False, indent=2),
+                            encoding='utf-8',
+                        )
+                        _add_to_auto_index(f'_auto_{run.display_id}.json')
+                        logger.info(
+                            '[LLM auto] Saved auto plan %s (%d steps)',
+                            plan_path.name, len(steps),
+                        )
+                except Exception as save_err:
+                    logger.warning('[LLM auto] Failed to save auto plan: %s', save_err)
+
         except Exception:
             tb = traceback.format_exc()
             logger.error(tb)
@@ -1013,8 +1172,8 @@ def run_llm_analysis_view(request):
                 run.run_log += f'\n[ERROR] {tb}'
                 run.error_message = tb
                 run.save(update_fields=['status', 'progress_msg', 'run_log', 'error_message'])
-            except Exception:
-                pass
+            except Exception as e:
+                logger.warning('save failed after pipeline error: %s', e)
 
     t = threading.Thread(
         target=_run_pipeline_worker,
@@ -1376,15 +1535,60 @@ from pathlib import Path
 _PLANS_DIR = Path(__file__).resolve().parent.parent / '.omo' / 'plans'
 _PLANS_DIR.mkdir(parents=True, exist_ok=True)
 
+_AUTO_INDEX_PATH = _PLANS_DIR / '_auto_index.json'
+_MAX_AUTO_PLANS = 10
+
+
+def _read_auto_index():
+    """Read the auto index file, returning list of {file, pinned}."""
+    import json as _j
+    if not _AUTO_INDEX_PATH.exists():
+        return []
+    try:
+        data = _j.loads(_AUTO_INDEX_PATH.read_text(encoding='utf-8'))
+        if not isinstance(data, list):
+            return []
+        return data
+    except Exception:
+        return []
+
+
+def _write_auto_index(index):
+    """Write the auto index list."""
+    import json as _j
+    _AUTO_INDEX_PATH.write_text(
+        _j.dumps(index, ensure_ascii=False, indent=2),
+        encoding='utf-8',
+    )
+
+
+def _add_to_auto_index(filename):
+    """Prepend filename to auto index, evict unpinned entries when >10."""
+    index = _read_auto_index()
+    # Remove if already present
+    index = [entry for entry in index if entry.get('file') != filename]
+    # Prepend
+    index.insert(0, {'file': filename, 'pinned': False})
+    # Evict unpinned entries beyond max
+    pinned_entries = [e for e in index if e.get('pinned')]
+    unpinned_entries = [e for e in index if not e.get('pinned')]
+    # Keep all pinned + fill remaining slots with unpinned
+    result = pinned_entries + unpinned_entries[:_MAX_AUTO_PLANS - len(pinned_entries)]
+    _write_auto_index(result)
+
 
 @csrf_exempt
 def tool_plan(request):
-    """Save, list, load, or delete an analysis plan."""
+    """Save, list, load, delete, or pin/unpin an analysis plan."""
     import json as _j
     if request.method == 'GET':
         name = request.GET.get('name')
         if name:
-            path = _PLANS_DIR / f'{name}.json'
+            # Auto plans stored with _auto_ prefix
+            if name.startswith('_auto_'):
+                path = _PLANS_DIR / f'{name}.json'
+            else:
+                path = _PLANS_DIR / f'{name}.json'
             if not path.exists():
                 return JsonResponse({'error': f'Plan "{name}" not found'}, status=404)
             try:
@@ -1392,20 +1596,74 @@ def tool_plan(request):
                 return JsonResponse(data)
             except Exception as e:
                 return JsonResponse({'error': str(e)}, status=500)
-        # List all plans
+
+        # List all plans (regular + auto)
+        auto_index = _read_auto_index()
+        auto_index_map = {e.get('file', ''): e.get('pinned', False) for e in auto_index}
+        auto_names = set()
         plans = []
         for p in sorted(_PLANS_DIR.glob('*.json')):
+            # Skip index file and auto plans (handled separately)
+            if p.name == '_auto_index.json':
+                continue
+            if p.name.startswith('_auto_'):
+                auto_names.add(p.name)
+                continue
             try:
                 content = _j.loads(p.read_text(encoding='utf-8'))
                 steps = len(content.get('steps', []))
                 plans.append({'name': p.stem, 'steps': steps})
             except Exception:
                 plans.append({'name': p.stem, 'steps': 0})
-        return JsonResponse({'plans': plans})
+
+        # Add auto plans with markers
+        auto_plans = []
+        for auto_name in sorted(auto_names):
+            try:
+                path = _PLANS_DIR / auto_name
+                content = _j.loads(path.read_text(encoding='utf-8'))
+                steps = len(content.get('steps', []))
+                auto_plans.append({
+                    'name': auto_name.replace('.json', ''),
+                    'steps': steps,
+                    'auto': True,
+                    'pinned': auto_index_map.get(auto_name, False),
+                })
+            except Exception:
+                auto_plans.append({
+                    'name': auto_name.replace('.json', ''),
+                    'steps': 0,
+                    'auto': True,
+                    'pinned': auto_index_map.get(auto_name, False),
+                })
+        # Interleave: auto plans first, then regular
+        return JsonResponse({'plans': auto_plans + plans})
 
     elif request.method == 'POST':
         try:
             body = _j.loads(request.body)
+            # Pin/unpin toggle for auto plans
+            if 'pin' in body and body.get('name', '').startswith('_auto_'):
+                filename = body['name'].replace('.json', '') + '.json' \
+                    if not body['name'].endswith('.json') else body['name']
+                index = _read_auto_index()
+                new_pinned = bool(body['pin'])
+                found = False
+                for entry in index:
+                    if entry.get('file') == filename:
+                        entry['pinned'] = new_pinned
+                        found = True
+                        break
+                if not found and new_pinned:
+                    index.append({'file': filename, 'pinned': True})
+                _write_auto_index(index)
+                return JsonResponse({
+                    'status': 'pinned' if new_pinned else 'unpinned',
+                    'name': body['name'],
+                    'pinned': new_pinned,
+                })
+
+            # Regular save
             name = body.get('name', '').strip()
             steps = body.get('steps', [])
             if not name:
@@ -1423,6 +1681,12 @@ def tool_plan(request):
         path = _PLANS_DIR / f'{name}.json'
         if path.exists():
             path.unlink()
+        # Also remove from auto index if applicable
+        if name.startswith('_auto_'):
+            filename = name + '.json' if not name.endswith('.json') else name
+            index = _read_auto_index()
+            index = [e for e in index if e.get('file') != filename]
+            _write_auto_index(index)
         return JsonResponse({'status': 'deleted', 'name': name})
 
     return JsonResponse({'error': 'Method not allowed'}, status=405)