fix: resolve 13 red-team CLI bugs (6 P0, 3 P1, 4 P2)

Closes #62

P0 trust/correctness:
- Resolve relative source_dir/build_dir against pipeline file, not cwd (#3)
- Clear synix_dir on --build-dir override to prevent stale routing (#4)
- Propagate source load failures instead of silently succeeding (#5)
- Add Layer.level read-only property to fix info crash (#8)
- Rewrite info/status to read .synix/ snapshot store, not legacy build/ (#9)
- Diff uses RefStore run history instead of legacy versions/ dir (#11)

P1 operator consistency:
- Planner uses estimated-count placeholders for downstream cardinality (#1)
- Standardize invalid ref handling to sys.exit(1) across all inspectors (#10)
- Clean also removes refs/releases/ ref files (#12)

P2 docs/discoverability:
- Mesh commands honor SYNIX_MESH_ROOT env var via resolve_mesh_root() (#2)
- Batch planner tracks DAG cardinality instead of estimate_output_count(1) (#6)
- Fix llms.txt diff syntax to match actual CLI (#7)
- Add refs/plans to refs list prefix scan (#13)

Author: marklubin

llms.txt

@@ -39,7 +39,7 @@ synix lineage <artifact-id>                      # Provenance tree
 synix list                                       # All artifacts in current snapshot
 synix show <id>                                  # Render artifact content
 synix plan pipeline.py                           # Dry-run with cost estimates
-synix diff <ref-a> <ref-b>                      # Structured diff between refs
+synix diff [ARTIFACT_ID] --old-build-dir DIR     # Diff artifacts between build dirs
 synix clean                                      # Remove releases and work state
 ```
 

src/synix/build/batch_runner.py

@@ -612,6 +612,7 @@ def plan_batch(pipeline: Pipeline) -> list[dict]:
     compute_levels(pipeline.layers)
     build_order = resolve_build_order(pipeline)
 
+    layer_cardinality: dict[str, int] = {}
     layers_info = []
     for layer in build_order:
         info: dict = {
@@ -622,6 +623,12 @@ def plan_batch(pipeline: Pipeline) -> list[dict]:
 
         if isinstance(layer, Source):
             info["mode"] = "source"
+            try:
+                source_config = {"source_dir": pipeline.source_dir}
+                artifacts = layer.load(source_config)
+                layer_cardinality[layer.name] = len(artifacts)
+            except Exception:
+                layer_cardinality[layer.name] = 1
         elif isinstance(layer, Transform):
             try:
                 mode = _resolve_batch_mode(layer, pipeline)
@@ -637,13 +644,15 @@ def plan_batch(pipeline: Pipeline) -> list[dict]:
             info["mode"] = mode
             info["batch_param"] = layer.batch
 
-            # Estimate work units
+            # Estimate work units using DAG-aware cardinality tracking
             dep_counts = sum(
-                dep_layer.estimate_output_count(1) for dep_layer in layer.depends_on if isinstance(dep_layer, Transform)
+                layer_cardinality.get(dep.name, 1)
+                for dep in layer.depends_on
             )
             if dep_counts == 0:
-                dep_counts = len(layer.depends_on) or 1
+                dep_counts = 1
             info["estimated_requests"] = layer.estimate_output_count(dep_counts)
+            layer_cardinality[layer.name] = layer.estimate_output_count(dep_counts)
         else:
             info["mode"] = "unknown"
 

src/synix/build/diff.py

@@ -4,13 +4,16 @@
 
 import difflib
 import json
+import logging
 from dataclasses import dataclass, field
 from pathlib import Path
 
-from synix.build.refs import synix_dir_for_build_dir
-from synix.build.snapshot_view import SnapshotArtifactCache
+from synix.build.refs import RefStore, synix_dir_for_build_dir
+from synix.build.snapshot_view import SnapshotArtifactCache, SnapshotView
 from synix.core.models import Artifact
 
+logger = logging.getLogger(__name__)
+
 
 @dataclass
 class ArtifactDiff:
@@ -134,7 +137,52 @@ def diff_artifact_by_label(
             return None
         return diff_artifact(old_art, new_art)
 
-    # No previous build dir — check for version history
+    # No previous build dir — try snapshot-era ref lookup.
+    # Find the previous run ref whose OID differs from HEAD.
+    try:
+        ref_store = RefStore(synix_dir)
+        head_oid = ref_store.read_ref("HEAD")
+        if head_oid is not None:
+            run_refs = ref_store.iter_refs("refs/runs")
+            # Find runs whose OID differs from HEAD (i.e. previous runs)
+            prev_ref: str | None = None
+            # iter_refs returns sorted ascending; walk in reverse to find
+            # the most recent run that is NOT the current HEAD.
+            for ref_name, oid in reversed(run_refs):
+                if oid != head_oid:
+                    prev_ref = ref_name
+                    break
+
+            if prev_ref is not None:
+                prev_view = SnapshotView.open(synix_dir, ref=prev_ref)
+                try:
+                    prev_data = prev_view.get_artifact(label)
+                except KeyError:
+                    prev_data = None
+
+                if prev_data is not None:
+                    from datetime import datetime
+
+                    created_at_str = prev_data.get("metadata", {}).get("created_at")
+                    created_at = (
+                        datetime.fromisoformat(created_at_str) if created_at_str else datetime.now()
+                    )
+                    old_art = Artifact(
+                        label=prev_data["label"],
+                        artifact_type=prev_data["artifact_type"],
+                        artifact_id=prev_data["artifact_id"],
+                        input_ids=prev_data.get("input_ids", []),
+                        prompt_id=prev_data.get("prompt_id"),
+                        model_config=prev_data.get("model_config"),
+                        created_at=created_at,
+                        content=prev_data["content"],
+                        metadata=prev_data.get("metadata", {}),
+                    )
+                    return diff_artifact(old_art, new_art)
+    except (ValueError, FileNotFoundError, KeyError):
+        logger.debug("Snapshot-era ref lookup failed for diff of %r", label, exc_info=True)
+
+    # Legacy fallback: check for build/versions/<label> directories
     versions_dir = Path(build_dir) / "versions" / label
     if not versions_dir.exists():
         return None

src/synix/build/pipeline.py

@@ -42,11 +42,16 @@ def load_pipeline(path: str) -> Pipeline:
         raise ValueError(f"Pipeline module {path} must define a 'pipeline' variable")
     if not isinstance(pipeline, Pipeline):
         raise TypeError(f"'pipeline' variable must be a Pipeline instance, got {type(pipeline)}")
+
+    # Resolve relative paths against the pipeline file location
+    pipeline_parent = filepath.parent
+    if not Path(pipeline.source_dir).is_absolute():
+        pipeline.source_dir = str((pipeline_parent / pipeline.source_dir).resolve())
+    if not Path(pipeline.build_dir).is_absolute():
+        pipeline.build_dir = str((pipeline_parent / pipeline.build_dir).resolve())
+
     if pipeline.synix_dir is None:
-        build_dir = Path(pipeline.build_dir)
-        if not build_dir.is_absolute():
-            build_dir = (filepath.parent / build_dir).resolve()
-        pipeline.synix_dir = str(synix_dir_for_build_dir(build_dir))
+        pipeline.synix_dir = str(synix_dir_for_build_dir(Path(pipeline.build_dir)))
 
     validate_pipeline(pipeline)
     return pipeline

src/synix/build/plan.py

@@ -315,9 +315,20 @@ def _plan_source_layer(
 
     try:
         artifacts = layer.load(source_config)
-    except Exception:
-        # If loading fails (e.g., source_dir doesn't exist), report 0 artifacts
-        artifacts = []
+    except Exception as exc:
+        layer_artifacts[layer.name] = []
+        return StepPlan(
+            name=layer.name,
+            level=layer._level,
+            status="error",
+            artifact_count=0,
+            rebuild_count=0,
+            cached_count=0,
+            estimated_llm_calls=0,
+            estimated_tokens=0,
+            estimated_cost=0.0,
+            reason=f"source load failed: {exc}",
+        )
 
     layer_artifacts[layer.name] = artifacts
     artifact_count = len(artifacts)
@@ -424,7 +435,10 @@ def _plan_transform_layer(
         estimated_count = layer.estimate_output_count(len(inputs))
 
         # Store placeholder artifacts for downstream planning
-        layer_artifacts[layer.name] = inputs  # approximate
+        layer_artifacts[layer.name] = [
+            Artifact(label=f"__plan__{layer.name}_{i}", artifact_type="placeholder", content="")
+            for i in range(estimated_count)
+        ]
 
         return StepPlan(
             name=layer.name,
@@ -508,7 +522,13 @@ def _plan_transform_layer(
         reason = "new"
 
     # Store existing artifacts for downstream planning
-    layer_artifacts[layer.name] = existing if existing else inputs
+    if existing:
+        layer_artifacts[layer.name] = existing
+    else:
+        layer_artifacts[layer.name] = [
+            Artifact(label=f"__plan__{layer.name}_{i}", artifact_type="placeholder", content="")
+            for i in range(total_count)
+        ]
 
     return StepPlan(
         name=layer.name,

src/synix/build/runner.py

@@ -168,9 +168,8 @@ def run(
 
             try:
                 artifacts = layer.load(source_config)
-            except Exception:
-                logger.warning("Source %s failed to load", layer.name, exc_info=True)
-                artifacts = []
+            except Exception as exc:
+                raise RuntimeError(f"Source '{layer.name}' failed to load: {exc}") from exc
 
             # Record source artifacts in snapshot
             for artifact in artifacts:

src/synix/cli/artifact_commands.py

@@ -58,9 +58,9 @@ def list_artifacts(layer: str | None, build_dir: str, synix_dir: str | None, ref
 
     try:
         view = SnapshotView.open(resolved_synix_dir, ref=ref)
-    except ValueError:
-        console.print("[dim]No artifacts found.[/dim]")
-        return
+    except ValueError as e:
+        console.print(f"[red]Cannot open snapshot:[/red] {e}")
+        sys.exit(1)
 
     artifacts = view.list_artifacts()
 

src/synix/cli/build_commands.py

@@ -4,6 +4,7 @@
 
 import sys
 import time
+from pathlib import Path
 
 import click
 from rich import box
@@ -162,9 +163,10 @@ def build(
         sys.exit(1)
 
     if source_dir:
-        pipeline.source_dir = source_dir
+        pipeline.source_dir = str(Path(source_dir).resolve())
     if build_dir:
-        pipeline.build_dir = build_dir
+        pipeline.build_dir = str(Path(build_dir).resolve())
+        pipeline.synix_dir = None  # Force recomputation from overridden build_dir
 
     concurrency_label = f"{concurrency} threads" if concurrency > 1 else "sequential"
     console.print(
@@ -428,9 +430,10 @@ def plan(
         sys.exit(1)
 
     if source_dir:
-        pipeline.source_dir = source_dir
+        pipeline.source_dir = str(Path(source_dir).resolve())
     if build_dir:
-        pipeline.build_dir = build_dir
+        pipeline.build_dir = str(Path(build_dir).resolve())
+        pipeline.synix_dir = None  # Force recomputation from overridden build_dir
 
     try:
         build_plan = plan_build(pipeline, source_dir=source_dir)
@@ -464,6 +467,7 @@ def plan(
         "cached": "cyan",
         "rebuild": "yellow",
         "new": "green",
+        "error": "red",
     }
 
     # Model label for root
@@ -663,8 +667,6 @@ def _display_source_change_warnings(build_plan, pipeline):
     if not build_dir:
         return
 
-    from pathlib import Path
-
     from synix.build.refs import synix_dir_for_build_dir
     from synix.build.snapshot_view import SnapshotArtifactCache
 

src/synix/cli/clean_commands.py

@@ -74,3 +74,13 @@ def clean(build_dir: str, synix_dir: str | None, release_name: str | None, yes:
     for name, path in targets:
         shutil.rmtree(path)
         console.print(f"[green]Cleaned:[/green] {name} ({path})")
+
+    # Clean corresponding release refs
+    if release_name:
+        ref_file = sd / "refs" / "releases" / release_name
+        if ref_file.exists():
+            ref_file.unlink()
+    else:
+        release_refs_dir = sd / "refs" / "releases"
+        if release_refs_dir.exists():
+            shutil.rmtree(release_refs_dir)