From eaa33352f08f9dc14aeca9073778089413d04bf7 Mon Sep 17 00:00:00 2001 From: Chenglong Wang Date: Fri, 14 Aug 2026 15:38:45 -0700 Subject: [PATCH] various fixes --- py-src/data_formulator/agents/context.py | 58 +----- .../analyst/skills/data_loading/SKILL.md | 40 +++- .../analyst/skills/data_loading/skill.py | 156 +++++++++++++++ .../analyst/skills/data_loading/tools.json | 45 +++++ py-src/data_formulator/data_connector.py | 30 +++ .../data_formulator/data_loader/__init__.py | 64 +++++- .../data_loader/external_data_loader.py | 24 +++ .../data_operations/discovery.py | 62 +++++- .../data_formulator/datalake/catalog_cache.py | 183 +++++++++++++++++- .../datalake/catalog_refresh.py | 159 +++++++++++++++ src/app/App.tsx | 6 +- src/app/connectorFormPersistence.ts | 16 ++ src/app/dfSlice.tsx | 30 ++- src/app/inputTablePreviewCache.ts | 11 +- src/app/store.ts | 14 +- src/app/useAutoSave.tsx | 5 +- src/app/utils.tsx | 2 +- src/app/workspaceService.ts | 46 ++++- src/components/ComponentType.tsx | 25 ++- src/components/ConnectorFormCard.tsx | 61 +++--- src/i18n/locales/en/common.json | 3 +- src/i18n/locales/zh/common.json | 3 +- src/views/DBTableManager.tsx | 88 +++++++-- src/views/DataFormulator.tsx | 46 +++-- src/views/DataSourceSidebar.tsx | 24 +-- src/views/DataThread.tsx | 55 ++++-- src/views/SimpleChartRecBox.tsx | 77 +++++++- src/views/VisualizationView.tsx | 62 +++++- .../agents/test_analyst_connector_skill.py | 99 ++++++++++ .../test_data_loading_discovery_tools.py | 56 ++++++ tests/backend/data/test_catalog_cache.py | 92 +++++++++ tests/backend/data/test_catalog_refresh.py | 109 +++++++++++ .../unit/app/connectorFormPersistence.test.ts | 53 +++++ .../unit/app/dfSliceTableCollections.test.ts | 69 +++++++ .../unit/app/fetchWithIdentity.test.ts | 17 ++ tests/frontend/unit/app/useAutoSave.test.tsx | 35 +++- .../unit/app/workspaceService.test.ts | 55 +++++- .../unit/views/DataSourceSidebar.test.tsx | 24 +++ 38 files changed, 1806 insertions(+), 198 deletions(-) create mode 100644 py-src/data_formulator/datalake/catalog_refresh.py create mode 100644 src/app/connectorFormPersistence.ts create mode 100644 tests/backend/agents/test_analyst_connector_skill.py create mode 100644 tests/backend/data/test_catalog_refresh.py create mode 100644 tests/frontend/unit/app/connectorFormPersistence.test.ts diff --git a/py-src/data_formulator/agents/context.py b/py-src/data_formulator/agents/context.py index c8988a83..8dc743c9 100644 --- a/py-src/data_formulator/agents/context.py +++ b/py-src/data_formulator/agents/context.py @@ -19,6 +19,7 @@ from data_formulator.agents.agent_utils import ( _format_import_options, ) from data_formulator.datalake.parquet_utils import normalize_dtype_to_app_type +from data_formulator.data_operations.discovery import ensure_no_auth_catalogs_cached logger = logging.getLogger(__name__) @@ -26,61 +27,6 @@ TABLE_SAMPLE_MAX_ROWS = 5 TABLE_SAMPLE_CHAR_LIMIT = 1000 -def _ensure_no_auth_catalogs_cached(user_home: Any) -> None: - """Populate the disk catalog cache for any admin connector that has no - required auth parameters and isn't cached yet. - - Used to surface zero-config admin connectors (notably the built-in - ``sample_datasets`` connector) to the agent's search/read tools on - first use, without requiring an explicit "Connect" step in the UI. - Silent on failure — auth-gated connectors will simply remain - un-synced until the user provides credentials through the normal - flow. - """ - if not user_home: - return - try: - from pathlib import Path - from data_formulator.data_connector import ( - DATA_CONNECTORS, - _ADMIN_CONNECTOR_IDS, - ) - from data_formulator.datalake.catalog_cache import save_catalog - - cache_dir = Path(user_home) / "catalog_cache" - for source_id in list(_ADMIN_CONNECTOR_IDS): - cache_path = cache_dir / f"{source_id}.json" - if cache_path.exists(): - continue - connector = DATA_CONNECTORS.get(source_id) - if not connector: - continue - loader_class = connector._loader_class - try: - params = loader_class.list_params() - except Exception: - continue - # Only auto-sync if no params are required (true no-auth case) - if any(p.get("required") for p in params): - continue - try: - loader = loader_class(connector._default_params or {}) - if not loader.test_connection(): - continue - tables = loader.sync_catalog_metadata() - save_catalog(Path(user_home), source_id, tables) - logger.info( - "Auto-synced catalog for '%s' (%d tables)", - source_id, len(tables), - ) - except Exception: - logger.debug( - "Auto-sync failed for '%s'", source_id, exc_info=True, - ) - except Exception: - logger.debug("Catalog auto-sync setup failed", exc_info=True) - - def _get_workspace_metadata_lookups(workspace: Any) -> tuple[dict[str, str], dict[str, dict[str, str]], dict[str, str]]: """Return table descriptions, column descriptions, and import options from workspace metadata.""" table_descs: dict[str, str] = {} @@ -430,7 +376,7 @@ def handle_read_catalog_metadata( return "Cannot read catalog metadata: user home not available." # Surface zero-config admin connectors (e.g. sample_datasets) on first use. - _ensure_no_auth_catalogs_cached(user_home) + ensure_no_auth_catalogs_cached(user_home) from pathlib import Path from data_formulator.datalake.catalog_cache import load_catalog diff --git a/py-src/data_formulator/analyst/skills/data_loading/SKILL.md b/py-src/data_formulator/analyst/skills/data_loading/SKILL.md index 5db77177..b6f177f8 100644 --- a/py-src/data_formulator/analyst/skills/data_loading/SKILL.md +++ b/py-src/data_formulator/analyst/skills/data_loading/SKILL.md @@ -1,20 +1,25 @@ --- name: data_loading description: >- - Discover connected data sources, inspect table metadata, and run bounded - read-only probes when the current workspace data is insufficient. + Discover connected data sources, add new data connectors through a + user-confirmed form, inspect table metadata, and run bounded read-only probes + when the current workspace data is insufficient. when_to_use: >- The user's question needs data that is not already available as a workspace - input, or the user asks what connected data is available. Not for analyzing - tables already listed in the workspace context. + input, the user asks what connected data is available, or the user wants to + connect a database, warehouse, or cloud source. Not for analyzing tables + already listed in the workspace context. always_on: false tools: - list_data - find_data - describe_data - probe_data + - list_connectors + - describe_connector actions: - - propose_data_operation + - propose_data_operation + - propose_connection --- # Skill: Data discovery @@ -28,6 +33,31 @@ Use these tools to determine whether connected sources contain data needed for the user's goal. They are read-only: discovering, describing, or probing a source does not add anything to the workspace analysis inputs. +## Adding a connector + +When the user wants to connect a new source, do not merely ask them to navigate +to settings and do not attempt to connect on their behalf. + +1. Call `list_connectors` first because available built-ins and plugins vary by + deployment. For a broad request such as "help me connect", summarize the + concrete available types and ask which one they use. +2. Once the source type is known, call `describe_connector` when field or auth + details are useful. +3. **When the requested source type is known and available, you MUST call + `propose_connection` in this same turn.** Do not stop with text such as + "I'll open the form", "you'll need to provide", or a list of required + fields. Only the action opens the form. Include one or two helpful sentences + alongside the action call explaining what the user should review or supply; + this text appears above the chat while the form opens on the canvas. Pass + `prefilled` values the user already supplied, including values parsed from a + connection string or config snippet. Never invent missing values. +4. The form is only a proposal. The user reviews it and clicks Connect; the + action must never connect automatically. + +Prefilled values may include credentials the user deliberately supplied. Do not +repeat those values in prose or subsequent tool output. They are transient form +seeds and are removed from persisted UI state. + ## Discovery sequence 1. Use `find_data` when the user names a business concept or table. Use diff --git a/py-src/data_formulator/analyst/skills/data_loading/skill.py b/py-src/data_formulator/analyst/skills/data_loading/skill.py index 52df39d9..4ebd049d 100644 --- a/py-src/data_formulator/analyst/skills/data_loading/skill.py +++ b/py-src/data_formulator/analyst/skills/data_loading/skill.py @@ -16,6 +16,11 @@ from data_formulator.data_operations import ( ) _PROBE_BUDGET_KEY = "data_loading.probe_budget" +_CONNECTORS_LISTED_KEY = "data_loading.connectors_listed" +_CONNECTORS_DISABLED_NOTE = ( + "External data connectors are disabled in this deployment. Use file upload " + "or built-in sample datasets instead." +) class DataLoadingSkill: @@ -36,6 +41,10 @@ class DataLoadingSkill: result = service.describe_data(args) elif name == "probe_data": result = service.probe_data(args, self._probe_budget(ctx)) + elif name == "list_connectors": + result = self._list_connectors(ctx) + elif name == "describe_connector": + result = self._describe_connector(args) else: result = {"error": f"data_loading has no tool '{name}'."} return ToolResult(text=json.dumps(result, ensure_ascii=False, default=str)) @@ -48,6 +57,8 @@ class DataLoadingSkill: ) -> Generator[Event, None, str | None]: if action == "propose_data_operation": return (yield from self._propose_data_operation(spec, ctx)) + if action == "propose_connection": + return (yield from self._propose_connection(spec, ctx)) message = f"data_loading has no committing action '{action}' in this phase." yield { "type": "error", @@ -56,6 +67,151 @@ class DataLoadingSkill: } return message + @staticmethod + def _connectors_disabled() -> bool: + try: + from flask import current_app + return bool(current_app.config.get("CLI_ARGS", {}).get("disable_data_connectors")) + except Exception: + return False + + @staticmethod + def _skill_state(ctx: SkillContext) -> dict[str, Any]: + state = ctx.payload.get("skill_state") + if not isinstance(state, dict): + state = {} + ctx.payload["skill_state"] = state + return state + + def _list_connectors(self, ctx: SkillContext) -> dict[str, Any]: + self._skill_state(ctx)[_CONNECTORS_LISTED_KEY] = True + if self._connectors_disabled(): + return {"connectors": [], "unavailable": [], "note": _CONNECTORS_DISABLED_NOTE} + + from data_formulator.data_loader import DATA_LOADERS, DISABLED_LOADERS + + connectors = [] + for key, loader_class in DATA_LOADERS.items(): + if key in ("local_folder", "sample_datasets"): + continue + try: + auth_mode = loader_class.auth_mode() + except Exception: + auth_mode = None + connectors.append({ + "type": key, + "name": loader_class.DISPLAY_NAME or key.replace("_", " ").title(), + "summary": loader_class.DESCRIPTION or "", + "auth_mode": auth_mode, + "available": True, + }) + return { + "connectors": connectors, + "unavailable": [ + { + "type": key, + "name": key.replace("_", " ").title(), + "install_hint": hint, + } + for key, hint in DISABLED_LOADERS.items() + if key not in ("local_folder", "sample_datasets") + ], + "next_action": ( + "If the user requested one of these connector types, call " + "propose_connection now. Do not end the turn by saying you will open a form." + ), + } + + def _describe_connector(self, args: dict[str, Any]) -> dict[str, Any]: + if self._connectors_disabled(): + return {"error": _CONNECTORS_DISABLED_NOTE} + + from data_formulator.data_loader import DATA_LOADERS, DISABLED_LOADERS + + source_type = str(args.get("source_type") or "").strip() + loader_class = DATA_LOADERS.get(source_type) + if loader_class is None: + hint = DISABLED_LOADERS.get(source_type) + detail = f" (needs: {hint})" if hint else "" + return {"error": f"Connector {source_type!r} is unavailable{detail}. Call list_connectors."} + + def safe(callable_): + try: + return callable_() + except Exception: + return None + + return { + "type": source_type, + "name": loader_class.DISPLAY_NAME or source_type.replace("_", " ").title(), + "summary": loader_class.DESCRIPTION or "", + "auth_mode": safe(loader_class.auth_mode), + "auth_paths": safe(loader_class.auth_paths), + "auth_instructions": safe(loader_class.auth_instructions), + "params": [ + { + "name": param.get("name"), + "required": bool(param.get("required")), + "tier": param.get("tier"), + "sensitive": bool(param.get("sensitive") or param.get("type") == "password"), + "description": param.get("description"), + } + for param in (safe(loader_class.list_params) or []) + if isinstance(param, dict) + ], + "next_action": ( + "Call propose_connection now to open this form. Describing the " + "requirements in text does not open it." + ), + } + + def _propose_connection( + self, + spec: dict[str, Any], + ctx: SkillContext, + ) -> Generator[Event, None, str | None]: + if self._connectors_disabled(): + yield {"type": "error", "message": _CONNECTORS_DISABLED_NOTE, "message_code": "agent.connectorsDisabled"} + return _CONNECTORS_DISABLED_NOTE + if not self._skill_state(ctx).get(_CONNECTORS_LISTED_KEY): + message = "Call list_connectors before propose_connection." + yield {"type": "error", "message": message, "message_code": "agent.invalidConnector"} + return message + + from data_formulator.data_loader import DATA_LOADERS, DISABLED_LOADERS + + source_type = str(spec.get("source_type") or "").strip() + if source_type not in DATA_LOADERS or source_type in ("local_folder", "sample_datasets"): + hint = DISABLED_LOADERS.get(source_type) + message = f"Connector {source_type!r} is unavailable" + (f" (needs: {hint})." if hint else ".") + yield {"type": "error", "message": message, "message_code": "agent.invalidConnector"} + return message + + prefilled_raw = spec.get("prefilled") or {} + prefilled = {} + if isinstance(prefilled_raw, dict): + prefilled = { + str(key): str(value) + for key, value in prefilled_raw.items() + if value not in (None, "") + } + display_name = DATA_LOADERS[source_type].DISPLAY_NAME or source_type + response = str(ctx.payload.get("action_narration") or "").strip() + yield { + "type": "interact", + "thought": spec.get("thought", ""), + "form": { + "kind": "connector", + "title": f"Connect to {display_name}", + "response": response or f"Complete the {display_name} connection form to add this data source.", + "connector": { + "source_type": source_type, + "prefilled": prefilled, + }, + }, + } + return None + @staticmethod def _already_loaded_tables(steps: tuple[ConnectorQueryStep, ...], workspace) -> list[str]: metadata = workspace.get_metadata() diff --git a/py-src/data_formulator/analyst/skills/data_loading/tools.json b/py-src/data_formulator/analyst/skills/data_loading/tools.json index 2d8e81ec..b19e4ebd 100644 --- a/py-src/data_formulator/analyst/skills/data_loading/tools.json +++ b/py-src/data_formulator/analyst/skills/data_loading/tools.json @@ -110,6 +110,51 @@ } } }, + { + "type": "function", + "function": { + "name": "list_connectors", + "description": "List connector types available in this deployment. Call this before propose_connection because built-ins, plugins, and missing dependencies vary by deployment. If the user's requested type is present, you MUST call propose_connection in the same turn; do not merely say you will open a form.", + "parameters": { + "type": "object", + "properties": {}, + "required": [] + } + } + }, + { + "type": "function", + "function": { + "name": "describe_connector", + "description": "Return setup fields and authentication choices for one source_type returned by list_connectors. After this, call propose_connection in the same turn; describing fields does not open the form.", + "parameters": { + "type": "object", + "properties": { + "source_type": { "type": "string", "description": "Connector type key returned by list_connectors." } + }, + "required": ["source_type"] + } + } + }, + { + "type": "function", + "function": { + "name": "propose_connection", + "description": "REQUIRED terminal action when the user wants an available connector and its source_type is known. This is the only operation that opens the user-confirmed add-connector form on the canvas. Call list_connectors first. Prefill only values the user supplied; never invent credentials or connect automatically.", + "parameters": { + "type": "object", + "properties": { + "source_type": { "type": "string", "description": "Connector type key returned by list_connectors." }, + "prefilled": { + "type": "object", + "description": "Optional connector field values already supplied by the user. Values seed the live form and must not be repeated in prose.", + "additionalProperties": {} + } + }, + "required": ["source_type"] + } + } + }, { "type": "function", "function": { diff --git a/py-src/data_formulator/data_connector.py b/py-src/data_formulator/data_connector.py index c086120f..3fb3f837 100644 --- a/py-src/data_formulator/data_connector.py +++ b/py-src/data_formulator/data_connector.py @@ -816,6 +816,34 @@ def resolve_live_loader(source_id: str) -> "ExternalDataLoader": return connector._require_loader() +def resolve_catalog_refresh_target( + source_id: str, +) -> "tuple[type[ExternalDataLoader], ExternalDataLoader | None]": + """Resolve policy and an existing loader without reconnecting credentials.""" + if source_id in _ADMIN_CONNECTOR_IDS and source_id in DATA_CONNECTORS: + connector = DATA_CONNECTORS[source_id] + else: + _, connector = _resolve_connector_with_key({"connector_id": source_id}) + loader_class = connector._loader_class + loader = None + if hasattr(connector, "_get_identity") and hasattr(connector, "_get_loader"): + identity = connector._get_identity() + loader = connector._get_loader(identity) + try: + auth_mode = _loader_auth_mode(loader_class) + except AttributeError: + auth_mode = ( + "credentials" + if any(param.get("required") for param in loader_class.list_params()) + else "none" + ) + if loader is None and auth_mode == "none": + loader = loader_class(connector._default_params or {}) + if hasattr(connector, "_loaders") and 'identity' in locals(): + connector._loaders[identity] = loader + return loader_class, loader + + def connector_is_available(source_id: str) -> bool | None: """Whether ``source_id`` could be loaded from right now, without touching it. @@ -1680,6 +1708,7 @@ def connector_connect(): save_catalog( user_home, source._source_id, flat_tables, mode=cache_mode, + refresh_kind="listing", ) except Exception: logger.debug("Failed to save catalog cache on connect for '%s'", @@ -1938,6 +1967,7 @@ def connector_get_catalog_tree(): save_catalog( user_home, source._source_id, flat_tables, mode="replace", + refresh_kind="listing", ) except Exception: logger.debug( diff --git a/py-src/data_formulator/data_loader/__init__.py b/py-src/data_formulator/data_loader/__init__.py index 278a19f7..f4a3c1e2 100644 --- a/py-src/data_formulator/data_loader/__init__.py +++ b/py-src/data_formulator/data_loader/__init__.py @@ -46,7 +46,11 @@ import os import sys from pathlib import Path -from data_formulator.data_loader.external_data_loader import ExternalDataLoader, CatalogNode +from data_formulator.data_loader.external_data_loader import ( + CatalogCachePolicy, + CatalogNode, + ExternalDataLoader, +) _log = logging.getLogger(__name__) @@ -83,12 +87,68 @@ PLUGIN_LOADERS: dict[str, str] = {} # key -> absolute source file path PLUGIN_ERRORS: list[dict] = [] # rejected plugin attempts (override / duplicate) _BUILTIN_KEYS: frozenset[str] = frozenset(spec[0] for spec in _LOADER_SPECS) +_BUILTIN_CATALOG_POLICIES: dict[str, CatalogCachePolicy] = { + "sample_datasets": CatalogCachePolicy( + listing_ttl_seconds=86_400, + metadata_ttl_seconds=86_400, + refresh_cost="free", + automatic_refresh="always", + automatic_refresh_kind="full", + ), + "local_folder": CatalogCachePolicy( + listing_ttl_seconds=0, + metadata_ttl_seconds=0, + refresh_cost="local", + automatic_refresh="while_connected", + minimum_retry_seconds=5, + ), + **{ + key: CatalogCachePolicy( + listing_ttl_seconds=3_600, + metadata_ttl_seconds=21_600, + refresh_cost="cheap", + automatic_refresh="while_connected", + automatic_refresh_kind="full", + ) + for key in ("mysql", "postgresql", "mssql", "clickhouse") + }, + **{ + key: CatalogCachePolicy( + listing_ttl_seconds=900, + metadata_ttl_seconds=21_600, + refresh_cost="moderate", + automatic_refresh="while_connected", + ) + for key in ("s3", "azure_blob") + }, + **{ + key: CatalogCachePolicy( + listing_ttl_seconds=7_200, + metadata_ttl_seconds=43_200, + refresh_cost="moderate", + automatic_refresh="while_connected", + ) + for key in ("databricks", "bigquery", "athena") + }, + **{ + key: CatalogCachePolicy( + listing_ttl_seconds=21_600 if key == "kusto" else 7_200, + metadata_ttl_seconds=86_400 if key == "kusto" else 43_200, + refresh_cost="expensive", + automatic_refresh="never", + ) + for key in ("kusto", "mongodb", "cosmosdb", "superset") + }, +} + def _scan_package_loaders() -> None: """Import built-in loaders from ``_LOADER_SPECS``.""" for key, module_path, cls_name, pip_pkg in _LOADER_SPECS: try: mod = importlib.import_module(module_path) - DATA_LOADERS[key] = getattr(mod, cls_name) + loader_class = getattr(mod, cls_name) + loader_class.CATALOG_CACHE_POLICY = _BUILTIN_CATALOG_POLICIES[key] + DATA_LOADERS[key] = loader_class except ImportError as exc: hint = f"pip install {pip_pkg}" DISABLED_LOADERS[key] = hint diff --git a/py-src/data_formulator/data_loader/external_data_loader.py b/py-src/data_formulator/data_loader/external_data_loader.py index 858bfad5..9545577e 100644 --- a/py-src/data_formulator/data_loader/external_data_loader.py +++ b/py-src/data_formulator/data_loader/external_data_loader.py @@ -33,6 +33,24 @@ if TYPE_CHECKING: logger = logging.getLogger(__name__) +@dataclass(frozen=True) +class CatalogCachePolicy: + listing_ttl_seconds: int | None = 21_600 + metadata_ttl_seconds: int | None = 86_400 + refresh_cost: str = "expensive" + automatic_refresh: str = "never" + automatic_refresh_kind: str = "listing" + minimum_retry_seconds: int = 300 + + def __post_init__(self) -> None: + if self.refresh_cost not in {"local", "free", "cheap", "moderate", "expensive"}: + raise ValueError(f"Unsupported catalog refresh cost: {self.refresh_cost}") + if self.automatic_refresh not in {"always", "while_connected", "never"}: + raise ValueError(f"Unsupported automatic catalog refresh mode: {self.automatic_refresh}") + if self.automatic_refresh_kind not in {"listing", "full"}: + raise ValueError(f"Unsupported automatic catalog refresh kind: {self.automatic_refresh_kind}") + + class ConnectorParamError(ValueError): """Raised when required connector parameters are missing or empty.""" @@ -384,6 +402,12 @@ class ExternalDataLoader(ABC): # so the frontend can show what's happening alongside the spinner. # Loaders that don't report progress simply never call the helper. progress_callback: Callable[[str], None] | None = None + CATALOG_CACHE_POLICY = CatalogCachePolicy() + + @classmethod + def catalog_cache_policy(cls) -> CatalogCachePolicy: + """Describe catalog freshness and safe automatic refresh behavior.""" + return cls.CATALOG_CACHE_POLICY def _report_progress(self, message: str) -> None: """Emit a high-level progress message if a sink is attached. diff --git a/py-src/data_formulator/data_operations/discovery.py b/py-src/data_formulator/data_operations/discovery.py index ee04e8fc..f0af874f 100644 --- a/py-src/data_formulator/data_operations/discovery.py +++ b/py-src/data_formulator/data_operations/discovery.py @@ -44,6 +44,41 @@ STANDALONE_PROBE_GUIDANCE = ProbeGuidance( ) +def ensure_catalogs_current(user_home: Any) -> dict[str, Any]: + """Apply source policies and return current snapshots, including stale ones.""" + if not user_home: + return {} + snapshots: dict[str, Any] = {} + try: + from data_formulator.data_connector import _ADMIN_CONNECTOR_IDS + from data_formulator.datalake.catalog_cache import list_cached_sources + from data_formulator.datalake.catalog_refresh import ensure_catalog_freshness + + source_ids = set(list_cached_sources(user_home)) | set(_ADMIN_CONNECTOR_IDS) + for source_id in source_ids: + snapshot = ensure_catalog_freshness(Path(user_home), source_id) + if snapshot is not None: + snapshots[source_id] = snapshot + except Exception: + logger.debug("Catalog freshness setup failed", exc_info=True) + return snapshots + + +def ensure_no_auth_catalogs_cached(user_home: Any) -> None: + """Backward-compatible alias for callers migrated from catalog bootstrap.""" + ensure_catalogs_current(user_home) + + +def _freshness_payload(snapshot: Any) -> dict[str, Any]: + return { + "listing": snapshot.listing_freshness, + "metadata": snapshot.metadata_freshness, + "listing_age_seconds": snapshot.listing_age_seconds, + "metadata_age_seconds": snapshot.metadata_age_seconds, + "last_refresh_error": snapshot.last_refresh_error, + } + + class DataDiscoveryService: """Read-only catalog discovery shared by data-loading entry points.""" @@ -59,6 +94,7 @@ class DataDiscoveryService: user_home = getattr(self.workspace, "user_home", None) if not user_home: return {"sources": []} + snapshots = ensure_catalogs_current(user_home) source_id = (args.get("source_id") or "").strip() if not source_id: @@ -73,6 +109,12 @@ class DataDiscoveryService: from data_formulator.data_connector import connector_is_available for source in sources: sid = source.get("source_id") or source.get("id") + if sid in snapshots and ( + snapshots[sid].listing_freshness != "fresh" + or snapshots[sid].metadata_freshness != "fresh" + or snapshots[sid].last_refresh_error + ): + source["freshness"] = _freshness_payload(snapshots[sid]) if sid and connector_is_available(sid) is False: source["connected"] = False except Exception: @@ -84,12 +126,15 @@ class DataDiscoveryService: return {"error": "path must be an array of strings"} try: - return list_path_children( + result = list_path_children( user_home, source_id, path=path, filter=args.get("filter"), ) + if source_id in snapshots: + result["freshness"] = _freshness_payload(snapshots[source_id]) + return result except Exception as exc: logger.debug("list_data: list_path_children failed", exc_info=True) return {"error": f"list_data failed: {exc}"} @@ -133,6 +178,7 @@ class DataDiscoveryService: source_ids = [scope_raw] user_home = getattr(self.workspace, "user_home", None) + snapshots = ensure_catalogs_current(user_home) results: list[dict[str, Any]] = [] if search_workspace: @@ -186,13 +232,25 @@ class DataDiscoveryService: return { "results": [], "valid_source_ids": known, + "catalog_freshness": { + source_id: _freshness_payload(snapshot) + for source_id, snapshot in snapshots.items() + }, "note": ( f"No tables matched query={query!r} scope={scope_raw!r}. " "Try a broader pattern, alternation (a|b), or list_data to browse." ), } - return {"results": results[:limit], "query": query, "scope": scope_raw} + return { + "results": results[:limit], + "query": query, + "scope": scope_raw, + "catalog_freshness": { + source_id: _freshness_payload(snapshot) + for source_id, snapshot in snapshots.items() + }, + } def describe_data(self, args: dict[str, Any]) -> dict[str, Any]: from data_formulator.agents.context import handle_read_catalog_metadata diff --git a/py-src/data_formulator/datalake/catalog_cache.py b/py-src/data_formulator/datalake/catalog_cache.py index 091f6929..68ea9810 100644 --- a/py-src/data_formulator/datalake/catalog_cache.py +++ b/py-src/data_formulator/datalake/catalog_cache.py @@ -23,9 +23,11 @@ import json import logging import os import re +from dataclasses import dataclass from datetime import datetime, timezone from pathlib import Path from typing import Any +from uuid import uuid4 from data_formulator.datalake.naming import safe_source_id from data_formulator.security.path_safety import ConfinedDir @@ -33,6 +35,53 @@ from data_formulator.security.path_safety import ConfinedDir logger = logging.getLogger(__name__) CATALOG_CACHE_DIR = "catalog_cache" +CATALOG_CACHE_SCHEMA_VERSION = 2 + + +@dataclass(frozen=True) +class CatalogSnapshot: + source_id: str + tables: list[dict[str, Any]] + listing_refreshed_at: str | None + metadata_refreshed_at: str | None + listing_age_seconds: int | None + metadata_age_seconds: int | None + listing_freshness: str + metadata_freshness: str + refresh_kind: str + refresh_status: str + last_refresh_attempt_at: str | None + last_refresh_error: str | None + + +def _utc_now() -> datetime: + return datetime.now(timezone.utc) + + +def _parse_timestamp(value: Any) -> datetime | None: + if not isinstance(value, str) or not value.strip(): + return None + try: + parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) + if parsed.tzinfo is None: + parsed = parsed.replace(tzinfo=timezone.utc) + return parsed.astimezone(timezone.utc) + except ValueError: + return None + + +def _age_and_freshness( + value: Any, + ttl_seconds: int | None, + now: datetime, +) -> tuple[int | None, str]: + timestamp = _parse_timestamp(value) + if timestamp is None: + return None, "unknown" + age = max(0, int((now - timestamp).total_seconds())) + if ttl_seconds is None: + return age, "fresh" + return age, "fresh" if age <= max(0, ttl_seconds) else "stale" class CatalogSearchError(ValueError): @@ -64,12 +113,65 @@ def _cache_file( return _cache_jail(workspace_root, mkdir=mkdir).resolve(_cache_filename(source_id)) +def _atomic_write_payload( + workspace_root: Path | str, + source_id: str, + payload: dict[str, Any], +) -> None: + jail = _cache_jail(workspace_root, mkdir=True) + filename = _cache_filename(source_id) + target = jail.resolve(filename) + temporary = jail.resolve(f".{filename}.{os.getpid()}.{uuid4().hex}.tmp") + try: + with open(temporary, "w", encoding="utf-8") as file: + json.dump(payload, file, ensure_ascii=False, default=str) + file.flush() + os.fsync(file.fileno()) + os.replace(temporary, target) + finally: + if temporary.exists(): + temporary.unlink() + + +def _merge_catalog_listing( + previous_tables: list[dict[str, Any]], + listing_tables: list[dict[str, Any]], +) -> list[dict[str, Any]]: + def identity(table: dict[str, Any]) -> tuple[Any, ...]: + return ( + table.get("table_key") or "", + tuple(table.get("path") or []), + table.get("name") or "", + ) + + previous_by_identity = {identity(table): table for table in previous_tables} + merged: list[dict[str, Any]] = [] + for table in listing_tables: + prior = previous_by_identity.get(identity(table)) + if not prior: + merged.append(table) + continue + combined = {**prior, **table} + prior_metadata = prior.get("metadata") or {} + listing_metadata = table.get("metadata") or {} + metadata = dict(prior_metadata) + for key, value in listing_metadata.items(): + if value not in (None, "", [], {}): + metadata[key] = value + if metadata: + combined["metadata"] = metadata + merged.append(combined) + return merged + + def save_catalog( workspace_root: Path | str, source_id: str, tables: list[dict[str, Any]], *, mode: str = "replace", + refresh_kind: str = "full", + refresh_status: str = "complete", ) -> None: """Persist catalog data to disk. Best-effort — errors are logged, not raised. @@ -85,18 +187,55 @@ def save_catalog( if mode not in ("replace", "seed_if_missing"): logger.debug("Unknown catalog cache save mode %s for %s", mode, source_id) mode = "replace" + previous = _load_catalog_raw(workspace_root, source_id) or {} + now = _utc_now().isoformat() + if refresh_kind not in ("listing", "full"): + refresh_kind = "full" + if refresh_kind == "listing": + tables = _merge_catalog_listing(previous.get("tables") or [], tables) + metadata_refreshed_at = ( + now + if refresh_kind == "full" + else previous.get("metadata_refreshed_at") or previous.get("synced_at") + ) payload = { + "schema_version": CATALOG_CACHE_SCHEMA_VERSION, "source_id": source_id, - "synced_at": datetime.now(timezone.utc).isoformat(), + "synced_at": now, + "listing_refreshed_at": now, + "metadata_refreshed_at": metadata_refreshed_at, + "refresh_kind": refresh_kind, + "refresh_status": refresh_status, + "last_refresh_attempt_at": now, + "last_refresh_error": None, "tables": tables, } - with open(path, "w", encoding="utf-8") as f: - json.dump(payload, f, ensure_ascii=False, default=str) + _atomic_write_payload(workspace_root, source_id, payload) logger.debug("Catalog cache written: %s (%d tables)", path, len(tables)) except Exception: logger.debug("Failed to write catalog cache for %s", source_id, exc_info=True) +def record_catalog_refresh_failure( + workspace_root: Path | str, + source_id: str, + error: str, +) -> None: + """Record a safe refresh failure without replacing the last good tables.""" + try: + payload = _load_catalog_raw(workspace_root, source_id) + if not payload: + return + payload = dict(payload) + payload["schema_version"] = CATALOG_CACHE_SCHEMA_VERSION + payload["refresh_status"] = "failed" + payload["last_refresh_attempt_at"] = _utc_now().isoformat() + payload["last_refresh_error"] = error[:160] + _atomic_write_payload(workspace_root, source_id, payload) + except Exception: + logger.debug("Failed to record catalog refresh failure for %s", source_id, exc_info=True) + + def _load_catalog_raw(workspace_root: Path | str, source_id: str) -> dict[str, Any] | None: """Load raw catalog JSON (including original ``source_id`` key).""" path: Path | None = None @@ -137,6 +276,44 @@ def load_catalog(workspace_root: Path | str, source_id: str) -> list[dict[str, A return raw.get("tables", []) +def load_catalog_snapshot( + workspace_root: Path | str, + source_id: str, + *, + listing_ttl_seconds: int | None, + metadata_ttl_seconds: int | None, + now: datetime | None = None, +) -> CatalogSnapshot | None: + """Load cached tables with independent listing and metadata freshness.""" + raw = _load_catalog_raw(workspace_root, source_id) + if raw is None: + return None + current = (now or _utc_now()).astimezone(timezone.utc) + legacy_synced_at = raw.get("synced_at") + listing_refreshed_at = raw.get("listing_refreshed_at") or legacy_synced_at + metadata_refreshed_at = raw.get("metadata_refreshed_at") or legacy_synced_at + listing_age, listing_freshness = _age_and_freshness( + listing_refreshed_at, listing_ttl_seconds, current, + ) + metadata_age, metadata_freshness = _age_and_freshness( + metadata_refreshed_at, metadata_ttl_seconds, current, + ) + return CatalogSnapshot( + source_id=str(raw.get("source_id") or source_id), + tables=raw.get("tables") or [], + listing_refreshed_at=listing_refreshed_at, + metadata_refreshed_at=metadata_refreshed_at, + listing_age_seconds=listing_age, + metadata_age_seconds=metadata_age, + listing_freshness=listing_freshness, + metadata_freshness=metadata_freshness, + refresh_kind=str(raw.get("refresh_kind") or "full"), + refresh_status=str(raw.get("refresh_status") or "complete"), + last_refresh_attempt_at=raw.get("last_refresh_attempt_at"), + last_refresh_error=raw.get("last_refresh_error"), + ) + + def delete_catalog(workspace_root: Path | str, source_id: str) -> None: """Remove cached catalog file. Best-effort.""" try: diff --git a/py-src/data_formulator/datalake/catalog_refresh.py b/py-src/data_formulator/datalake/catalog_refresh.py new file mode 100644 index 00000000..9bcffdc2 --- /dev/null +++ b/py-src/data_formulator/datalake/catalog_refresh.py @@ -0,0 +1,159 @@ +from __future__ import annotations + +import logging +import threading +from concurrent.futures import ThreadPoolExecutor +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + +from data_formulator.datalake.catalog_cache import ( + CatalogSnapshot, + load_catalog_snapshot, + record_catalog_refresh_failure, + save_catalog, +) +from data_formulator.data_loader.external_data_loader import CatalogCachePolicy + +logger = logging.getLogger(__name__) + +_REFRESH_EXECUTOR = ThreadPoolExecutor(max_workers=4, thread_name_prefix="catalog-refresh") +_REFRESH_LOCK = threading.Lock() +_REFRESHING: set[tuple[str, str]] = set() + + +def _retry_allowed(snapshot: CatalogSnapshot, policy: CatalogCachePolicy) -> bool: + if not snapshot.last_refresh_error or not snapshot.last_refresh_attempt_at: + return True + try: + attempted_at = datetime.fromisoformat( + snapshot.last_refresh_attempt_at.replace("Z", "+00:00") + ) + if attempted_at.tzinfo is None: + attempted_at = attempted_at.replace(tzinfo=timezone.utc) + elapsed = (datetime.now(timezone.utc) - attempted_at).total_seconds() + return elapsed >= max(0, policy.minimum_retry_seconds) + except ValueError: + return True + + +def _refresh_catalog( + workspace_root: Path, + source_id: str, + loader: Any, + policy: CatalogCachePolicy, +) -> None: + try: + if policy.automatic_refresh_kind == "full": + tables = loader.sync_catalog_metadata() + else: + tables = loader.list_tables() + if hasattr(loader, "ensure_table_keys"): + loader.ensure_table_keys(tables) + save_catalog( + workspace_root, + source_id, + tables, + refresh_kind=policy.automatic_refresh_kind, + ) + except Exception as exc: + record_catalog_refresh_failure(workspace_root, source_id, str(exc)) + logger.debug("Automatic catalog refresh failed for %s", source_id, exc_info=True) + + +def _complete_refresh( + key: tuple[str, str], + workspace_root: Path, + source_id: str, + loader: Any, + policy: CatalogCachePolicy, +) -> None: + try: + _refresh_catalog(workspace_root, source_id, loader, policy) + finally: + with _REFRESH_LOCK: + _REFRESHING.discard(key) + + +def ensure_catalog_freshness( + workspace_root: Path | str, + source_id: str, +) -> CatalogSnapshot | None: + """Return the current snapshot and safely refresh it when policy permits.""" + from data_formulator.data_connector import resolve_catalog_refresh_target + + root = Path(workspace_root) + try: + loader_class, loader = resolve_catalog_refresh_target(source_id) + except Exception: + return load_catalog_snapshot( + root, + source_id, + listing_ttl_seconds=None, + metadata_ttl_seconds=None, + ) + + if hasattr(loader_class, "catalog_cache_policy"): + policy = loader_class.catalog_cache_policy() + else: + policy = CatalogCachePolicy( + refresh_cost="free", + automatic_refresh="always", + automatic_refresh_kind="full", + ) + snapshot = load_catalog_snapshot( + root, + source_id, + listing_ttl_seconds=policy.listing_ttl_seconds, + metadata_ttl_seconds=policy.metadata_ttl_seconds, + ) + may_refresh = ( + loader is not None + and policy.automatic_refresh != "never" + and ( + policy.automatic_refresh == "always" + or policy.automatic_refresh == "while_connected" + ) + ) + if not may_refresh: + return snapshot + + target_freshness = ( + snapshot.metadata_freshness + if snapshot is not None and policy.automatic_refresh_kind == "full" + else snapshot.listing_freshness if snapshot is not None else "unknown" + ) + should_refresh = ( + snapshot is None + or ( + target_freshness != "fresh" + and _retry_allowed(snapshot, policy) + ) + ) + if not should_refresh: + return snapshot + + key = (str(root.resolve()), source_id) + with _REFRESH_LOCK: + if key in _REFRESHING: + return snapshot + _REFRESHING.add(key) + + if snapshot is None: + _complete_refresh(key, root, source_id, loader, policy) + return load_catalog_snapshot( + root, + source_id, + listing_ttl_seconds=policy.listing_ttl_seconds, + metadata_ttl_seconds=policy.metadata_ttl_seconds, + ) + + _REFRESH_EXECUTOR.submit( + _complete_refresh, + key, + root, + source_id, + loader, + policy, + ) + return snapshot \ No newline at end of file diff --git a/src/app/App.tsx b/src/app/App.tsx index 585982d4..55211f54 100644 --- a/src/app/App.tsx +++ b/src/app/App.tsx @@ -95,7 +95,7 @@ import FolderOpenIcon from '@mui/icons-material/FolderOpen'; import RefreshIcon from '@mui/icons-material/Refresh'; import { getUrls } from './utils'; import { apiRequest } from './apiClient'; -import { listWorkspaces, loadWorkspace, deleteWorkspace, saveWorkspaceState, onWorkspaceListChanged } from './workspaceService'; +import { listWorkspaces, loadWorkspace, deleteWorkspace, saveWorkspaceState, onWorkspaceListChanged, WorkspaceLoadSupersededError } from './workspaceService'; import { getSerializableState } from './useAutoSave'; import store, { persistor } from './store'; import { UnifiedDataUploadDialog } from '../views/UnifiedDataUploadDialog'; @@ -500,6 +500,10 @@ const WorkspacePickerDialog: React.FC<{open: boolean, onClose: () => void}> = ({ dispatch(dfActions.addMessages({ timestamp: Date.now(), component: "Workspace", type: "error", value: t('workspace.failedToOpenWorkspace') })); } } catch (e) { + if (e instanceof WorkspaceLoadSupersededError) { + setLoading(false); + return; + } dispatch(dfActions.addMessages({ timestamp: Date.now(), component: "Workspace", type: "error", value: t('workspace.failedToOpenWorkspace') })); } setLoading(false); diff --git a/src/app/connectorFormPersistence.ts b/src/app/connectorFormPersistence.ts new file mode 100644 index 00000000..0ad5bbb4 --- /dev/null +++ b/src/app/connectorFormPersistence.ts @@ -0,0 +1,16 @@ +// Connector-form prefills may contain credentials supplied in chat. Keep them +// in memory only, regardless of which persistence path serializes the state. +export const stripConnectorPrefillFromEntries = (entries: unknown) => { + if (!Array.isArray(entries)) return entries; + return entries.map((entry: any) => { + if (entry?.form?.kind === 'connector' && entry.form.connector?.prefilled) { + const { prefilled, ...connector } = entry.form.connector; + return { ...entry, form: { ...entry.form, connector } }; + } + if (entry?.connectorForm?.prefilled) { + const { prefilled, ...connectorForm } = entry.connectorForm; + return { ...entry, connectorForm }; + } + return entry; + }); +}; \ No newline at end of file diff --git a/src/app/dfSlice.tsx b/src/app/dfSlice.tsx index 60b83f10..0fe323fa 100644 --- a/src/app/dfSlice.tsx +++ b/src/app/dfSlice.tsx @@ -2105,7 +2105,23 @@ export const dataFormulatorSlice = createSlice({ const turnId = action.payload; const turn = state.textTurns.find(a => a.id === turnId); const wasFocused = state.focusedId?.type === 'text' && state.focusedId.textId === turnId; + const parentTurn = turn + ? state.textTurns.find(candidate => candidate.id === turn.parentNodeId) + : undefined; + const hasSiblingTurns = !!turn && state.textTurns.some(candidate => + candidate.id !== turnId && candidate.parentNodeId === turn.parentNodeId); + const hasProducedArtifacts = !!turn && ( + state.derivedTables.some(table => table.parentNodeId === turnId) + || state.loadedTableNodes.some(node => node.parentNodeId === turnId) + || state.draftNodes.some(draft => draft.parentNodeId === turnId) + || state.generatedReports.some(report => report.parentNodeId === turnId) + || state.textTurns.some(child => child.parentNodeId === turnId) + ); state.textTurns = state.textTurns.filter(a => a.id !== turnId); + if (parentTurn?.answered && parentTurn.answer && !hasSiblingTurns && !hasProducedArtifacts) { + parentTurn.answered = false; + delete parentTurn.answer; + } if (turn) { state.textTurns = state.textTurns.map(child => child.parentNodeId === turnId @@ -2646,20 +2662,24 @@ export const dfSelectors = { if (focusedId?.type !== 'text') return focusedId; const art = textTurns.find(a => a.id === focusedId.textId); if (!art) return undefined; - if (art.dataOperation) return focusedId; + if (art.dataOperation || art.form) return focusedId; if (art.sourceChartId && [...userCharts, ...triggerCharts].some(c => c.id === art.sourceChartId)) { return { type: 'chart', chartId: art.sourceChartId }; } - // No source chart: resolve the turn's thread position to the nearest - // artifact above it (design-docs/42) — that table's chart when it has - // one, since the chart is what the turn is talking about. + // Resolve the turn's thread position to the nearest canvas-owning + // artifact above it. Follow-up explanations keep a form or data + // preview visible while their own text opens over the chat. let cur: TextTurn | undefined = art; const seen = new Set(); while (cur && !seen.has(cur.id)) { seen.add(cur.id); const p: string | undefined = cur.parentNodeId; if (!p) break; + const parentTurn = textTurns.find(tt => tt.id === p); + if (parentTurn?.dataOperation || parentTurn?.form) { + return { type: 'text', textId: parentTurn.id }; + } if (tables.some(t => t.id === p)) { // Charts render in `getAllCharts` order, so the last one on // the table is the chart sitting just above this turn. @@ -2667,7 +2687,7 @@ export const dfSelectors = { const nearest = tableCharts[tableCharts.length - 1]; return nearest ? { type: 'chart', chartId: nearest.id } : { type: 'table', tableId: p }; } - cur = textTurns.find(tt => tt.id === p); + cur = parentTurn; } return undefined; }, diff --git a/src/app/inputTablePreviewCache.ts b/src/app/inputTablePreviewCache.ts index e2e05dd5..e5cf674c 100644 --- a/src/app/inputTablePreviewCache.ts +++ b/src/app/inputTablePreviewCache.ts @@ -37,4 +37,13 @@ export const invalidateInputTablePreview = (tableId: string): void => { } }; -export const clearInputTablePreviewCache = (): void => previewCache.clear(); \ No newline at end of file +export const clearInputTablePreviewCache = (): void => previewCache.clear(); + +export const replaceInputTablePreviews = ( + previews: Array<{ table: InputTable; rows: Record[]; fetchedAt?: number }>, +): void => { + previewCache.clear(); + for (const { table, rows, fetchedAt } of previews) { + setInputTablePreview(table, rows, fetchedAt); + } +}; \ No newline at end of file diff --git a/src/app/store.ts b/src/app/store.ts index bbd62e36..14b9677b 100644 --- a/src/app/store.ts +++ b/src/app/store.ts @@ -7,6 +7,7 @@ import { dataFormulatorReducer } from './dfSlice'; import { persistReducer, persistStore, createTransform } from 'redux-persist' import localforage from 'localforage'; import { migrateState } from './stateMigrations'; +import { stripConnectorPrefillFromEntries } from './connectorFormPersistence'; export type AppDispatch = typeof store.dispatch @@ -17,18 +18,9 @@ export type AppDispatch = typeof store.dispatch // `prefilled` so the form can seed once on render; this transform drops it on the // way to localForage, so nothing is stored until the user clicks Connect. const stripConnectorPrefill = createTransform( - (inboundState: any) => { - if (!Array.isArray(inboundState)) return inboundState; - return inboundState.map((message: any) => { - if (message?.connectorForm?.prefilled) { - const { prefilled, ...connectorForm } = message.connectorForm; - return { ...message, connectorForm }; - } - return message; - }); - }, + stripConnectorPrefillFromEntries, (outboundState: any) => outboundState, - { whitelist: ['dataLoadingChatMessages'] }, + { whitelist: ['dataLoadingChatMessages', 'textTurns'] }, ); const persistConfig = { diff --git a/src/app/useAutoSave.tsx b/src/app/useAutoSave.tsx index cfb5bec9..0289bcb8 100644 --- a/src/app/useAutoSave.tsx +++ b/src/app/useAutoSave.tsx @@ -7,6 +7,7 @@ import { DataFormulatorState, dfSelectors } from './dfSlice'; import { saveWorkspaceState } from './workspaceService'; import { handleApiError } from './errorHandler'; import { DF_STATE_VERSION } from './stateMigrations'; +import { stripConnectorPrefillFromEntries } from './connectorFormPersistence'; /** * Fields excluded from auto-save (secrets / ephemeral / fetched-on-startup). @@ -39,7 +40,9 @@ export function getSerializableState(state: DataFormulatorState): Record = {}; for (const [key, value] of Object.entries(state)) { if (!EXCLUDED_FIELDS.has(key)) { - result[key] = value; + result[key] = key === 'dataLoadingChatMessages' || key === 'textTurns' + ? stripConnectorPrefillFromEntries(value) + : value; } } // Stamp the schema version so `migrateState` can upgrade this payload on a diff --git a/src/app/utils.tsx b/src/app/utils.tsx index 64894831..62b3635b 100644 --- a/src/app/utils.tsx +++ b/src/app/utils.tsx @@ -185,7 +185,7 @@ async function _doFetch( headers.set('X-Identity-Id', namespacedIdentity); const workspaceId = await getActiveWorkspaceId(); - if (workspaceId) { + if (workspaceId && !headers.has('X-Workspace-Id')) { headers.set('X-Workspace-Id', workspaceId); } diff --git a/src/app/workspaceService.ts b/src/app/workspaceService.ts index 5230708e..d4c9fd4d 100644 --- a/src/app/workspaceService.ts +++ b/src/app/workspaceService.ts @@ -11,7 +11,7 @@ import { fetchWithIdentity, getUrls } from './utils'; import { apiRequest, ApiRequestError, assertDownloadResponseOk } from './apiClient'; import { workspaceDB, TableIndexEntry } from './workspaceDB'; -import { clearInputTablePreviewCache, INPUT_TABLE_PREVIEW_ROW_LIMIT, setInputTablePreview } from './inputTablePreviewCache'; +import { INPUT_TABLE_PREVIEW_ROW_LIMIT, replaceInputTablePreviews } from './inputTablePreviewCache'; import { migrateState } from './stateMigrations'; import { workspaceTableIdOf } from './tableResolution'; import type { InputTable } from '../components/ComponentType'; @@ -80,22 +80,46 @@ function _notifyListChanged(): void { window.dispatchEvent(new Event(WORKSPACE_LIST_CHANGED)); } -async function hydrateInputTablePreviews(state: Record): Promise { - clearInputTablePreviewCache(); +type PreparedInputTablePreview = { + table: InputTable; + rows: Record[]; +}; + +let workspaceLoadGeneration = 0; + +export class WorkspaceLoadSupersededError extends Error { + readonly name = 'WorkspaceLoadSupersededError'; +} + +function assertCurrentWorkspaceLoad(generation: number): void { + if (generation !== workspaceLoadGeneration) { + throw new WorkspaceLoadSupersededError('A newer workspace load has started'); + } +} + +async function prepareInputTablePreviews( + state: Record, + workspaceId: string, +): Promise { const inputTables = (state.inputTables || []) as InputTable[]; - await Promise.all(inputTables.map(async table => { + const previews = await Promise.all(inputTables.map(async table => { const workspaceTableId = workspaceTableIdOf(table); try { const { data } = await apiRequest<{ rows: Record[] }>(getUrls().SAMPLE_TABLE, { method: 'POST', - headers: { 'Content-Type': 'application/json' }, + headers: { + 'Content-Type': 'application/json', + 'X-Workspace-Id': workspaceId, + }, body: JSON.stringify({ table: workspaceTableId, size: INPUT_TABLE_PREVIEW_ROW_LIMIT }), }); - setInputTablePreview(table, data.rows || []); + return { table, rows: data.rows || [] }; } catch (error) { console.warn(`Failed to hydrate preview for ${table.id}:`, error); + return null; } })); + return previews.filter((preview): preview is PreparedInputTablePreview => preview !== null); } // ── Workspace CRUD ────────────────────────────────────────────────────── @@ -123,7 +147,9 @@ export async function listWorkspaces(): Promise { /** Load a workspace's saved state. Returns null if not found. */ export async function loadWorkspace(id: string): Promise<{ state: Record; displayName: string; readOnly: boolean } | null> { + const generation = ++workspaceLoadGeneration; const ephemeral = await isEphemeralBackend(); + assertCurrentWorkspaceLoad(generation); try { const { data } = await apiRequest(getUrls().SESSION_LOAD, { method: 'POST', @@ -132,7 +158,9 @@ export async function loadWorkspace(id: string): Promise<{ state: Record void; } -export const ConnectorFormCard: React.FC = ({ messageId, prompt, defaultExpanded = true, variant = 'card' }) => { +export const ConnectorFormCard: React.FC = ({ messageId, prompt, defaultExpanded = true, variant = 'card', onResolved }) => { const theme = useTheme(); const { t } = useTranslation(); const dispatch = useDispatch(); @@ -202,12 +208,16 @@ export const ConnectorFormCard: React.FC = ({ messageId, // Connection succeeded even if the follow-up list fetch fails. } } - dispatch(dfActions.resolveConnectorForm({ - messageId, - status: 'connected', + const resolution = { + status: 'connected' as const, connectorId: cid ?? undefined, connectionName: resolvedName, - })); + }; + if (onResolved) { + onResolved(resolution); + } else { + dispatch(dfActions.resolveConnectorForm({ messageId, ...resolution })); + } // Make the new source show up in the data-source sidebar. dispatch(dfActions.requestConnectorRefresh()); dispatch(dfActions.addMessages({ @@ -221,24 +231,26 @@ export const ConnectorFormCard: React.FC = ({ messageId, // source and give a comprehensive overview). Sent as a hidden trigger — // it is part of the agent's context but never shown as a user bubble; // the agent's reply is visible (design 38 §7). - dispatch(dfActions.setDataLoadingChatPending({ - text: t('chatConnector.connectedAgentTrigger', { - name: resolvedName, - type: sourceType, - defaultValue: - 'I just connected a new data source "{{name}}" (type: {{type}}). ' - + 'Browse it and give me a concise but comprehensive overview: what ' - + 'databases/schemas it contains, the notable tables in each (with a ' - + 'one-line hint of what they hold and their approximate size where ' - + 'known), and any groupings or themes you notice. Then suggest a ' - + 'couple of good starting points and ask what I would like to ' - + 'explore or load.', - }), - images: [], - attachments: [], - hidden: true, - })); - }, [messageId, meta, sourceType, dispatch, t]); + if (!onResolved) { + dispatch(dfActions.setDataLoadingChatPending({ + text: t('chatConnector.connectedAgentTrigger', { + name: resolvedName, + type: sourceType, + defaultValue: + 'I just connected a new data source "{{name}}" (type: {{type}}). ' + + 'Browse it and give me a concise but comprehensive overview: what ' + + 'databases/schemas it contains, the notable tables in each (with a ' + + 'one-line hint of what they hold and their approximate size where ' + + 'known), and any groupings or themes you notice. Then suggest a ' + + 'couple of good starting points and ask what I would like to ' + + 'explore or load.', + }), + images: [], + attachments: [], + hidden: true, + })); + } + }, [messageId, meta, sourceType, dispatch, t, onResolved]); const cardSx = { mt: 1, @@ -268,7 +280,8 @@ export const ConnectorFormCard: React.FC = ({ messageId, authMode={meta.auth_mode} authPaths={meta.auth_paths} compact - hideInstructions + comfortableSpacing={isBare} + hideInstructions={!isBare} onImport={() => {}} onFinish={(status, message) => { dispatch(dfActions.addMessages({ diff --git a/src/i18n/locales/en/common.json b/src/i18n/locales/en/common.json index 5a30ee26..457334ff 100644 --- a/src/i18n/locales/en/common.json +++ b/src/i18n/locales/en/common.json @@ -479,7 +479,8 @@ "hideEarlierTurns": "Hide earlier turns", "working": "working...", "waitingForClarification": "waiting for clarification...", - "emptySession": "No data yet \u2014 ask the agent below to load some, and it will show up here.", + "emptySessionTitle": "No data here yet", + "emptySession": "Ask the agent below to load some. It will appear here when it is ready.", "startingRun": "Working on your request\u2026", "rename": "Rename", "refreshSettings": "Refresh settings", diff --git a/src/i18n/locales/zh/common.json b/src/i18n/locales/zh/common.json index 324c38f3..1d20a31d 100644 --- a/src/i18n/locales/zh/common.json +++ b/src/i18n/locales/zh/common.json @@ -479,7 +479,8 @@ "hideEarlierTurns": "收起更早的对话", "working": "处理中...", "waitingForClarification": "等待澄清...", - "emptySession": "还没有数据 \u2014 在下方让智能体加载数据，结果会显示在这里。", + "emptySessionTitle": "这里还没有数据", + "emptySession": "请在下方让智能体加载数据。准备好后，数据会显示在这里。", "startingRun": "正在处理您的请求\u2026", "rename": "重命名", "refreshSettings": "刷新设置", diff --git a/src/views/DBTableManager.tsx b/src/views/DBTableManager.tsx index d527d19b..81a4f106 100644 --- a/src/views/DBTableManager.tsx +++ b/src/views/DBTableManager.tsx @@ -9,6 +9,7 @@ import { CircularProgress, Checkbox, FormControlLabel, + Switch, IconButton, Tooltip, Accordion, @@ -111,6 +112,8 @@ export const DataLoaderForm: React.FC<{ /** When true, lay parameters out in a single column and tighten spacing * so the form fits inside a chat card (design 38). */ compact?: boolean, + /** Retain compact control sizing while adding canvas-appropriate spacing. */ + comfortableSpacing?: boolean, /** When true, keep setup instructions collapsed initially in compact forms. */ hideInstructions?: boolean, /** One-time seed for sensitive fields (passwords/tokens) the user handed to @@ -120,7 +123,7 @@ export const DataLoaderForm: React.FC<{ /** Hands the user to the data agent chat with a seeded question when they * get stuck on setup. Omitted inside the chat card itself. */ onAskAgent?: (prompt: string) => void, -}> = ({dataLoaderType, loaderType, paramDefs, authInstructions, connectorId, autoConnect, ssoAutoConnect, delegatedLogin, authMode, authPaths = [], formTitle, onImport, onFinish, onConnected, onBeforeConnect, hasStoredCredentials, compact = false, hideInstructions = false, initialSensitiveParams, onAskAgent}) => { +}> = ({dataLoaderType, loaderType, paramDefs, authInstructions, connectorId, autoConnect, ssoAutoConnect, delegatedLogin, authMode, authPaths = [], formTitle, onImport, onFinish, onConnected, onBeforeConnect, hasStoredCredentials, compact = false, comfortableSpacing = false, hideInstructions = false, initialSensitiveParams, onAskAgent}) => { const { t } = useTranslation(); const dispatch = useDispatch(); const loaderTypeKey = loaderType || dataLoaderType; @@ -296,6 +299,39 @@ export const DataLoaderForm: React.FC<{ })); } }, [dataLoaderType, dispatch, sensitiveParamNames]); + const renderBooleanParam = useCallback((paramDef: typeof paramDefs[number]) => { + const rawValue = params[paramDef.name] ?? paramDef.default ?? false; + const checked = String(rawValue).toLowerCase() === 'true'; + return ( + { + dispatch(dfActions.updateDataLoaderConnectParam({ + dataLoaderType, + paramName: paramDef.name, + paramValue: String(nextChecked), + })); + }} + inputProps={{ 'aria-label': paramDef.name }} + /> + )} + label={checked + ? t('common.on', { defaultValue: 'On' }) + : t('common.off', { defaultValue: 'Off' })} + sx={{ + m: 0, + minHeight: 40, + '& .MuiFormControlLabel-label': { + fontSize: compact ? '0.75rem' : '0.8125rem', + lineHeight: 1.4, + }, + }} + /> + ); + }, [compact, dataLoaderType, dispatch, paramDefs, params, t]); const selectAuthPath = useCallback((pathId: string) => { const selectedPath = authPaths.find(path => path.id === pathId); if (!selectedPath) return; @@ -569,16 +605,24 @@ export const DataLoaderForm: React.FC<{ fontSize: bodyFontSize, fontWeight: 500, lineHeight: 1.4, - mb: compact ? 0.25 : 0.5, + mb: compact && !comfortableSpacing ? 0.25 : 0.5, }; - const fieldGap = compact ? 1 : 1.5; - const sectionGap = compact ? 1.25 : 2; + const fieldGap = compact ? (comfortableSpacing ? 1.75 : 1) : 1.5; + const sectionGap = compact ? (comfortableSpacing ? 2.25 : 1.25) : 2; // Inputs otherwise keep MUI's 14px, which is the one size that breaks the scale. const fieldSx = { '& .MuiInputBase-root': { fontSize: bodyFontSize }, ...(compact ? { + '& .MuiOutlinedInput-root': { height: 32 }, '& .MuiOutlinedInput-input': { paddingTop: '5.5px', paddingBottom: '5.5px' }, - '& .MuiAutocomplete-inputRoot': { paddingTop: '1px', paddingBottom: '1px' }, + '& .MuiAutocomplete-inputRoot': { + paddingTop: '0 !important', + paddingBottom: '0 !important', + }, + '& .MuiAutocomplete-inputRoot .MuiAutocomplete-input': { + paddingTop: '5.5px !important', + paddingBottom: '5.5px !important', + }, '& .MuiFormHelperText-root': { fontSize: '0.6875rem', marginTop: '2px' }, } : {}), }; @@ -701,7 +745,7 @@ export const DataLoaderForm: React.FC<{ ? { xs: 'minmax(0, 1fr)', md: 'minmax(440px, 1.55fr) minmax(280px, 1fr)' } : 'minmax(0, 1fr)', columnGap: { xs: 0, md: 4 }, - rowGap: compact ? 2 : 3, + rowGap: compact ? (comfortableSpacing ? 2.5 : 2) : 3, alignItems: 'start', width: '100%', maxWidth: compact ? '100%' : (showSideGuide ? 940 : 520), @@ -732,7 +776,7 @@ export const DataLoaderForm: React.FC<{ {paramDef.name}{paramDef.required ? ' *' : ''} - updateParamDraft(paramDef.name, value)} onCommit={(value) => commitParamDraft(paramDef.name, value)} - /> + />} ))} @@ -756,12 +800,25 @@ export const DataLoaderForm: React.FC<{ loaderTypeKey === 'kusto' && name === 'kusto_cluster'; const isKustoDatabase = (name: string) => loaderTypeKey === 'kusto' && name === 'kusto_database'; - const renderFieldRow = (paramDef: typeof tierParams[number], input: React.ReactNode) => ( + const renderFieldRow = (paramDef: typeof tierParams[number], input: React.ReactNode, action?: React.ReactNode) => ( {paramDef.name}{paramDef.required ? ' *' : ''} - {input} + + {input} + {action && ( + + {action} + + )} + ); return ( @@ -769,11 +826,11 @@ export const DataLoaderForm: React.FC<{ {tierParams.map((paramDef) => ( isKustoCluster(paramDef.name) ? ( renderFieldRow(paramDef, - { dispatch(dfActions.updateDataLoaderConnectParam({ @@ -803,18 +860,18 @@ export const DataLoaderForm: React.FC<{ placeholder={getParamHelp(paramDef) || getParamPlaceholder(paramDef)} /> )} - /> + />, - ) ) : isKustoDatabase(paramDef.name) ? ( renderFieldRow(paramDef, { setDatabaseMenuOpen(true); @@ -865,11 +922,14 @@ export const DataLoaderForm: React.FC<{ )} /> ) + ) : paramDef.type === 'boolean' || paramDef.type === 'bool' ? ( + renderFieldRow(paramDef, renderBooleanParam(paramDef)) ) : paramDef.options ? ( renderFieldRow(paramDef, { dispatch(dfActions.updateDataLoaderConnectParam({ diff --git a/src/views/DataFormulator.tsx b/src/views/DataFormulator.tsx index e1711f0c..1e4f969b 100644 --- a/src/views/DataFormulator.tsx +++ b/src/views/DataFormulator.tsx @@ -25,7 +25,6 @@ import { useTheme, useMediaQuery, alpha, - CircularProgress, Backdrop, Link, Select, @@ -39,6 +38,7 @@ import { borderColor, radius, transition } from '../app/tokens'; import { VisualizationViewFC } from './VisualizationView'; +import { AnvilLoader } from '../components/AnvilLoader'; import { DndProvider } from 'react-dnd' import { HTML5Backend } from 'react-dnd-html5-backend' @@ -66,7 +66,7 @@ import { useDataRefresh, useDerivedTableRefresh } from '../app/useDataRefresh'; import { useTranslation } from 'react-i18next'; import { fetchWithIdentity, getUrls, CONNECTOR_URLS } from '../app/utils'; import { apiRequest } from '../app/apiClient'; -import { listWorkspaces, loadWorkspace, deleteWorkspace, exportWorkspace, importWorkspace, onWorkspaceListChanged, updateWorkspaceMeta } from '../app/workspaceService'; +import { listWorkspaces, loadWorkspace, deleteWorkspace, exportWorkspace, importWorkspace, onWorkspaceListChanged, updateWorkspaceMeta, WorkspaceLoadSupersededError } from '../app/workspaceService'; import type { WorkspaceSummary } from '../app/workspaceService'; import { AppDispatch, store } from '../app/store'; import { generateUUID } from '../app/identity'; @@ -190,14 +190,21 @@ export const DataFormulatorFC = ({ }) => { dispatch(dfActions.setSessionLoading({ loading: true, label: t('workspace.openingWorkspace') })); try { const result = await loadWorkspace(name); - if (result && Object.keys(result.state).length > 0) { + if (result) { const displayName = metaDisplayName || result.displayName; dispatch(dfActions.loadState({ ...result.state, activeWorkspace: { id: name, displayName, readOnly: result.readOnly } })); } else { - dispatch(dfActions.setActiveWorkspace({ id: name, displayName: metaDisplayName || 'Untitled Session' })); + dispatch(dfActions.addMessages({ + timestamp: Date.now(), type: 'error', component: 'workspace', + value: t('workspace.failedToOpenWorkspace'), + })); } - } catch { - dispatch(dfActions.setActiveWorkspace({ id: name, displayName: metaDisplayName || 'Untitled Session' })); + } catch (error) { + if (error instanceof WorkspaceLoadSupersededError) return; + dispatch(dfActions.addMessages({ + timestamp: Date.now(), type: 'error', component: 'workspace', + value: t('workspace.failedToOpenWorkspace'), + })); } dispatch(dfActions.setSessionLoading({ loading: false })); }, [dispatch]); @@ -1146,18 +1153,21 @@ export const DataFormulatorFC = ({ }) => { gap: 2, }} > - - - {sessionLoadingLabel || t('session.loadingSessions')} - - + dispatch(dfActions.setSessionLoading({ loading: false }))} + sx={{ minWidth: 0, px: 0.5, textTransform: 'none', color: 'text.secondary' }} + > + {t('app.cancel')} + + )} + sx={{ width: '100%' }} + /> {selectedModelId == undefined && ( 0) { + if (result) { const displayName = metaDisplayName || result.displayName; dispatch(dfActions.loadState({ ...result.state, activeWorkspace: { id: sessionId, displayName, readOnly: result.readOnly } })); } else { - dispatch(dfActions.setActiveWorkspace({ id: sessionId, displayName: metaDisplayName || 'Untitled Session' })); + dispatch(dfActions.addMessages({ + timestamp: Date.now(), type: 'error', component: 'workspace', + value: t('workspace.failedToOpenWorkspace'), + })); } - } catch { - dispatch(dfActions.setActiveWorkspace({ id: sessionId, displayName: metaDisplayName || 'Untitled Session' })); + } catch (error) { + if (error instanceof WorkspaceLoadSupersededError) return; + dispatch(dfActions.addMessages({ + timestamp: Date.now(), type: 'error', component: 'workspace', + value: t('workspace.failedToOpenWorkspace'), + })); } dispatch(dfActions.setSessionLoading({ loading: false })); }, [dispatch]); @@ -2241,12 +2248,7 @@ const DataSourceSidebarPanel: React.FC<{ }}> { - const now = new Date(); - const date = `${now.getFullYear()}${String(now.getMonth() + 1).padStart(2, '0')}${String(now.getDate()).padStart(2, '0')}`; - const time = `${String(now.getHours()).padStart(2, '0')}${String(now.getMinutes()).padStart(2, '0')}${String(now.getSeconds()).padStart(2, '0')}`; - const short = generateUUID().slice(0, 4); - const wsId = `session_${date}_${time}_${short}`; - dispatch(dfActions.loadState({ tables: [], charts: [], draftNodes: [], conceptShelfItems: [], activeWorkspace: { id: wsId, displayName: 'Untitled Session' } })); + dispatch(dfActions.resetState()); }} sx={{ display: 'flex', diff --git a/src/views/DataThread.tsx b/src/views/DataThread.tsx index 89f9d318..58746952 100644 --- a/src/views/DataThread.tsx +++ b/src/views/DataThread.tsx @@ -33,6 +33,7 @@ import type { CatalogTableDragItem } from '../components/DndTypes'; import { ScrollFadeEdge, useScrollFade } from '../components/ScrollFade'; import { loadTable } from '../app/tableThunks'; import { AppDispatch } from '../app/store'; +import dfLogo from '../assets/df-logo.svg'; import DeleteIcon from '@mui/icons-material/Delete'; import PersonIcon from '@mui/icons-material/Person'; @@ -576,7 +577,7 @@ let SingleThreadGroupView: FC<{ if (!turn) return undefined; // A data-operation turn replaces the canvas, so no table is the // canvas subject; other turns leave the chart/table canvas up. - if (turn.dataOperation) return undefined; + if (turn.dataOperation || turn.form) return undefined; if (turn.sourceChartId) { const chart = charts.find(c => c.id === turn.sourceChartId); if (chart?.tableRef) return chart.tableRef; @@ -1259,7 +1260,12 @@ let SingleThreadGroupView: FC<{ const buildTextTurnTimelineItem = (turn: TextTurn, highlighted: boolean, showPrompt: boolean) => { const isFocused = focusedId?.type === 'text' && focusedId.textId === turn.id; const rowHL = highlighted || isFocused; - const preview = (turn.content || '') + const formStatus = turn.form?.kind === 'connector' + ? (turn.form.connector.status === 'connected' + ? `Connected to ${turn.form.connector.connectionName || turn.form.connector.sourceType}` + : turn.form.title) + : undefined; + const preview = (formStatus || turn.content || '') .replace(/[#*`>|]/g, ' ').replace(/\s+/g, ' ').trim(); // Once answered, the turn is history: it drops its card chrome and reads // as muted agent prose so the thread foregrounds what it produced. @@ -1275,11 +1281,14 @@ let SingleThreadGroupView: FC<{ || draftNodes.some(draft => draft.parentNodeId === turn.id); // Every turn is an agent remark, so its glyph sits ON the spine like any // other entry, while the card keeps the exchange readable as one unit. - const awaitingAnswer = !turn.answered && (turn.options?.length ?? 0) > 0; - const gutterIcon = getEntryGutterIcon( - { from: 'data-agent', to: 'user', role: turn.textKind, content: '' }, - rowHL ? theme.palette.text.secondary : 'rgba(0,0,0,0.15)', - ); + const awaitingAnswer = !turn.answered && ((turn.options?.length ?? 0) > 0 || !!turn.form); + const iconColor = rowHL ? theme.palette.text.secondary : 'rgba(0,0,0,0.15)'; + const gutterIcon = turn.form + ? + : getEntryGutterIcon( + { from: 'data-agent', to: 'user', role: turn.textKind, content: '' }, + iconColor, + ); const card = ( + {workPending ? ( @@ -3320,9 +3339,23 @@ export const DataThread: FC<{sx?: SxProps, centered?: boolean, denseColumns?: bo ) : ( - - {t('dataThread.emptySession')} - + + )} ); diff --git a/src/views/SimpleChartRecBox.tsx b/src/views/SimpleChartRecBox.tsx index d39d5311..1072394a 100644 --- a/src/views/SimpleChartRecBox.tsx +++ b/src/views/SimpleChartRecBox.tsx @@ -1040,6 +1040,54 @@ export const SimpleChartRecBox: FC<{ onInputFocus?: () => void }> = function ({ }; const processStreamingResult = async (result: any) => { + if (result.type === "interact" && result.form) { + if (result.form.kind !== 'connector') { + throw new Error(`Unsupported form artifact kind: ${String(result.form.kind)}`); + } + const sourceType = String(result.form.connector?.source_type || '').trim(); + if (!sourceType) { + throw new Error('Connector form artifact requires source_type'); + } + const turnId = `textTurn_${actionId}_${String(Date.now())}`; + const firstEntry = currentDraftInteraction[0]; + dispatch(dfActions.addTextTurn({ + kind: 'text', + id: turnId, + displayId: turnId, + textKind: 'explain', + content: String(result.form.response || result.form.title || `Connect to ${sourceType}`), + ...(!runIsContinuationRef.current && firstEntry?.role === 'prompt' + ? { prompt: firstEntry.displayContent || firstEntry.content } + : {}), + form: { + kind: 'connector', + title: String(result.form.title || `Connect to ${sourceType}`), + connector: { + sourceType, + prefilled: result.form.connector?.prefilled || {}, + status: 'pending', + }, + }, + parentNodeId: runLastNodeRef.current || askedFromTable || focusedTableId || ROOTLESS_THREAD_ID, + ...(runSourceChartIdRef.current ? { sourceChartId: runSourceChartIdRef.current } : {}), + actionId, + createdAt: Date.now(), + })); + runLastNodeRef.current = turnId; + dispatch(dfActions.setFocused({ type: 'text', textId: turnId })); + if (currentDraftId) { + dispatch(dfActions.removeDraftNode(currentDraftId)); + currentDraftId = null; + } + setIsChatFormulating(false); + agentAbortRef.current = null; + clearTimeout(timeoutId); + setChatPrompt(""); + setAttachedImages([]); + setAttachedFiles([]); + isCompleted = true; + return; + } // ── interact: the unified agent's clarify/explain pause ── // Alias to the legacy clarify path (same questions[] shape + the // backend now stamps trajectory/completed_step_count for resume). @@ -1949,17 +1997,26 @@ export const SimpleChartRecBox: FC<{ onInputFocus?: () => void }> = function ({ if (focusedTextTurn) { setClarifyAnswers({}); textTurnSubmittedRef.current = null; } }, [focusedTextTurn?.id]); - // Close a focused text turn → hand focus to whatever the canvas is already - // showing for it (its nearest artifact), keeping the node in the thread. - // Deriving it here separately let the two drift: the canvas resolved to the - // chart while this landed on the table, or on nothing at all. + // Close a focused turn without deleting it: return to its source chart or + // nearest parent artifact. Root-level turns explicitly clear focus so the + // canvas returns to its empty state. const closeTextTurn = useCallback(() => { - if (canvasTarget && canvasTarget.type !== 'text') { - dispatch(dfActions.setFocused(canvasTarget)); + if (!focusedTextTurn) return; + if (focusedTextTurn.sourceChartId && charts.some(chart => chart.id === focusedTextTurn.sourceChartId)) { + dispatch(dfActions.setFocused({ type: 'chart', chartId: focusedTextTurn.sourceChartId })); return; } - switchFocusToPreviousChart(); - }, [canvasTarget, dispatch, switchFocusToPreviousChart]); + const parentTableId = resolveNodeTable(focusedTextTurn.parentNodeId, textTurns, tables); + if (parentTableId) { + const tableCharts = charts.filter(chart => chart.tableRef === parentTableId); + const nearestChart = tableCharts[tableCharts.length - 1]; + dispatch(dfActions.setFocused(nearestChart + ? { type: 'chart', chartId: nearestChart.id } + : { type: 'table', tableId: parentTableId })); + return; + } + dispatch(dfActions.setFocused(undefined)); + }, [charts, dispatch, focusedTextTurn, tables, textTurns]); // Answer a focused clarify turn. Resumes the run iff the turn carries the // backend's opaque resume token (§12); otherwise a fresh turn. The turn stays @@ -2548,6 +2605,10 @@ export const SimpleChartRecBox: FC<{ onInputFocus?: () => void }> = function ({ // so the pending question isn't lost. Clicks on a thread card // switch focus normally, so ignore those here. if (!focusedTextTurn) return; + // A form turn owns the canvas. Interacting with any field or + // control there must keep both the form and its feedback open; + // only either surface's explicit close button dismisses it. + if (focusedTextTurn.form) return; const isActiveClarify = focusedTextTurn.textKind === 'clarify' && !focusedTextTurn.answered && (focusedTextTurn.options?.length ?? 0) > 0; diff --git a/src/views/VisualizationView.tsx b/src/views/VisualizationView.tsx index 44ad9398..98f20438 100644 --- a/src/views/VisualizationView.tsx +++ b/src/views/VisualizationView.tsx @@ -54,7 +54,8 @@ import { displayRowsCache } from '../app/displayRowsCache'; import { buildEmbeddedDataForChart, applyVariantConfigUI } from '../app/restyle'; import { apiRequest } from '../app/apiClient'; import embed from 'vega-embed'; -import { Chart, EncodingItem, EncodingMap, FieldItem, FieldSemanticsInfo, computeInsightKey } from '../components/ComponentType'; +import { Chart, EncodingItem, EncodingMap, FieldItem, FieldSemanticsInfo, FormArtifact, TextTurn, computeInsightKey } from '../components/ComponentType'; +import { ConnectorFormCard } from '../components/ConnectorFormCard'; import TerminalIcon from '@mui/icons-material/Terminal'; import QuestionAnswerIcon from '@mui/icons-material/QuestionAnswer'; @@ -86,7 +87,7 @@ import { EncodingShelfCard } from './EncodingShelfCard'; import { ChartQuickConfig } from './ChartQuickConfig'; import { ChartVariantStrip } from './ChartVariantStrip'; import { CustomReactTable } from './ReactTable'; -import { InsightIcon } from '../icons'; +import { getConnectorIcon, InsightIcon } from '../icons'; import { FreeDataViewFC } from './DataView'; import { formatCellValue } from './ViewUtils'; @@ -296,6 +297,50 @@ const DataOperationCanvas: FC<{ operation: DataOperation }> = ({ operation }) => ); }; +const FormArtifactCanvas: FC<{ turn: TextTurn; form: FormArtifact }> = ({ turn, form }) => { + const dispatch = useDispatch(); + + switch (form.kind) { + case 'connector': + return ( + + + + {getConnectorIcon(form.connector.sourceType, { + sx: { fontSize: iconVar.lg, color: 'text.secondary', flexShrink: 0 }, + })} + + {form.title} + + + { + dispatch(dfActions.updateTextTurn({ + id: turn.id, + answered: true, + answer: `Connected to ${resolution.connectionName}`, + form: { + kind: 'connector', + title: form.title, + connector: { + sourceType: form.connector.sourceType, + status: resolution.status, + connectorId: resolution.connectorId, + connectionName: resolution.connectionName, + }, + }, + })); + }} + /> + + + ); + } +}; + // Re-export shared utilities from ChartUtils (canonical location) import { generateChartSkeleton, getDataTable, checkChartAvailability } from './ChartUtils'; export { generateChartSkeleton, getDataTable, checkChartAvailability }; @@ -1675,15 +1720,17 @@ export const VisualizationViewFC: FC = function VisualizationView const { t } = useTranslation(); let allCharts = useSelector(dfSelectors.getAllCharts); - const rawFocusedId = useSelector((state: DataFormulatorState) => state.focusedId); const textTurns = useSelector((state: DataFormulatorState) => state.textTurns); - const focusedOperationTurn = rawFocusedId?.type === 'text' - ? textTurns.find(turn => turn.id === rawFocusedId.textId && turn.dataOperation) - : undefined; // Resolve the canvas target: a focused text turn (clarify/explain) is // non-canvas-owning — the canvas keeps showing its source chart, or the // source table when no chart exists (design-docs/41). let focusedId = useSelector(dfSelectors.selectCanvasTarget); + const focusedOperationTurn = focusedId?.type === 'text' + ? textTurns.find(turn => turn.id === focusedId.textId && turn.dataOperation) + : undefined; + const focusedFormTurn = focusedId?.type === 'text' + ? textTurns.find(turn => turn.id === focusedId.textId && turn.form) + : undefined; let focusedChartId = focusedId?.type === 'chart' ? focusedId.chartId : undefined; let focusedTableId = React.useMemo(() => { if (!focusedId) return undefined; @@ -1707,6 +1754,9 @@ export const VisualizationViewFC: FC = function VisualizationView if (focusedOperationTurn?.dataOperation) { return ; } + if (focusedFormTurn?.form) { + return ; + } let focusedChart = allCharts.find(c => c.id == focusedChartId) as Chart; let synthesisRunning = focusedChartId ? chartSynthesisInProgress.includes(focusedChartId) : false; diff --git a/tests/backend/agents/test_analyst_connector_skill.py b/tests/backend/agents/test_analyst_connector_skill.py new file mode 100644 index 00000000..d4122035 --- /dev/null +++ b/tests/backend/agents/test_analyst_connector_skill.py @@ -0,0 +1,99 @@ +from __future__ import annotations + +import json +from types import SimpleNamespace +from unittest.mock import patch + +import pytest + +from data_formulator.analyst.skills.base import SkillContext +from data_formulator.analyst.skills.data_loading.skill import DataLoadingSkill + +pytestmark = pytest.mark.backend + + +class _Loader: + DISPLAY_NAME = "PostgreSQL" + DESCRIPTION = "Query PostgreSQL databases." + + @staticmethod + def auth_mode(): + return "credentials" + + @staticmethod + def auth_paths(): + return [] + + @staticmethod + def auth_instructions(): + return "Enter database credentials." + + @staticmethod + def list_params(): + return [ + {"name": "host", "required": True, "tier": "connection"}, + {"name": "password", "required": True, "tier": "auth", "sensitive": True}, + ] + + +def _context() -> SkillContext: + return SkillContext(client=None, workspace=SimpleNamespace(), payload={}) + + +def test_list_and_describe_connectors() -> None: + skill = DataLoadingSkill() + ctx = _context() + with ( + patch.dict("data_formulator.data_loader.DATA_LOADERS", {"postgresql": _Loader}, clear=True), + patch.dict("data_formulator.data_loader.DISABLED_LOADERS", {}, clear=True), + ): + listed = json.loads(skill.handle_tool("list_connectors", {}, ctx).text) + described = json.loads(skill.handle_tool( + "describe_connector", {"source_type": "postgresql"}, ctx, + ).text) + + assert listed["connectors"][0]["type"] == "postgresql" + assert "call propose_connection now" in listed["next_action"] + assert described["params"][1]["sensitive"] is True + assert "Call propose_connection now" in described["next_action"] + + +def test_propose_connection_requires_listing_first() -> None: + skill = DataLoadingSkill() + ctx = _context() + with patch.dict("data_formulator.data_loader.DATA_LOADERS", {"postgresql": _Loader}, clear=True): + events = list(skill.handle_action( + "propose_connection", {"source_type": "postgresql"}, ctx, + )) + + assert events[0]["type"] == "error" + assert "list_connectors" in events[0]["message"] + + +def test_propose_connection_emits_prefilled_canvas_form_without_echo() -> None: + skill = DataLoadingSkill() + ctx = _context() + with ( + patch.dict("data_formulator.data_loader.DATA_LOADERS", {"postgresql": _Loader}, clear=True), + patch.dict("data_formulator.data_loader.DISABLED_LOADERS", {}, clear=True), + ): + skill.handle_tool("list_connectors", {}, ctx) + events = list(skill.handle_action("propose_connection", { + "source_type": "postgresql", + "prefilled": {"host": "db.example.com", "password": "secret", "empty": ""}, + }, ctx)) + + assert events == [{ + "type": "interact", + "thought": "", + "form": { + "kind": "connector", + "title": "Connect to PostgreSQL", + "response": "Complete the PostgreSQL connection form to add this data source.", + "connector": { + "source_type": "postgresql", + "prefilled": {"host": "db.example.com", "password": "secret"}, + }, + }, + }] + assert "secret" not in json.dumps({key: value for key, value in events[0].items() if key != "form"}) \ No newline at end of file diff --git a/tests/backend/agents/test_data_loading_discovery_tools.py b/tests/backend/agents/test_data_loading_discovery_tools.py index f925dedb..b0afefba 100644 --- a/tests/backend/agents/test_data_loading_discovery_tools.py +++ b/tests/backend/agents/test_data_loading_discovery_tools.py @@ -7,6 +7,7 @@ Covers the navigation surface introduced by design-docs/32: from __future__ import annotations from pathlib import Path +from types import SimpleNamespace from unittest.mock import patch import pytest @@ -61,7 +62,47 @@ _SAMPLE_TABLES = [ ] +class _NoAuthLoader: + @classmethod + def list_params(cls): + return [] + + def __init__(self, _params): + pass + + def test_connection(self): + return True + + def sync_catalog_metadata(self): + return _SAMPLE_TABLES + + +def _zero_auth_connector(): + return SimpleNamespace(_loader_class=_NoAuthLoader, _default_params={}) + + class TestListData: + def test_bootstraps_uncached_zero_auth_connector(self, tmp_path: Path) -> None: + agent = DataLoadingAgent(client=None, workspace=_FakeWorkspace(tmp_path)) + + with ( + patch("data_formulator.data_connector._ADMIN_CONNECTOR_IDS", {"sample_datasets"}), + patch.dict( + "data_formulator.data_connector.DATA_CONNECTORS", + {"sample_datasets": _zero_auth_connector()}, + clear=True, + ), + ): + result = agent._tool_list_data({}) + + assert result["sources"] == [ + { + "source_id": "sample_datasets", + "table_count": 3, + "is_hierarchical": True, + } + ] + def test_no_args_returns_sources_summary(self, tmp_path: Path) -> None: save_catalog(tmp_path, "pg_prod", _SAMPLE_TABLES) save_catalog(tmp_path, "flat_src", [{"name": "t1", "table_key": "k1", "metadata": {}}]) @@ -127,6 +168,21 @@ class TestListData: # ------------------------------------------------------------------ class TestFindData: + def test_bootstraps_uncached_zero_auth_connector(self, tmp_path: Path) -> None: + agent = DataLoadingAgent(client=None, workspace=_FakeWorkspace(tmp_path)) + + with ( + patch("data_formulator.data_connector._ADMIN_CONNECTOR_IDS", {"sample_datasets"}), + patch.dict( + "data_formulator.data_connector.DATA_CONNECTORS", + {"sample_datasets": _zero_auth_connector()}, + clear=True, + ), + ): + result = agent._tool_find_data({"query": "customers", "scope": "connected"}) + + assert [item["name"] for item in result["results"]] == ["customers"] + def test_empty_query_returns_error(self) -> None: agent = DataLoadingAgent(client=None, workspace=_FakeWorkspace()) result = agent._tool_find_data({"query": ""}) diff --git a/tests/backend/data/test_catalog_cache.py b/tests/backend/data/test_catalog_cache.py index 7efac880..bb15ca92 100644 --- a/tests/backend/data/test_catalog_cache.py +++ b/tests/backend/data/test_catalog_cache.py @@ -12,6 +12,7 @@ Covers: from __future__ import annotations import json +from datetime import datetime, timedelta, timezone from pathlib import Path from typing import Any from unittest.mock import MagicMock, patch @@ -24,6 +25,8 @@ from data_formulator.datalake.catalog_cache import ( delete_catalog, list_cached_sources, load_catalog, + load_catalog_snapshot, + record_catalog_refresh_failure, save_catalog, search_catalog_cache, _search_python, @@ -106,6 +109,95 @@ class TestSaveLoadCatalog: assert loaded is not None assert len(loaded) == 2 + def test_legacy_synced_at_populates_both_freshness_clocks(self, tmp_path: Path) -> None: + cache_dir = tmp_path / "catalog_cache" + cache_dir.mkdir() + synced_at = "2026-01-01T00:00:00Z" + (cache_dir / "legacy.json").write_text(json.dumps({ + "source_id": "legacy", + "synced_at": synced_at, + "tables": SAMPLE_TABLES, + }), encoding="utf-8") + + snapshot = load_catalog_snapshot( + tmp_path, + "legacy", + listing_ttl_seconds=60, + metadata_ttl_seconds=60, + now=datetime(2026, 1, 1, 0, 0, 30, tzinfo=timezone.utc), + ) + + assert snapshot is not None + assert snapshot.listing_refreshed_at == synced_at + assert snapshot.metadata_refreshed_at == synced_at + assert snapshot.listing_freshness == "fresh" + assert snapshot.metadata_freshness == "fresh" + + def test_listing_and_metadata_have_independent_freshness(self, tmp_path: Path) -> None: + save_catalog(tmp_path, "src1", SAMPLE_TABLES) + path = tmp_path / "catalog_cache" / "src1.json" + data = json.loads(path.read_text(encoding="utf-8")) + now = datetime(2026, 1, 2, tzinfo=timezone.utc) + data["listing_refreshed_at"] = (now - timedelta(seconds=30)).isoformat() + data["metadata_refreshed_at"] = (now - timedelta(hours=2)).isoformat() + path.write_text(json.dumps(data), encoding="utf-8") + + snapshot = load_catalog_snapshot( + tmp_path, + "src1", + listing_ttl_seconds=60, + metadata_ttl_seconds=60, + now=now, + ) + + assert snapshot is not None + assert snapshot.listing_freshness == "fresh" + assert snapshot.metadata_freshness == "stale" + + def test_listing_refresh_preserves_enriched_metadata(self, tmp_path: Path) -> None: + rich = [{ + "name": "public.orders", + "path": ["public", "orders"], + "table_key": "orders-key", + "metadata": {"description": "Orders", "columns": [{"name": "id"}]}, + }] + save_catalog(tmp_path, "src1", rich) + save_catalog(tmp_path, "src1", [{ + "name": "public.orders", + "path": ["public", "orders"], + "table_key": "orders-key", + "metadata": {"row_count": 10}, + }], refresh_kind="listing") + + loaded = load_catalog(tmp_path, "src1") + assert loaded is not None + assert loaded[0]["metadata"] == { + "description": "Orders", + "columns": [{"name": "id"}], + "row_count": 10, + } + + def test_refresh_failure_preserves_last_good_tables(self, tmp_path: Path) -> None: + save_catalog(tmp_path, "src1", SAMPLE_TABLES) + record_catalog_refresh_failure(tmp_path, "src1", "connection timed out") + + assert load_catalog(tmp_path, "src1") == SAMPLE_TABLES + snapshot = load_catalog_snapshot( + tmp_path, "src1", listing_ttl_seconds=60, metadata_ttl_seconds=60, + ) + assert snapshot is not None + assert snapshot.refresh_status == "failed" + assert snapshot.last_refresh_error == "connection timed out" + + def test_atomic_replace_failure_preserves_existing_catalog(self, tmp_path: Path) -> None: + save_catalog(tmp_path, "src1", [{"name": "old_table"}]) + + with patch("data_formulator.datalake.catalog_cache.os.replace", side_effect=OSError): + save_catalog(tmp_path, "src1", [{"name": "new_table"}]) + + assert load_catalog(tmp_path, "src1") == [{"name": "old_table"}] + assert list((tmp_path / "catalog_cache").glob("*.tmp")) == [] + # ================================================================== # Tests: delete diff --git a/tests/backend/data/test_catalog_refresh.py b/tests/backend/data/test_catalog_refresh.py new file mode 100644 index 00000000..b334676f --- /dev/null +++ b/tests/backend/data/test_catalog_refresh.py @@ -0,0 +1,109 @@ +from __future__ import annotations + +import json +from pathlib import Path +from unittest.mock import MagicMock, patch + +import pytest + +from data_formulator.datalake.catalog_cache import ( + load_catalog, + record_catalog_refresh_failure, + save_catalog, +) +from data_formulator.datalake.catalog_refresh import ensure_catalog_freshness +from data_formulator.data_loader.external_data_loader import CatalogCachePolicy + +pytestmark = [pytest.mark.backend, pytest.mark.plugin] + + +class _LoaderClass: + policy = CatalogCachePolicy( + listing_ttl_seconds=0, + metadata_ttl_seconds=0, + refresh_cost="cheap", + automatic_refresh="while_connected", + minimum_retry_seconds=300, + ) + + @classmethod + def catalog_cache_policy(cls) -> CatalogCachePolicy: + return cls.policy + + +def _make_stale(tmp_path: Path, source_id: str) -> None: + path = tmp_path / "catalog_cache" / f"{source_id}.json" + raw = json.loads(path.read_text(encoding="utf-8")) + raw["listing_refreshed_at"] = "2020-01-01T00:00:00Z" + raw["metadata_refreshed_at"] = "2020-01-01T00:00:00Z" + path.write_text(json.dumps(raw), encoding="utf-8") + + +def test_disconnected_source_serves_stale_without_refresh(tmp_path: Path) -> None: + save_catalog(tmp_path, "src", [{"name": "stale"}]) + _make_stale(tmp_path, "src") + + with patch( + "data_formulator.data_connector.resolve_catalog_refresh_target", + return_value=(_LoaderClass, None), + ): + snapshot = ensure_catalog_freshness(tmp_path, "src") + + assert snapshot is not None + assert snapshot.listing_freshness == "stale" + assert snapshot.tables == [{"name": "stale"}] + + +def test_missing_connected_source_refreshes_synchronously(tmp_path: Path) -> None: + loader = MagicMock() + loader.list_tables.return_value = [{"name": "fresh"}] + + with patch( + "data_formulator.data_connector.resolve_catalog_refresh_target", + return_value=(_LoaderClass, loader), + ): + snapshot = ensure_catalog_freshness(tmp_path, "src") + + assert snapshot is not None + assert snapshot.tables == [{"name": "fresh"}] + loader.list_tables.assert_called_once_with() + + +def test_stale_refresh_is_deduplicated_and_serves_stale(tmp_path: Path) -> None: + save_catalog(tmp_path, "src", [{"name": "stale"}]) + _make_stale(tmp_path, "src") + loader = MagicMock() + + with ( + patch( + "data_formulator.data_connector.resolve_catalog_refresh_target", + return_value=(_LoaderClass, loader), + ), + patch("data_formulator.datalake.catalog_refresh._REFRESH_EXECUTOR.submit") as submit, + ): + first = ensure_catalog_freshness(tmp_path, "src") + second = ensure_catalog_freshness(tmp_path, "src") + + assert first is not None and second is not None + assert first.tables == second.tables == [{"name": "stale"}] + submit.assert_called_once() + + +def test_recent_failure_suppresses_retry_and_preserves_tables(tmp_path: Path) -> None: + save_catalog(tmp_path, "src", [{"name": "stale"}]) + _make_stale(tmp_path, "src") + record_catalog_refresh_failure(tmp_path, "src", "timeout") + + with ( + patch( + "data_formulator.data_connector.resolve_catalog_refresh_target", + return_value=(_LoaderClass, MagicMock()), + ), + patch("data_formulator.datalake.catalog_refresh._REFRESH_EXECUTOR.submit") as submit, + ): + snapshot = ensure_catalog_freshness(tmp_path, "src") + + assert snapshot is not None + assert snapshot.last_refresh_error == "timeout" + assert load_catalog(tmp_path, "src") == [{"name": "stale"}] + submit.assert_not_called() \ No newline at end of file diff --git a/tests/frontend/unit/app/connectorFormPersistence.test.ts b/tests/frontend/unit/app/connectorFormPersistence.test.ts new file mode 100644 index 00000000..db28ef1b --- /dev/null +++ b/tests/frontend/unit/app/connectorFormPersistence.test.ts @@ -0,0 +1,53 @@ +import { describe, expect, it } from 'vitest'; + +import { stripConnectorPrefillFromEntries } from '../../../../src/app/connectorFormPersistence'; + +describe('connector form persistence', () => { + it('removes transient prefills from standalone chat messages', () => { + const entries = [{ + id: 'entry-1', + connectorForm: { + sourceType: 'postgresql', + status: 'pending', + prefilled: { host: 'db.example.com', password: 'secret' }, + }, + }]; + + expect(stripConnectorPrefillFromEntries(entries)).toEqual([{ + id: 'entry-1', + connectorForm: { + sourceType: 'postgresql', + status: 'pending', + }, + }]); + expect(entries[0].connectorForm.prefilled.password).toBe('secret'); + }); + + it('removes transient prefills from generalized form artifacts', () => { + const entries = [{ + id: 'turn-1', + form: { + kind: 'connector', + title: 'Connect to PostgreSQL', + connector: { + sourceType: 'postgresql', + status: 'pending', + prefilled: { host: 'db.example.com', password: 'secret' }, + }, + }, + }]; + + expect(stripConnectorPrefillFromEntries(entries)).toEqual([{ + id: 'turn-1', + form: { + kind: 'connector', + title: 'Connect to PostgreSQL', + connector: { + sourceType: 'postgresql', + status: 'pending', + }, + }, + }]); + expect(entries[0].form.connector.prefilled.password).toBe('secret'); + }); +}); \ No newline at end of file diff --git a/tests/frontend/unit/app/dfSliceTableCollections.test.ts b/tests/frontend/unit/app/dfSliceTableCollections.test.ts index c9e6ab7d..7f4c1ed0 100644 --- a/tests/frontend/unit/app/dfSliceTableCollections.test.ts +++ b/tests/frontend/unit/app/dfSliceTableCollections.test.ts @@ -3,6 +3,7 @@ import { describe, expect, it } from "vitest"; import { dataFormulatorReducer, dfActions, + dfSelectors, fetchFieldSemanticType, } from "../../../../src/app/dfSlice"; @@ -300,6 +301,45 @@ describe("split table collections", () => { expect(state.draftNodes[0].parentNodeId).toBe("orders"); }); + it("removes a terminal response's paired follow-up question", () => { + let state = dataFormulatorReducer( + undefined, + dfActions.addTextTurn({ + kind: "text", + id: "textTurn-parent", + displayId: "Parent response", + textKind: "explain", + content: "How can I help?", + parentNodeId: "root", + answered: true, + answer: "Tell me more", + createdAt: 1, + }) + ); + state = dataFormulatorReducer( + state, + dfActions.addTextTurn({ + kind: "text", + id: "textTurn-child", + displayId: "Child response", + textKind: "explain", + content: "Here is more detail.", + parentNodeId: "textTurn-parent", + createdAt: 2, + }) + ); + + state = dataFormulatorReducer(state, dfActions.removeTextTurn("textTurn-child")); + + expect(state.textTurns).toEqual([ + expect.objectContaining({ + id: "textTurn-parent", + answered: false, + }), + ]); + expect(state.textTurns[0]).not.toHaveProperty("answer"); + }); + it("removes loaded-table references with their shelf table", () => { let state = dataFormulatorReducer( undefined, @@ -362,3 +402,32 @@ describe("split table collections", () => { expect(state.draftNodes[0].derive.trigger.tableId).toBe("orders"); }); }); + +describe("text artifact canvas ownership", () => { + it.each([ + ["form", { form: { kind: "connector", title: "Connect", connector: { sourceType: "kusto", status: "pending" } } }], + ["data operation", { dataOperation: { id: "operation-1", plans: [] } }], + ])("keeps a parent %s open for a follow-up explanation", (_label, artifact) => { + const state = { + ...dataFormulatorReducer(undefined, { type: "test/init" }), + focusedId: { type: "text", textId: "explanation-1" }, + textTurns: [ + { + kind: "text", id: "artifact-1", displayId: "artifact-1", + textKind: "explain", content: "Artifact", parentNodeId: "root", + createdAt: 1, ...artifact, + }, + { + kind: "text", id: "explanation-1", displayId: "explanation-1", + textKind: "explain", content: "Follow-up", parentNodeId: "artifact-1", + createdAt: 2, + }, + ], + }; + + expect(dfSelectors.selectCanvasTarget(state as any)).toEqual({ + type: "text", + textId: "artifact-1", + }); + }); +}); diff --git a/tests/frontend/unit/app/fetchWithIdentity.test.ts b/tests/frontend/unit/app/fetchWithIdentity.test.ts index dbcfa342..8d05388a 100644 --- a/tests/frontend/unit/app/fetchWithIdentity.test.ts +++ b/tests/frontend/unit/app/fetchWithIdentity.test.ts @@ -37,6 +37,7 @@ vi.mock('../../../../src/i18n', () => ({ // ---- Imports (after mocks) ----------------------------------------------- import { fetchWithIdentity } from '../../../../src/app/utils'; +import { store } from '../../../../src/app/store'; import { getAccessToken, getUserManager } from '../../../../src/app/oidcConfig'; // ---- Helpers ------------------------------------------------------------- @@ -92,6 +93,22 @@ describe('fetchWithIdentity', () => { expect(headers.get('X-Identity-Id')).toBe('browser:test-browser-id'); }); + it('preserves an explicit workspace instead of the active workspace', async () => { + (store.getState as Mock).mockReturnValue({ + identity: { type: 'browser', id: 'test-browser-id' }, + activeWorkspace: { id: 'workspace-a', displayName: 'Workspace A' }, + serverConfig: { WORKSPACE_BACKEND: 'local' }, + }); + + await fetchWithIdentity('/api/test', { + headers: { 'X-Workspace-Id': 'workspace-b' }, + }); + + const callArgs = (globalThis.fetch as Mock).mock.calls[0]; + const headers = callArgs[1].headers as Headers; + expect(headers.get('X-Workspace-Id')).toBe('workspace-b'); + }); + it('should not modify headers for non-API URLs', async () => { await fetchWithIdentity('/static/file.js'); diff --git a/tests/frontend/unit/app/useAutoSave.test.tsx b/tests/frontend/unit/app/useAutoSave.test.tsx index 101b49c4..acf4e3e1 100644 --- a/tests/frontend/unit/app/useAutoSave.test.tsx +++ b/tests/frontend/unit/app/useAutoSave.test.tsx @@ -25,7 +25,7 @@ vi.mock('../../../../src/app/errorHandler', () => ({ handleApiError: (...args: any[]) => mocks.handleApiError(...args), })); -import { useAutoSave } from '../../../../src/app/useAutoSave'; +import { getSerializableState, useAutoSave } from '../../../../src/app/useAutoSave'; function AutoSaveHarness() { useAutoSave(); @@ -56,4 +56,37 @@ describe('useAutoSave', () => { expect(mocks.handleApiError).toHaveBeenCalledWith(err, 'Auto-save'); }); + + it('strips connector form prefills from workspace snapshots', () => { + const state = { + ...mocks.state, + textTurns: [{ + id: 'turn-1', + form: { + kind: 'connector', + title: 'Connect to PostgreSQL', + connector: { + sourceType: 'postgresql', + status: 'pending', + prefilled: { host: 'db.example.com', password: 'secret' }, + }, + }, + }], + }; + + const snapshot = getSerializableState(state as any); + + expect(snapshot.textTurns).toEqual([{ + id: 'turn-1', + form: { + kind: 'connector', + title: 'Connect to PostgreSQL', + connector: { + sourceType: 'postgresql', + status: 'pending', + }, + }, + }]); + expect(state.textTurns[0].form.connector.prefilled.password).toBe('secret'); + }); }); diff --git a/tests/frontend/unit/app/workspaceService.test.ts b/tests/frontend/unit/app/workspaceService.test.ts index 964af63e..70925cdb 100644 --- a/tests/frontend/unit/app/workspaceService.test.ts +++ b/tests/frontend/unit/app/workspaceService.test.ts @@ -22,7 +22,8 @@ vi.mock('../../../../src/app/stateMigrations', () => ({ import { ApiRequestError } from '../../../../src/app/apiClient'; import { workspaceDB } from '../../../../src/app/workspaceDB'; -import { listWorkspaces, loadWorkspace, saveWorkspaceState } from '../../../../src/app/workspaceService'; +import { listWorkspaces, loadWorkspace, saveWorkspaceState, WorkspaceLoadSupersededError } from '../../../../src/app/workspaceService'; +import { getInputTablePreview } from '../../../../src/app/inputTablePreviewCache'; beforeEach(() => { vi.restoreAllMocks(); @@ -105,4 +106,56 @@ describe('local workspace parity', () => { expect(requestSpy).toHaveBeenCalledOnce(); expect(recoverySaveSpy).not.toHaveBeenCalled(); }); + + it('hydrates previews against the workspace being loaded', async () => { + mockState.serverConfig.WORKSPACE_BACKEND = 'local'; + mockState.activeWorkspace = { id: 'workspace-a', displayName: 'Workspace A' }; + const inputTable = { + kind: 'input-table', + id: 'sales', + displayId: 'Sales', + source: { kind: 'workspace', tableId: 'sales_data' }, + snapshot: { columns: [], rowCount: 1, capturedAt: 1 }, + description: '', + addedAt: 1, + } as const; + const requestSpy = vi.spyOn(await import('../../../../src/app/apiClient'), 'apiRequest') + .mockImplementation(async (url: string, options?: RequestInit) => { + if (url === '/api/sessions/load') { + return { data: { state: { inputTables: [inputTable] } } }; + } + if (url === '/api/tables/sample-table') { + return { data: { rows: [{ amount: 42 }] } }; + } + throw new Error(`Unexpected URL: ${url}`); + }); + + await loadWorkspace('workspace-b'); + + const sampleCall = requestSpy.mock.calls.find(([url]) => url === '/api/tables/sample-table'); + expect(new Headers(sampleCall?.[1]?.headers).get('X-Workspace-Id')).toBe('workspace-b'); + expect(getInputTablePreview(inputTable as any)?.rows).toEqual([{ amount: 42 }]); + }); + + it('rejects a load superseded by a newer workspace switch', async () => { + mockState.serverConfig.WORKSPACE_BACKEND = 'local'; + let resolveFirstLoad!: (value: { data: { state: Record } }) => void; + const firstLoadResponse = new Promise<{ data: { state: Record } }>(resolve => { + resolveFirstLoad = resolve; + }); + const requestSpy = vi.spyOn(await import('../../../../src/app/apiClient'), 'apiRequest') + .mockImplementation(async (_url: string, options?: RequestInit) => { + const workspaceId = JSON.parse(String(options?.body)).id; + if (workspaceId === 'workspace-b') return firstLoadResponse; + return { data: { state: { inputTables: [] } } }; + }); + + const firstLoad = loadWorkspace('workspace-b'); + await vi.waitFor(() => expect(requestSpy).toHaveBeenCalledOnce()); + const secondLoad = loadWorkspace('workspace-c'); + await expect(secondLoad).resolves.toMatchObject({ readOnly: false }); + resolveFirstLoad({ data: { state: { inputTables: [] } } }); + + await expect(firstLoad).rejects.toBeInstanceOf(WorkspaceLoadSupersededError); + }); }); diff --git a/tests/frontend/unit/views/DataSourceSidebar.test.tsx b/tests/frontend/unit/views/DataSourceSidebar.test.tsx index 92357a11..c7b995fd 100644 --- a/tests/frontend/unit/views/DataSourceSidebar.test.tsx +++ b/tests/frontend/unit/views/DataSourceSidebar.test.tsx @@ -8,6 +8,7 @@ const { dispatch, mockState } = vi.hoisted(() => ({ dispatch: vi.fn(), mockState: { dataSourceSidebarOpen: true, + dataSourceSidebarTab: 'sources', serverConfig: { DISABLE_DATA_CONNECTORS: false }, activeWorkspace: null, identity: { type: 'browser', id: 'test-browser' }, @@ -42,6 +43,7 @@ vi.mock('../../../../src/app/dfSlice', () => ({ setSessionLoading: (payload: any) => ({ type: 'session/setLoading', payload }), loadState: (payload: any) => ({ type: 'state/load', payload }), setActiveWorkspace: (payload: any) => ({ type: 'workspace/setActive', payload }), + resetState: () => ({ type: 'state/reset' }), }, dfSelectors: { getAllTables: (state: any) => [...(state.inputTables ?? []), ...(state.derivedTables ?? [])], @@ -99,7 +101,14 @@ vi.mock('../../../../src/views/KnowledgePanel', () => ({ describe('DataSourceSidebar', () => { beforeEach(() => { dispatch.mockClear(); + mockState.dataSourceSidebarTab = 'sources'; + vi.stubGlobal('ResizeObserver', class { + observe() {} + unobserve() {} + disconnect() {} + }); vi.mocked(apiRequest).mockReset(); + vi.mocked(apiRequest).mockResolvedValue({ data: { connectors: [] } }); }); it('leaves loading state when catalog fetch fails', async () => { @@ -138,4 +147,19 @@ describe('DataSourceSidebar', () => { })); }); }); + + it('returns to the landing state without creating an empty workspace', async () => { + mockState.dataSourceSidebarTab = 'sessions'; + render(); + + fireEvent.click(await screen.findByText('New session')); + + expect(dispatch).toHaveBeenCalledWith({ type: 'state/reset' }); + expect(dispatch).not.toHaveBeenCalledWith(expect.objectContaining({ + type: 'state/load', + payload: expect.objectContaining({ + activeWorkspace: expect.objectContaining({ displayName: 'Untitled Session' }), + }), + })); + }); });