Store parquet blob refs as basenames; honor DISK_TREE_ROOT

Scan blobs are stored as basenames (`<uuid>.parquet`) in SQLite's
`scan.blob` and in the `child_scan_id` column inside chunked parquets,
so the on-disk tree is portable across machines and `DISK_TREE_ROOT`
locations.

Storage backends (`hybrid`, `parquet`) gain a `_resolve()` helper that
treats absolute refs as legacy and resolves basenames against
`SCANS_DIR`. Server gets a top-level `resolve_blob()` for the same.

Adds `disk-tree migrate-blobs` to convert existing absolute refs (both
in SQLite and inside parquet `child_scan_id` columns) to basenames.

Also: `DISK_TREE_ROOT` env var now drives the root data directory (was
hard-coded to `~/.config/disk-tree`).

Author: ryan-williams

src/disk_tree/cli/index.py

@@ -65,8 +65,11 @@ def index(
     if scan.error_count:
         summary += f", {scan.error_count} permission errors"
     print(summary)
-    stat = os.stat(scan.blob)
-    print(f"Scan cached path: {scan.blob} ({iec(stat.st_size)})")
+    from os.path import isabs, join
+    from disk_tree.config import SCANS_DIR
+    blob_path = scan.blob if isabs(scan.blob) else join(SCANS_DIR, scan.blob)
+    stat = os.stat(blob_path)
+    print(f"Scan cached path: {blob_path} ({iec(stat.st_size)})")
     if scan.error_count:
         import json
         error_paths = json.loads(scan.error_paths) if scan.error_paths else []

src/disk_tree/cli/migrate.py

@@ -1,6 +1,6 @@
 """Migration commands for disk-tree database."""
 from os import makedirs, rename
-from os.path import basename, isfile, join
+from os.path import basename, isabs, isfile, join
 
 import pandas as pd
 from click import command, option
@@ -251,3 +251,91 @@ def migrate_hybrid(dry_run: bool):
     else:
         err(f"Migration complete: {migrated} migrated ({chunked} chunked), {skipped} skipped, {errors} errors")
         err(f"Originals backed up to: {backup_dir}")
+
+
+@cli.command('migrate-blobs')
+@option('-n', '--dry-run', is_flag=True, help="Show what would be done without making changes")
+def migrate_blobs(dry_run: bool):
+    """Convert absolute parquet blob refs to basenames (relative to SCANS_DIR).
+
+    Rewrites:
+    - scan.blob column in SQLite
+    - child_scan_id column inside hybrid parquet files
+    """
+    import sqlite3
+
+    if not isfile(DB_PATH):
+        err(f"Database not found: {DB_PATH}")
+        return
+
+    conn = sqlite3.connect(DB_PATH)
+    conn.row_factory = sqlite3.Row
+    cursor = conn.cursor()
+
+    cursor.execute("SELECT id, blob FROM scan")
+    rows = cursor.fetchall()
+    db_updated = 0
+    db_skipped = 0
+    for row in rows:
+        blob = row['blob']
+        if not isabs(blob):
+            db_skipped += 1
+            continue
+        new_blob = basename(blob)
+        if dry_run:
+            err(f"  scan.id={row['id']}: {blob} -> {new_blob}")
+        else:
+            cursor.execute("UPDATE scan SET blob = ? WHERE id = ?", (new_blob, row['id']))
+        db_updated += 1
+    if not dry_run:
+        conn.commit()
+    err(f"SQLite scan rows: {db_updated} updated, {db_skipped} already relative")
+
+    # Rewrite child_scan_id inside each parquet
+    cursor.execute("SELECT id, blob FROM scan")
+    rows = cursor.fetchall()
+    counts = {'updated': 0, 'skipped': 0, 'errors': 0}
+    for row in rows:
+        blob = row['blob']
+        blob_path = blob if isabs(blob) else join(SCANS_DIR, blob)
+        _normalize_parquet_chunks(blob_path, dry_run, counts)
+
+    conn.close()
+    err(f"Parquet files: {counts['updated']} rewritten, {counts['skipped']} unchanged, {counts['errors']} errors")
+
+
+def _normalize_parquet_chunks(blob_path: str, dry_run: bool, counts: dict) -> None:
+    """Rewrite child_scan_id column to basenames; recurse into referenced chunks."""
+    if not isfile(blob_path):
+        err(f"  Parquet not found: {blob_path}")
+        counts['errors'] += 1
+        return
+    try:
+        df = pd.read_parquet(blob_path)
+    except Exception as e:
+        err(f"  Error reading {blob_path}: {e}")
+        counts['errors'] += 1
+        return
+
+    if 'child_scan_id' not in df.columns:
+        counts['skipped'] += 1
+        return
+
+    refs = df['child_scan_id'].dropna()
+    abs_refs = refs[refs.apply(isabs)]
+    if abs_refs.empty:
+        counts['skipped'] += 1
+    else:
+        if dry_run:
+            err(f"  Would rewrite {len(abs_refs)} refs in {basename(blob_path)}")
+        else:
+            df['child_scan_id'] = df['child_scan_id'].apply(
+                lambda v: basename(v) if isinstance(v, str) and isabs(v) else v
+            )
+            df.to_parquet(blob_path, index=False)
+        counts['updated'] += 1
+
+    # Recurse into chunk parquets (resolve via basename in case still abs-on-disk)
+    for child_ref in df['child_scan_id'].dropna():
+        child_path = child_ref if isabs(child_ref) else join(SCANS_DIR, basename(child_ref))
+        _normalize_parquet_chunks(child_path, dry_run, counts)

src/disk_tree/config.py

@@ -1,10 +1,11 @@
 from os import environ as env, makedirs, walk
-from os.path import join, exists
+from os.path import join, exists, expanduser
 
 DISK_TREE_ROOT_VAR = 'DISK_TREE_ROOT'
 HOME = env['HOME']
 CONFIG_DIR = join(HOME, '.config')
-ROOT_DIR = join(CONFIG_DIR, 'disk-tree')
+DEFAULT_ROOT_DIR = join(CONFIG_DIR, 'disk-tree')
+ROOT_DIR = expanduser(env.get(DISK_TREE_ROOT_VAR, DEFAULT_ROOT_DIR))
 
 if not exists(ROOT_DIR):
     makedirs(ROOT_DIR)

src/disk_tree/server.py

@@ -1,6 +1,6 @@
 import json
 from os import listdir, makedirs, remove, stat
-from os.path import abspath, dirname, exists, isdir, isfile, join
+from os.path import abspath, dirname, exists, isabs, isdir, isfile, join
 import shutil
 import sqlite3
 import subprocess
@@ -13,9 +13,19 @@
 from flask import Flask, jsonify, redirect, request, g, Response, send_from_directory, url_for
 from flask_cors import CORS
 
-from disk_tree.config import SQLITE_PATH
+from disk_tree.config import SCANS_DIR, SQLITE_PATH
 from disk_tree.storage import get_backend
 
+
+def resolve_blob(blob_ref: str) -> str:
+    """Resolve a parquet blob ref to its absolute path.
+
+    Honors legacy absolute refs and ignores DuckDB/SQLite opaque refs.
+    """
+    if not blob_ref or blob_ref.startswith(('ddb:', 'sqlite:')):
+        return blob_ref
+    return blob_ref if isabs(blob_ref) else join(SCANS_DIR, blob_ref)
+
 app = Flask(__name__)
 CORS(app)
 
@@ -116,7 +126,7 @@ def resolve_chunk_for_path(blob_ref: str, rel_path: str) -> tuple[str, str]:
         return blob_ref, rel_path
 
     # Load the root parquet to check for child_scan_id
-    df = pd.read_parquet(blob_ref)
+    df = pd.read_parquet(resolve_blob(blob_ref))
     if 'child_scan_id' not in df.columns:
         return blob_ref, rel_path
 
@@ -130,7 +140,7 @@ def resolve_chunk_for_path(blob_ref: str, rel_path: str) -> tuple[str, str]:
             if pd.notna(row.get('child_scan_id')):
                 # This ancestor is chunked - resolve to the chunk
                 chunk_ref = row['child_scan_id']
-                if exists(chunk_ref):
+                if exists(resolve_blob(chunk_ref)):
                     # Rebase the remaining path relative to chunk root
                     remaining = '/'.join(parts[i+1:]) if i + 1 < len(parts) else '.'
                     # Recursively resolve in case of nested chunks
@@ -421,8 +431,8 @@ def get_scan():
         uri = '/'
 
     # Find the best matching scan (exact match or ancestor)
-    if uri.startswith('s3://'):
-        search_path = uri
+    if '://' in uri:
+        search_path = uri  # s3://, ssh://, etc.
     else:
         search_path = uri if uri.startswith('/') else f'/{uri}'
 
@@ -443,6 +453,7 @@ def get_scan():
     else:
         # Find the most recent scan covering this path (exact or ancestor)
         # Collect all candidate scans, then pick the freshest
+        from disk_tree.backends import url_parent
         candidate_scans = []
         test_path = search_path
         while test_path:
@@ -453,12 +464,8 @@ def get_scan():
             row = cursor.fetchone()
             if row:
                 candidate_scans.append(dict(row))
-            # Go up one directory
-            if test_path == '/' or test_path == 's3://':
-                break
-            parent = dirname(test_path)
-            # Avoid infinite loop: if dirname didn't change, we're at root
-            if parent == test_path:
+            parent = url_parent(test_path)
+            if parent is None or parent == test_path:
                 break
             test_path = parent
 
@@ -809,9 +816,9 @@ def row_to_dict(row):
         child_scan_dfs = []
         for _, row in direct_children_df.iterrows():
             child_scan = row.get('child_scan_id')
-            if pd.notna(child_scan) and exists(child_scan):
+            if pd.notna(child_scan) and exists(resolve_blob(child_scan)):
                 try:
-                    child_df = pd.read_parquet(child_scan)
+                    child_df = pd.read_parquet(resolve_blob(child_scan))
                     # Only load direct children (depth=1) from child scans
                     # These become depth=2 in the parent context
                     child_df = child_df[child_df['depth'] == 1]
@@ -1688,7 +1695,7 @@ def delete_path():
                                 df.loc[mask, 'n_children'] = df.loc[mask, 'n_children'] - 1
 
                     # Rewrite parquet (this is the expensive part)
-                    df.to_parquet(blob_ref, index=False)
+                    df.to_parquet(resolve_blob(blob_ref), index=False)
 
                     # Update denormalized stats in SQLite scan metadata
                     root_row = df[df['path'] == '.']