diff --git a/plugins/kbagent/skills/kbagent/references/gotchas.md b/plugins/kbagent/skills/kbagent/references/gotchas.md index c0d7b08d..40e1fcc9 100644 --- a/plugins/kbagent/skills/kbagent/references/gotchas.md +++ b/plugins/kbagent/skills/kbagent/references/gotchas.md @@ -392,6 +392,25 @@ seamlessly. This is transparent -- no user action required. - Skipped for: dev/editable installs, `update`/`version` commands - Never crashes the CLI -- update failures are silently ignored +## `lineage build` and sync layouts + +`lineage build` reads synced data from disk and supports both layouts produced by +`kbagent sync pull`: + +- **Flat** (after `sync pull --project X`): `./.keboola/manifest.json` directly in CWD. +- **Nested** (after `sync pull --all-projects`): `.//.keboola/manifest.json` + for each project side by side. + +Pass the matching directory to `--directory` / `-d`: + +- Flat: `kbagent lineage build -d . -o lineage.json` +- Nested: `kbagent lineage build -d /path/to/parent -o lineage.json` + +If the scan finds zero projects, the build still writes the cache file but +emits a warning (both in the human-readable output and as a `warnings` array +in `--json` mode) with a hint about the expected layouts. In JSON mode, inspect +`result["data"]["warnings"]` to detect this situation programmatically. + ## Sync and dev branches When an active branch is set (`branch use --branch ID`), sync commands automatically diff --git a/plugins/kbagent/skills/kbagent/references/lineage-deep-workflow.md b/plugins/kbagent/skills/kbagent/references/lineage-deep-workflow.md index 39d56581..9b7a8a6d 100644 --- a/plugins/kbagent/skills/kbagent/references/lineage-deep-workflow.md +++ b/plugins/kbagent/skills/kbagent/references/lineage-deep-workflow.md @@ -15,6 +15,19 @@ kbagent sync pull --all-projects The sync'd directory structure is the input for lineage analysis. +### Supported layouts + +`lineage build` auto-detects both layouts produced by `sync pull`: + +| Source command | Layout | Example manifest path | +|----------------|--------|------------------------| +| `sync pull --project X` | **Flat** -- CWD *is* the project | `./.keboola/manifest.json` | +| `sync pull --all-projects` | **Nested** -- one subdir per project | `.//.keboola/manifest.json` | + +Pass the directory that *contains* the manifest (flat) or the parent of all +project subdirs (nested). When the build finds zero projects it emits a +warning with a hint rather than silently returning an empty graph. + ## Build lineage ```bash diff --git a/src/keboola_agent_cli/commands/context.py b/src/keboola_agent_cli/commands/context.py index cd32ec30..e2a806d8 100644 --- a/src/keboola_agent_cli/commands/context.py +++ b/src/keboola_agent_cli/commands/context.py @@ -252,8 +252,12 @@ kbagent lineage build --directory PATH --output PATH [--ai] [--refresh] Build column-level lineage graph from sync'd data. Scans all sync'd projects, detects dependencies via config mappings and SQL parsing, saves to cache file. - --refresh runs sync pull first. --ai generates .lineage_ai_tasks.json with - AI analysis tasks for an AI agent to process (2-step flow). + Auto-detects both sync layouts: flat (./.keboola/manifest.json from + sync pull --project X) and nested (.//.keboola/manifest.json from + sync pull --all-projects). Emits a warning in response data if no projects + are found. --refresh runs sync pull first. --ai generates + .lineage_ai_tasks.json with AI analysis tasks for an AI agent to process + (2-step flow). kbagent lineage show --load PATH [--upstream NODE] [--downstream NODE] [--column COL] [--columns] [--project ALIAS] [--depth N] diff --git a/src/keboola_agent_cli/commands/lineage.py b/src/keboola_agent_cli/commands/lineage.py index da57cfb1..3000bc7a 100644 --- a/src/keboola_agent_cli/commands/lineage.py +++ b/src/keboola_agent_cli/commands/lineage.py @@ -158,6 +158,9 @@ def lineage_build( formatter.console.print( " Next: let your AI agent process the tasks, then re-run this command." ) + # Surface warnings (e.g. empty scan) so users notice layout issues. + for warning in result.get("warnings", []) or []: + formatter.console.print(f"\n[yellow]Warning:[/yellow] {warning}") formatter.console.print(f"\n Saved to: {output}") diff --git a/src/keboola_agent_cli/services/deep_lineage_service.py b/src/keboola_agent_cli/services/deep_lineage_service.py index f28e2ba6..bf9c47d8 100644 --- a/src/keboola_agent_cli/services/deep_lineage_service.py +++ b/src/keboola_agent_cli/services/deep_lineage_service.py @@ -375,18 +375,40 @@ def build_lineage( if present. Use generate_ai_tasks=True to write a task file for the AI agent to process. + Supports both sync layouts: + - Flat (``sync pull --project X``): ``root/.keboola/manifest.json`` + - Nested (``sync pull --all-projects``): ``root//.keboola/manifest.json`` + Args: - root: Root directory containing sync'd project subdirectories. + root: Root directory containing sync'd project data. Can be either + the project directory itself (flat layout) or a parent directory + with one subdirectory per project (nested layout). generate_ai_tasks: If True, write .lineage_ai_tasks.json for AI. Returns: - Dict with lineage graph data, summary, and ai_status. + Dict with lineage graph data, summary, ai_status, and ``warnings`` + (list of human-readable warnings emitted during the build, e.g. when + no sync'd projects were found). """ project_id_to_alias = self._build_project_map() + warnings: list[str] = [] # Phase 1: Scan graph = self._scan_projects(root, project_id_to_alias) + # Empty-scan warning -- most often caused by a layout mismatch between + # ``sync pull --project X`` (flat) and ``sync pull --all-projects`` (nested). + if not graph.tables and not graph.configurations: + warning = ( + f"No synced projects found in {root}. Expected either:\n" + f" - {root}/.keboola/manifest.json (single-project flat layout), or\n" + f" - {root}//.keboola/manifest.json (multi-project nested layout)\n" + "Hint: run 'kbagent sync pull --all-projects' or pass --directory " + "pointing to the parent of your synced projects." + ) + warnings.append(warning) + logger.warning(warning) + # Phase 2: Deterministic edges self._build_deterministic_edges(graph, project_id_to_alias) @@ -403,6 +425,7 @@ def build_lineage( result = graph.to_dict() result["ai_status"] = ai_status + result["warnings"] = warnings return result def build_and_cache( @@ -480,53 +503,99 @@ def _build_project_map(self) -> dict[int, str]: return mapping def _scan_projects(self, root: Path, project_id_to_alias: dict[int, str]) -> LineageGraph: - """Scan all sync'd projects and build initial graph.""" + """Scan all sync'd projects and build initial graph. + + Supports two directory layouts (written by ``kbagent sync pull``): + - **Flat** (``sync pull --project X``): manifest lives directly at + ``root/.keboola/manifest.json`` and ``root`` itself *is* the project. + - **Nested** (``sync pull --all-projects``): each project is a + subdirectory, i.e. ``root//.keboola/manifest.json``. + + Flat layout is detected first and, when present, returned exclusively + (nested iteration would be redundant -- in the flat case there are no + sibling project directories under ``root``). + """ graph = LineageGraph() - for project_dir in sorted(root.iterdir()): - if not project_dir.is_dir() or project_dir.name.startswith("."): - continue - manifest_path = project_dir / ".keboola" / "manifest.json" - if not manifest_path.exists(): - continue + flat_manifest = root / ".keboola" / "manifest.json" + if flat_manifest.exists(): + project_alias = self._resolve_alias_from_manifest(flat_manifest, fallback=root.name) + self._scan_one_project(root, project_alias, project_id_to_alias, graph) + else: + for project_dir in sorted(root.iterdir()): + if not project_dir.is_dir() or project_dir.name.startswith("."): + continue + manifest_path = project_dir / ".keboola" / "manifest.json" + if not manifest_path.exists(): + continue + self._scan_one_project(project_dir, project_dir.name, project_id_to_alias, graph) + # Store mapping for cross-project resolution + graph._project_id_to_alias = project_id_to_alias # type: ignore[attr-defined] + return graph + + def _resolve_alias_from_manifest(self, manifest_path: Path, *, fallback: str) -> str: + """Resolve a project alias for a flat-layout root. + + In a flat layout the parent directory name is meaningless (it's the + user's CWD, not the project alias). We prefer the alias configured in + ``ConfigStore`` for the project id recorded in the manifest; if no such + mapping exists, we fall back to the directory name so the graph stays + consistent but still disambiguated from other projects on disk. + """ + try: with open(manifest_path) as f: manifest = json.load(f) + project_id = int(manifest.get("project", {}).get("id", 0) or 0) + except (OSError, json.JSONDecodeError, ValueError): + return fallback - project_alias = project_dir.name - project_id = manifest.get("project", {}).get("id", 0) - project_id_to_alias[project_id] = project_alias - - # Scan storage tables - storage_dir = project_dir / "storage" / "tables" - if storage_dir.exists(): - for bucket_dir in sorted(storage_dir.iterdir()): - if not bucket_dir.is_dir(): - continue - for table_file in sorted(bucket_dir.glob("*.json")): - with open(table_file) as f: - meta = json.load(f) - table = Table( - table_id=meta["id"], - project_alias=project_alias, - project_id=project_id, - bucket_id=meta["id"].rsplit(".", 1)[0], - name=meta["name"], - columns=meta.get("columns", []), - primary_key=meta.get("primary_key", []), - rows_count=meta.get("rows_count", 0), - ) - graph.tables[table.fqn] = table + if project_id: + app_config = self._config_store.load() + for alias, project in app_config.projects.items(): + if project.project_id == project_id: + return alias + return fallback - # Scan configurations - for config_entry in manifest.get("configurations", []): - self._scan_configuration( - project_dir, config_entry, project_alias, project_id, graph - ) + def _scan_one_project( + self, + project_dir: Path, + project_alias: str, + project_id_to_alias: dict[int, str], + graph: LineageGraph, + ) -> None: + """Populate ``graph`` with tables and configs from a single project dir.""" + manifest_path = project_dir / ".keboola" / "manifest.json" + with open(manifest_path) as f: + manifest = json.load(f) + + project_id = manifest.get("project", {}).get("id", 0) + project_id_to_alias[project_id] = project_alias + + # Scan storage tables + storage_dir = project_dir / "storage" / "tables" + if storage_dir.exists(): + for bucket_dir in sorted(storage_dir.iterdir()): + if not bucket_dir.is_dir(): + continue + for table_file in sorted(bucket_dir.glob("*.json")): + with open(table_file) as f: + meta = json.load(f) + table = Table( + table_id=meta["id"], + project_alias=project_alias, + project_id=project_id, + bucket_id=meta["id"].rsplit(".", 1)[0], + name=meta["name"], + columns=meta.get("columns", []), + primary_key=meta.get("primary_key", []), + rows_count=meta.get("rows_count", 0), + ) + graph.tables[table.fqn] = table - # Store mapping for cross-project resolution - graph._project_id_to_alias = project_id_to_alias # type: ignore[attr-defined] - return graph + # Scan configurations + for config_entry in manifest.get("configurations", []): + self._scan_configuration(project_dir, config_entry, project_alias, project_id, graph) def _scan_configuration( self, diff --git a/tests/test_deep_lineage_service.py b/tests/test_deep_lineage_service.py index 007b33f6..b1e70c9a 100644 --- a/tests/test_deep_lineage_service.py +++ b/tests/test_deep_lineage_service.py @@ -605,3 +605,256 @@ def test_missing_cache_file(self) -> None: ], ) assert result.exit_code == 1 + + +# --------------------------------------------------------------------------- +# Sync layout handling: flat (single project in CWD) vs. nested (multi-project) +# --------------------------------------------------------------------------- + + +def _create_flat_project( + root: Path, + *, + project_id: int = 42, + with_storage: bool = True, + with_config: bool = True, +) -> None: + """Create a single-project flat layout directly under ``root``. + + This mirrors what ``kbagent sync pull --project X`` produces: the + ``.keboola/manifest.json`` lives at the provided root (no alias subdir). + """ + keboola = root / ".keboola" + keboola.mkdir(parents=True) + configurations: list[dict] = [] + if with_config: + configurations.append( + { + "branchId": 1, + "componentId": "keboola.snowflake-transformation", + "id": "cfg-flat", + "path": "transformation/keboola.snowflake-transformation/flat-transform", + "rows": [], + } + ) + manifest = { + "version": 2, + "project": {"id": project_id, "name": "Flat Project"}, + "configurations": configurations, + "branches": [], + } + (keboola / "manifest.json").write_text(json.dumps(manifest)) + + if with_storage: + storage = root / "storage" / "tables" / "in-c-flat" + storage.mkdir(parents=True) + (storage / "accounts.json").write_text( + json.dumps( + { + "id": "in.c-flat.accounts", + "name": "accounts", + "columns": ["id", "name"], + "primary_key": ["id"], + "rows_count": 10, + } + ) + ) + + if with_config: + transform_dir = ( + root / "main" / "transformation" / "keboola.snowflake-transformation" / "flat-transform" + ) + transform_dir.mkdir(parents=True) + (transform_dir / "_config.yml").write_text( + yaml.dump( + { + "version": 2, + "name": "Flat Transform", + "input": { + "tables": [ + {"source": "in.c-flat.accounts", "destination": "accounts"}, + ] + }, + "output": {"tables": []}, + } + ) + ) + (transform_dir / "transform.sql").write_text('SELECT * FROM "in.c-flat"."accounts"\n') + + +class TestDeepLineageLayouts: + """Covers the flat vs. nested sync-layout detection in ``_scan_projects``.""" + + def test_flat_layout_detects_single_project(self, tmp_path: Path) -> None: + """``root/.keboola/manifest.json`` is treated as a single project.""" + root = tmp_path / "synced-foo" + root.mkdir() + _create_flat_project(root) + + store = ConfigStore(config_dir=tmp_path / "cfg") + (tmp_path / "cfg").mkdir() + service = DeepLineageService(config_store=store) + + with patch.object(service, "_add_cross_project_lineage"): + result = service.build_lineage(root) + + assert result["summary"]["tables"] == 1 + assert result["summary"]["configurations"] == 1 + # Edges: input_mapping + sql_tokenizer (table referenced in SQL) + assert result["summary"]["edges"] >= 1 + assert result["warnings"] == [] + + def test_flat_layout_uses_config_store_alias_when_available(self, tmp_path: Path) -> None: + """In flat mode the alias comes from ConfigStore, not the CWD name.""" + from keboola_agent_cli.models import AppConfig, ProjectConfig + + root = tmp_path / "some-random-cwd" + root.mkdir() + _create_flat_project(root, project_id=77) + + cfg_dir = tmp_path / "cfg" + cfg_dir.mkdir() + store = ConfigStore(config_dir=cfg_dir) + config = AppConfig() + config.projects["prod"] = ProjectConfig( + stack_url="https://connection.keboola.com", + token="test-token", + project_id=77, + ) + store.save(config) + + service = DeepLineageService(config_store=store) + with patch.object(service, "_add_cross_project_lineage"): + result = service.build_lineage(root) + + # Exactly one table keyed by the ConfigStore alias, not by dir name. + assert list(result["tables"].keys()) == ["prod:in.c-flat.accounts"] + + def test_flat_layout_falls_back_to_dir_name_without_config_store_match( + self, tmp_path: Path + ) -> None: + """If no alias matches ``project.id`` we keep the directory name.""" + root = tmp_path / "my-alias" + root.mkdir() + _create_flat_project(root, project_id=999) + + cfg_dir = tmp_path / "cfg" + cfg_dir.mkdir() + store = ConfigStore(config_dir=cfg_dir) # empty store + service = DeepLineageService(config_store=store) + + with patch.object(service, "_add_cross_project_lineage"): + result = service.build_lineage(root) + + assert list(result["tables"].keys()) == ["my-alias:in.c-flat.accounts"] + + def test_nested_layout_still_works(self, tmp_path: Path) -> None: + """Regression guard: nested ``sync pull --all-projects`` layout.""" + root = _create_sync_tree(tmp_path) # nested layout helper + store = ConfigStore(config_dir=tmp_path / "cfg") + (tmp_path / "cfg").mkdir() + service = DeepLineageService(config_store=store) + + with patch.object(service, "_add_cross_project_lineage"): + result = service.build_lineage(root) + + assert result["summary"]["tables"] == 2 + assert result["summary"]["configurations"] == 2 + assert result["warnings"] == [] + + def test_empty_directory_emits_warning(self, tmp_path: Path) -> None: + """Neither flat nor nested layout -> zero-scan + hint warning.""" + root = tmp_path / "empty" + root.mkdir() + store = ConfigStore(config_dir=tmp_path / "cfg") + (tmp_path / "cfg").mkdir() + service = DeepLineageService(config_store=store) + + with patch.object(service, "_add_cross_project_lineage"): + result = service.build_lineage(root) + + assert result["summary"]["tables"] == 0 + assert result["summary"]["configurations"] == 0 + assert len(result["warnings"]) == 1 + warning = result["warnings"][0] + assert "No synced projects found" in warning + assert "flat layout" in warning + assert "nested layout" in warning + assert "sync pull --all-projects" in warning + + +class TestLineageBuildCli: + """CLI-layer smoke tests for the flat/empty layouts.""" + + def _runner(self): + return __import__("typer.testing", fromlist=["CliRunner"]).CliRunner() + + def test_build_flat_layout_non_empty_graph(self, tmp_path: Path) -> None: + """``lineage build`` against a flat-layout dir produces a non-empty graph.""" + from keboola_agent_cli.cli import app + + # Place synced project next to cfg dir so cwd doesn't matter. + synced = tmp_path / "synced" + synced.mkdir() + _create_flat_project(synced) + cache_path = tmp_path / "lineage.json" + + with patch( + "keboola_agent_cli.services.deep_lineage_service." + "DeepLineageService._add_cross_project_lineage" + ): + result = self._runner().invoke( + app, + [ + "--json", + "--config-dir", + str(tmp_path / "cfg"), + "lineage", + "build", + "--directory", + str(synced), + "--output", + str(cache_path), + ], + ) + + assert result.exit_code == 0, result.output + data = json.loads(result.output)["data"] + assert data["summary"]["tables"] == 1 + assert data["summary"]["configurations"] == 1 + assert data["warnings"] == [] + assert cache_path.exists() + + def test_build_empty_directory_exits_zero_with_warning(self, tmp_path: Path) -> None: + """Empty dir -> exit 0 with a warning describing the expected layouts.""" + from keboola_agent_cli.cli import app + + empty = tmp_path / "empty" + empty.mkdir() + cache_path = tmp_path / "lineage.json" + + with patch( + "keboola_agent_cli.services.deep_lineage_service." + "DeepLineageService._add_cross_project_lineage" + ): + result = self._runner().invoke( + app, + [ + "--json", + "--config-dir", + str(tmp_path / "cfg"), + "lineage", + "build", + "--directory", + str(empty), + "--output", + str(cache_path), + ], + ) + + assert result.exit_code == 0, result.output + data = json.loads(result.output)["data"] + assert data["summary"]["tables"] == 0 + assert len(data["warnings"]) == 1 + assert "No synced projects found" in data["warnings"][0] + assert cache_path.exists()