various fixes

This commit is contained in:
Chenglong Wang committed 2026-08-14 15:38:45 -07:00
1 parent 432ab7259c
commit eaa33352f0
38 files changed
+1806 -198

No files matched your search

+2 -56
View File
@@ -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
@@ -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
@@ -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()
@@ -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": {
+30
View File
@@ -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(
+62 -2
View File
@@ -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
@@ -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.
@@ -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
@@ -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:
@@ -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
+5 -1
View File
@@ -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);
+16
View File
@@ -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;
});
};
+25 -5
View File
@@ -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<string>();
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;
},
+10 -1
View File
@@ -37,4 +37,13 @@ export const invalidateInputTablePreview = (tableId: string): void => {
}
};
export const clearInputTablePreviewCache = (): void => previewCache.clear();
export const clearInputTablePreviewCache = (): void => previewCache.clear();
export const replaceInputTablePreviews = (
previews: Array<{ table: InputTable; rows: Record<string, unknown>[]; fetchedAt?: number }>,
): void => {
previewCache.clear();
for (const { table, rows, fetchedAt } of previews) {
setInputTablePreview(table, rows, fetchedAt);
}
};
+3 -11
View File
@@ -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 = {
+4 -1
View File
@@ -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<string,
const result: Record<string, unknown> = {};
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
+1 -1
View File
@@ -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);
}
+39 -7
View File
@@ -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<string, any>): Promise<void> {
clearInputTablePreviewCache();
type PreparedInputTablePreview = {
table: InputTable;
rows: Record<string, unknown>[];
};
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<string, any>,
workspaceId: string,
): Promise<PreparedInputTablePreview[]> {
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<string, unknown>[] }>(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<WorkspaceSummary[]> {
/** Load a workspace's saved state. Returns null if not found. */
export async function loadWorkspace(id: string): Promise<{ state: Record<string, any>; 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<string,
});
if (!data.state) return null;
const state = migrateState(data.state);
await hydrateInputTablePreviews(state);
const previews = await prepareInputTablePreviews(state, id);
assertCurrentWorkspaceLoad(generation);
replaceInputTablePreviews(previews);
const savedWs = state.activeWorkspace;
const displayName = savedWs?.displayName || id;
if (ephemeral) {
@@ -140,11 +168,15 @@ export async function loadWorkspace(id: string): Promise<{ state: Record<string,
}
return { state, displayName, readOnly: false };
} catch (error) {
if (error instanceof WorkspaceLoadSupersededError) throw error;
assertCurrentWorkspaceLoad(generation);
const unavailable = error instanceof ApiRequestError
&& ['WORKSPACE_EXPIRED', 'TABLE_NOT_FOUND'].includes(error.apiError.code);
if (!ephemeral || !unavailable) throw error;
const recovery = await workspaceDB.load(id);
if (!recovery) return null;
assertCurrentWorkspaceLoad(generation);
replaceInputTablePreviews([]);
return {
state: createRecoveryState(migrateState(recovery.state)),
displayName: recovery.displayName,
+18 -7
View File
@@ -128,13 +128,13 @@ export interface DraftNode {
export type ThreadNode = DraftNode | DictTable | LoadedTableNode;
/**
* A first-class **text turn** (clarify / explain) in the thread — a sibling to
* charts, tables, and reports (see design-docs/41). Placed in the thread by its
* one authored edge, `parentNodeId` (design-docs/42): the node the user was
* asking from when the turn was created. Focusing one overlays a panel above the
* chat without taking over the canvas (the canvas keeps showing `sourceChartId`);
* deleting one is a generic artifact delete. (Delegate is NOT a text turn — a
* hand-off is an agent action, handled directly.)
* A first-class interaction in the thread: either a clarify/explain turn or a
* canvas-owning artifact such as a data operation or form. It is placed by its
* one authored edge, `parentNodeId` (design-docs/42). Plain text turns overlay
* the chat without taking over the canvas. Embedded artifacts own the canvas;
* form artifacts also show their accompanying text above chat as feedback.
* Deleting either uses the same generic artifact path. Delegate is not a turn;
* a hand-off is an agent action handled directly.
*/
export interface TextTurn {
kind: 'text';
@@ -151,6 +151,8 @@ export interface TextTurn {
options?: ClarificationQuestion[];
/** Display-only immutable loading alternatives for a data-operation pause. */
dataOperation?: DataOperation;
/** A user-confirmed form artifact that owns the canvas while focused. */
form?: FormArtifact;
/** True once the user has responded to THIS clarify — it then locks
* (read-only). A later response is a *new* conversation, not a re-answer. */
answered?: boolean;
@@ -271,6 +273,15 @@ export interface ConnectorFormPrompt {
tableCount?: number; // optional: tables discovered on connect
}
export interface ConnectorFormArtifact {
kind: 'connector';
title: string;
connector: ConnectorFormPrompt;
}
/** Canvas-owning form artifacts. Add future form kinds to this union. */
export type FormArtifact = ConnectorFormArtifact;
export interface ChatMessage {
id: string;
role: 'user' | 'assistant';
+37 -24
View File
@@ -49,9 +49,15 @@ interface ConnectorFormCardProps {
defaultExpanded?: boolean;
/** 'bare' drops the card chrome — the canvas already frames the form. */
variant?: 'card' | 'bare';
/** Analyst canvas owns TextTurn state; standalone chat uses its message reducer. */
onResolved?: (resolution: {
status: 'connected';
connectorId?: string;
connectionName: string;
}) => void;
}
export const ConnectorFormCard: React.FC<ConnectorFormCardProps> = ({ messageId, prompt, defaultExpanded = true, variant = 'card' }) => {
export const ConnectorFormCard: React.FC<ConnectorFormCardProps> = ({ messageId, prompt, defaultExpanded = true, variant = 'card', onResolved }) => {
const theme = useTheme();
const { t } = useTranslation();
const dispatch = useDispatch<AppDispatch>();
@@ -202,12 +208,16 @@ export const ConnectorFormCard: React.FC<ConnectorFormCardProps> = ({ 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<ConnectorFormCardProps> = ({ 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<ConnectorFormCardProps> = ({ messageId,
authMode={meta.auth_mode}
authPaths={meta.auth_paths}
compact
hideInstructions
comfortableSpacing={isBare}
hideInstructions={!isBare}
onImport={() => {}}
onFinish={(status, message) => {
dispatch(dfActions.addMessages({
+2 -1
View File
@@ -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",
+2 -1
View File
@@ -479,7 +479,8 @@
"hideEarlierTurns": "收起更早的对话",
"working": "处理中...",
"waitingForClarification": "等待澄清...",
"emptySession": "还没有数据 \u2014 在下方让智能体加载数据,结果会显示在这里。",
"emptySessionTitle": "这里还没有数据",
"emptySession": "请在下方让智能体加载数据。准备好后,数据会显示在这里。",
"startingRun": "正在处理您的请求\u2026",
"rename": "重命名",
"refreshSettings": "刷新设置",
+74 -14
View File
@@ -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<AppDispatch>();
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 (
<FormControlLabel
control={(
<Switch
size="small"
checked={checked}
onChange={(_event, nextChecked) => {
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<{
<Typography variant="body2" sx={labelSx}>
{paramDef.name}{paramDef.required ? ' *' : ''}
</Typography>
<DraftTextField
{paramDef.type === 'boolean' || paramDef.type === 'bool' ? renderBooleanParam(paramDef) : <DraftTextField
size="small" fullWidth
sx={fieldSx}
type={paramDef.type === 'password' ? 'password' : 'text'}
@@ -740,7 +784,7 @@ export const DataLoaderForm: React.FC<{
placeholder={getParamPlaceholder(paramDef)}
onDraftChange={(value) => updateParamDraft(paramDef.name, value)}
onCommit={(value) => commitParamDraft(paramDef.name, value)}
/>
/>}
</Box>
))}
</Box>
@@ -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) => (
<Box key={paramDef.name} sx={{ minWidth: 0 }}>
<Typography variant="body2" sx={labelSx}>
{paramDef.name}{paramDef.required ? ' *' : ''}
</Typography>
{input}
<Box sx={{
display: 'grid',
gridTemplateColumns: action ? 'minmax(0, 1fr) 32px' : 'minmax(0, 1fr)',
columnGap: action ? 0.5 : 0,
alignItems: 'center',
minWidth: 0,
}}>
{input}
{action && (
<Box sx={{ width: 32, height: 32, display: 'flex', alignItems: 'center', justifyContent: 'center' }}>
{action}
</Box>
)}
</Box>
</Box>
);
return (
@@ -769,11 +826,11 @@ export const DataLoaderForm: React.FC<{
{tierParams.map((paramDef) => (
isKustoCluster(paramDef.name) ? (
renderFieldRow(paramDef,
<Box sx={{ display: 'flex', alignItems: 'center', gap: 0.5 }}>
<Autocomplete
sx={{ flex: 1, minWidth: 0 }}
sx={{ width: '100%', minWidth: 0 }}
freeSolo
options={[KUSTO_HELP_CLUSTER]}
slotProps={{ listbox: { sx: { fontSize: bodyFontSize } } }}
value={params[paramDef.name] ?? ''}
onChange={(_event, value) => {
dispatch(dfActions.updateDataLoaderConnectParam({
@@ -803,18 +860,18 @@ export const DataLoaderForm: React.FC<{
placeholder={getParamHelp(paramDef) || getParamPlaceholder(paramDef)}
/>
)}
/>
/>,
<Tooltip title={t('db.findClusterPortal', { defaultValue: 'Find your cluster in the Azure portal' })}>
<IconButton size="small" component="a" href="https://portal.azure.com/#browse/Microsoft.Kusto%2Fclusters" target="_blank" rel="noopener noreferrer">
<OpenInNewIcon sx={{ fontSize: iconVar.md }} />
</IconButton>
</Tooltip>
</Box>
)
) : isKustoDatabase(paramDef.name) ? (
renderFieldRow(paramDef,
<Autocomplete
freeSolo
slotProps={{ listbox: { sx: { fontSize: bodyFontSize } } }}
open={databaseMenuOpen}
onOpen={() => {
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,
<Autocomplete
freeSolo
options={paramDef.options}
slotProps={{ listbox: { sx: { fontSize: bodyFontSize } } }}
value={params[paramDef.name] ?? ''}
onChange={(_event, value) => {
dispatch(dfActions.updateDataLoaderConnectParam({
+28 -18
View File
@@ -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,
}}
>
<CircularProgress size={40} />
<Typography variant="body1" color="text.secondary">
{sessionLoadingLabel || t('session.loadingSessions')}
</Typography>
<Button
variant="text"
size="small"
onClick={() => dispatch(dfActions.setSessionLoading({ loading: false }))}
sx={{ mt: 1, textTransform: 'none', color: 'text.secondary' }}
>
{t('app.cancel')}
</Button>
<AnvilLoader
height="100%"
label={sessionLoadingLabel || t('session.loadingSessions')}
action={(
<Button
variant="text"
size="small"
onClick={() => dispatch(dfActions.setSessionLoading({ loading: false }))}
sx={{ minWidth: 0, px: 0.5, textTransform: 'none', color: 'text.secondary' }}
>
{t('app.cancel')}
</Button>
)}
sx={{ width: '100%' }}
/>
</Backdrop>
{selectedModelId == undefined && (
<Box sx={{
+13 -11
View File
@@ -71,7 +71,7 @@ import { apiRequest } from '../app/apiClient';
import { LoadableState, errorLoadable, loadingLoadable, successLoadable } from '../app/loadableState';
import { getConnectorIcon, connectorSortOrder, RelationalDBIcon } from '../icons';
import { loadTable } from '../app/tableThunks';
import { listWorkspaces, loadWorkspace, deleteWorkspace, exportWorkspace, importWorkspace, updateWorkspaceMeta, onWorkspaceListChanged } from '../app/workspaceService';
import { listWorkspaces, loadWorkspace, deleteWorkspace, exportWorkspace, importWorkspace, updateWorkspaceMeta, onWorkspaceListChanged, WorkspaceLoadSupersededError } from '../app/workspaceService';
import type { WorkspaceSummary } from '../app/workspaceService';
import { borderColor } from '../app/tokens';
@@ -692,14 +692,21 @@ const DataSourceSidebarPanel: React.FC<{
dispatch(dfActions.setSessionLoading({ loading: true, label: t('sidebar.openingWorkspace') }));
try {
const result = await loadWorkspace(sessionId);
if (result && Object.keys(result.state).length > 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<{
}}>
<Box
onClick={() => {
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',
+44 -11
View File
@@ -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
? <AttachFileIcon sx={{ width: 14, height: 14, color: iconColor }} />
: getEntryGutterIcon(
{ from: 'data-agent', to: 'user', role: turn.textKind, content: '' },
iconColor,
);
const card = (
<Card className={`data-thread-card ${isFocused ? 'selected-card' : ''}`} elevation={0}
sx={{
@@ -3311,7 +3320,17 @@ export const DataThread: FC<{sx?: SxProps, centered?: boolean, denseColumns?: bo
) : (
// A session can open before any data exists (the agent loads it), so
// say so rather than leaving the panel blank.
<Box sx={{ direction: 'ltr', boxSizing: 'border-box', px: 2, py: 3, width: panelWidth }}>
<Box sx={{
direction: 'ltr',
boxSizing: 'border-box',
px: 2,
width: panelWidth,
height: '100%',
display: 'flex',
alignItems: 'center',
justifyContent: 'center',
textAlign: 'center',
}}>
{workPending ? (
<Box sx={{ display: 'flex', alignItems: 'center', gap: 1 }}>
<CircularProgress size={12} thickness={5} sx={{ color: theme.palette.primary.main }} />
@@ -3320,9 +3339,23 @@ export const DataThread: FC<{sx?: SxProps, centered?: boolean, denseColumns?: bo
</Typography>
</Box>
) : (
<Typography variant="body2" sx={{ fontSize: textVar.xs, color: 'text.disabled' }}>
{t('dataThread.emptySession')}
</Typography>
<Box sx={{ display: 'flex', alignItems: 'center', gap: 1.5, maxWidth: 560, textAlign: 'left' }}>
<Box
component="img"
src={dfLogo}
alt=""
aria-hidden="true"
sx={{ width: 32, height: 30, opacity: 0.82, flexShrink: 0 }}
/>
<Box sx={{ minWidth: 0 }}>
<Typography sx={{ fontSize: textVar.sm, lineHeight: 1.4, fontWeight: 600, color: 'text.primary' }}>
{t('dataThread.emptySessionTitle')}
</Typography>
<Typography variant="body2" sx={{ mt: 0.25, fontSize: textVar.xs, lineHeight: 1.5, color: 'text.secondary' }}>
{t('dataThread.emptySession')}
</Typography>
</Box>
</Box>
)}
</Box>
);
+69 -8
View File
@@ -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;
+56 -6
View File
@@ -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 (
<Box id="vis-view-canvas" sx={{ width: '100%', height: '100%', overflow: 'auto', bgcolor: 'background.default' }}>
<Box sx={{ width: '100%', maxWidth: 624, mx: 'auto', px: { xs: 2, sm: 3, md: 4 }, pt: { xs: 2, md: 3 }, pb: { xs: 3, md: 4 }, boxSizing: 'border-box' }}>
<Box sx={{ display: 'flex', alignItems: 'center', gap: 1, pb: 1.5, mb: 2, borderBottom: 1, borderColor: 'divider' }}>
{getConnectorIcon(form.connector.sourceType, {
sx: { fontSize: iconVar.lg, color: 'text.secondary', flexShrink: 0 },
})}
<Typography sx={{ fontSize: textVar.lg, fontWeight: 600, lineHeight: 1.35 }}>
{form.title}
</Typography>
</Box>
<ConnectorFormCard
messageId={turn.id}
prompt={form.connector}
variant="bare"
onResolved={(resolution) => {
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,
},
},
}));
}}
/>
</Box>
</Box>
);
}
};
// 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<VisPanelProps> = 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<VisPanelProps> = function VisualizationView
if (focusedOperationTurn?.dataOperation) {
return <DataOperationCanvas operation={focusedOperationTurn.dataOperation} />;
}
if (focusedFormTurn?.form) {
return <FormArtifactCanvas turn={focusedFormTurn} form={focusedFormTurn.form} />;
}
let focusedChart = allCharts.find(c => c.id == focusedChartId) as Chart;
let synthesisRunning = focusedChartId ? chartSynthesisInProgress.includes(focusedChartId) : false;
@@ -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"})
@@ -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": ""})
+92
View File
@@ -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
+109
View File
@@ -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()
@@ -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');
});
});
@@ -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",
});
});
});
@@ -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');
+34 -1
View File
@@ -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');
});
});
@@ -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<string, unknown> } }) => void;
const firstLoadResponse = new Promise<{ data: { state: Record<string, unknown> } }>(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);
});
});
@@ -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(<DataSourceSidebar />);
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' }),
}),
}));
});
});