Discover and update modules declared by app behaviors

AGENTS.md

@@ -105,6 +105,14 @@ Module cache force-refresh identity is Foundation's exact URL + ref cache key,
 never a semantic bundle name or module entry-point ID. Verify it with
 `uv run pytest tests/test_module_cache.py`.
 
+Ordinary updates must walk every cached root/app manifest before repository-row
+deduplication, then check all five declared module kinds with runtime source
+override precedence. Preserve exact module URIs (ref, `.git`, subdirectory);
+refresh/install them through Foundation's selective module update, never load
+them as bundles. Missing manifests are incomplete discovery, not current.
+Convert explicit file URIs to native paths with one decode; keep ordinary paths literal.
+Keep `--check-only --force` read-only; run `tests/test_declared_module_updates.py`.
+
 ## Persisted reminder display
 
 `ui.is_displayable_session_message()` is display-only: history and replay hide

amplifier_app_cli/utils/file_uri.py

@@ -0,0 +1,51 @@
+"""Convert explicit file URIs to native paths without filesystem access."""
+
+import nturl2path
+from urllib.parse import unquote, urlsplit
+
+
+def file_uri_path_value(uri: str, *, windows: bool) -> str | None:
+    """Return a local filename from a file URI for the named platform.
+
+    ``Path.as_uri()`` emits ``file:///C:/...`` on Windows, while the legacy
+    CLI interpolation emits ``file://C:\\...``.  Both forms name a local drive
+    there.  A named host is a UNC path only on Windows; on POSIX it is remote
+    and deliberately not reinterpreted as a local path.
+    """
+
+    try:
+        parsed = urlsplit(uri)
+    except ValueError:
+        return None
+    if parsed.scheme.lower() != "file":
+        return None
+
+    netloc = parsed.netloc
+    if not netloc or netloc.lower() == "localhost":
+        encoded_path = parsed.path
+    elif windows and len(netloc) >= 2 and netloc[1] == ":" and netloc[0].isalpha():
+        # Legacy ``file://C:\path`` serialization.  ``nturl2path`` accepts
+        # both backslash and slash separators after the drive.
+        encoded_path = netloc + parsed.path
+    elif windows and "@" not in netloc and ":" not in netloc:
+        # Standard UNC syntax: file://server/share/path -> \\server\share\path.
+        try:
+            hostname = parsed.hostname
+        except ValueError:
+            return None
+        if not hostname or hostname.lower() != netloc.lower():
+            return None
+        if not parsed.path.strip("/"):
+            return None
+        encoded_path = f"//{hostname}{parsed.path}"
+    else:
+        return None
+
+    # urllib.request.url2pathname follows the host OS. Use percent-decoding
+    # explicitly for POSIX so the requested platform, not the host, wins.
+    converter = nturl2path.url2pathname if windows else unquote
+    try:
+        path_value = converter(encoded_path)
+    except (OSError, ValueError):
+        return None
+    return path_value if path_value and "\x00" not in path_value else None

amplifier_app_cli/utils/include_graph.py

@@ -39,13 +39,18 @@
 from __future__ import annotations
 
 import logging
+import os
+from collections.abc import AsyncIterator
 from dataclasses import dataclass
+from dataclasses import replace
 from pathlib import Path
 from typing import TYPE_CHECKING
 from typing import Mapping
 
+from .file_uri import file_uri_path_value
+
 if TYPE_CHECKING:
-    from amplifier_foundation import BundleRegistry
+    from amplifier_foundation import Bundle, BundleRegistry
     from amplifier_foundation.sources.protocol import SourceStatus
 
 logger = logging.getLogger(__name__)
@@ -143,7 +148,15 @@ def _local_path_for(uri: str, cache_dir: Path) -> Path | None:
         return active if active.exists() else None
 
     if parsed.is_file:
-        path = Path(parsed.path)
+        if uri.lower().startswith("file:"):
+            native = file_uri_path_value(uri, windows=os.name == "nt")
+            if native is None:
+                return None
+            path = Path(native)
+        else:
+            path = Path(parsed.path)
+        if parsed.subpath:
+            path = path / parsed.subpath
         return path if path.exists() else None
 
     return None
@@ -153,33 +166,26 @@ def _is_git_uri(uri: str) -> bool:
     return uri.startswith("git+") or uri.startswith("git://")
 
 
-async def collect_transitive_git_sources(
+async def _iter_cached_bundles(
     roots: Mapping[str, str],
     *,
-    registry: "BundleRegistry",
+    registry: BundleRegistry,
     cache_dir: Path,
-    known_uris: set[str] | None = None,
     max_depth: int = 16,
-) -> dict[str, TransitiveSource]:
-    """Walk ``includes:`` from *roots* and return the git sources found.
-
-    Args:
-        roots: Display name -> configured URI for every direct update target.
-        registry: Loader registry -- supplies include parsing/resolution and
-            the bundle-file parser.  Namespace includes resolve against its
-            state, so pass the same registry the CLI builds elsewhere.
-        cache_dir: Cache root (``<amplifier home>/cache``).
-        known_uris: Fragment-stripped URIs already covered by a direct row.
-            Anything here is skipped -- it is not "transitive", it is the
-            source that was already going to be checked.
-        max_depth: Belt-and-braces bound on graph depth.  Cycles are already
-            handled by the visited set; this only caps a pathological chain.
-
-    Returns:
-        Fragment-stripped URI -> TransitiveSource, in discovery order.
+    failures: dict[str, str] | None = None,
+) -> AsyncIterator[tuple[str, str, str, Bundle | None]]:
+    """Yield root, immediate parent, full URI and cached manifest (if readable).
+
+    Both collectors use this bounded, read-only walk. A missing manifest still
+    yields its URI so include status checking retains its existing behavior.
+    Failure text deliberately contains neither configured URIs nor exceptions.
     """
-    known = set(known_uris or set())
-    found: dict[str, TransitiveSource] = {}
+    from amplifier_foundation.paths.resolution import parse_uri
+
+    def failed(root: str) -> None:
+        if failures is not None:
+            failures[root] = "Incomplete cached module discovery"
+        logger.debug("Could not completely read cached bundle declarations")
 
     # Visited on the FULL uri (fragment included): two behaviors in one repo
     # are different files with different includes, so collapsing them on the
@@ -197,59 +203,145 @@ async def collect_transitive_git_sources(
     while queue:
         root_label, label, uri, depth = queue.pop(0)
         if depth >= max_depth:
-            logger.debug(f"Include walk hit max depth at: {uri}")
-            continue
-
-        path = _local_path_for(uri, cache_dir)
-        if path is None:
+            failed(root_label)
+            yield root_label, label, uri, None
             continue
 
         try:
+            path = _local_path_for(uri, cache_dir)
+            if path is None:
+                failed(root_label)
+                yield root_label, label, uri, None
+                continue
             bundle = await registry._load_from_path(path)
-        except Exception as exc:  # noqa: BLE001 - an unreadable bundle is not fatal
-            logger.debug(f"Could not read includes from {uri}: {exc}")
+        except Exception:  # noqa: BLE001 - an unreadable bundle is not fatal
+            failed(root_label)
+            yield root_label, label, uri, None
             continue
 
+        yield root_label, label, uri, bundle
         parent_label = bundle.name or label
 
         for raw_include in bundle.includes or []:
-            spec = registry._parse_include(raw_include)
-            if not spec:
-                continue
-
             try:
+                spec = registry._parse_include(raw_include)
+                if not spec:
+                    failed(root_label)
+                    continue
                 resolved = registry._resolve_include_source(spec)
-            except Exception as exc:  # noqa: BLE001
-                logger.debug(f"Could not resolve include '{spec}': {exc}")
-                continue
-
-            if not resolved:
-                continue
-
-            # A plain name is a registry alias; the loader lets _load_single
-            # look it up, so look it up the same way here.
-            if "://" not in resolved and not resolved.startswith("git+"):
-                looked_up = registry.find(resolved)
-                if not looked_up:
+                if not resolved:
+                    failed(root_label)
                     continue
-                resolved = looked_up
+                # A plain name is a registry alias, not a filesystem source.
+                if (
+                    "://" not in resolved
+                    and not resolved.startswith("git+")
+                    and not parse_uri(resolved).is_file
+                ):
+                    resolved = registry.find(resolved)
+                    if not resolved:
+                        failed(root_label)
+                        continue
+            except Exception:  # noqa: BLE001
+                failed(root_label)
+                continue
 
             if resolved in visited:
                 continue
             visited.add(resolved)
+            queue.append((root_label, parent_label, resolved, depth + 1))
 
-            if _is_git_uri(resolved):
-                key = strip_uri_fragment(resolved)
-                if key not in known and key not in found:
-                    found[key] = TransitiveSource(
-                        uri=key, parent=parent_label, root=root_label, via=resolved
-                    )
 
-            queue.append((root_label, parent_label, resolved, depth + 1))
+async def collect_transitive_git_sources(
+    roots: Mapping[str, str],
+    *,
+    registry: "BundleRegistry",
+    cache_dir: Path,
+    known_uris: set[str] | None = None,
+    max_depth: int = 16,
+) -> dict[str, TransitiveSource]:
+    """Return fragment-stripped included git sources, excluding direct targets.
 
+    Visit full URIs (including fragments) and reuse registry include resolution.
+    Missing caches are never downloaded; their include URIs are still returned.
+    """
+    known = set(known_uris or set())
+    root_uris = set(roots.values())
+    found: dict[str, TransitiveSource] = {}
+    async for root, parent, uri, _ in _iter_cached_bundles(
+        roots, registry=registry, cache_dir=cache_dir, max_depth=max_depth
+    ):
+        if uri in root_uris or not _is_git_uri(uri):
+            continue
+        key = strip_uri_fragment(uri)
+        if key not in known and key not in found:
+            found[key] = TransitiveSource(uri=key, parent=parent, root=root, via=uri)
     return found
 
 
+@dataclass(frozen=True)
+class DeclaredModuleSource:
+    """First declaration plus replaced URIs, so legacy caches honor overrides."""
+
+    module_id: str
+    declared_by: str
+    root: str
+    overridden_sources: tuple[str, ...] = ()
+
+
+async def collect_declared_module_sources(
+    roots: Mapping[str, str],
+    *,
+    registry: BundleRegistry,
+    cache_dir: Path,
+    overrides: Mapping[str, str] | None = None,
+) -> tuple[dict[str, DeclaredModuleSource], dict[str, str]]:
+    """Return exact effective URI -> declaration metadata, and root failures.
+
+    Inspect the same five module locations as Foundation's source collector.
+    Module sources are declarations, never bundle manifests to walk. First
+    provenance wins when several module IDs share the same effective source.
+    """
+    found: dict[str, DeclaredModuleSource] = {}
+    failures: dict[str, str] = {}
+    overrides = overrides or {}
+    async for root, parent, _, bundle in _iter_cached_bundles(
+        roots, registry=registry, cache_dir=cache_dir, failures=failures
+    ):
+        if bundle is None:
+            continue
+        session = bundle.session or {}
+        declarations = [
+            session.get("orchestrator"),
+            session.get("context"),
+            *(bundle.providers or []),
+            *(bundle.tools or []),
+            *(bundle.hooks or []),
+        ]
+        for declaration in declarations:
+            if not isinstance(declaration, dict):
+                continue
+            module_id = declaration.get("module")
+            if not isinstance(module_id, str):
+                continue
+            uri = overrides.get(module_id, declaration.get("source"))
+            if not isinstance(uri, str) or not uri:
+                continue
+            entry = found.setdefault(
+                uri, DeclaredModuleSource(module_id, bundle.name or parent, root)
+            )
+            original = declaration.get("source")
+            if (
+                isinstance(original, str)
+                and original != uri
+                and original not in entry.overridden_sources
+            ):
+                found[uri] = replace(
+                    entry, overridden_sources=(*entry.overridden_sources, original)
+                )
+    return found, failures
+
+
 async def check_transitive_sources(
     sources: Mapping[str, TransitiveSource],
     *,
@@ -264,6 +356,7 @@ async def check_transitive_sources(
     """
     from amplifier_foundation.paths.resolution import parse_uri
     from amplifier_foundation.sources.git import GitSourceHandler
+    from amplifier_foundation.sources.protocol import SourceStatus
 
     git_handler = GitSourceHandler()
     results: dict[str, TransitiveStatus] = {}
@@ -274,9 +367,14 @@ async def check_transitive_sources(
             if not git_handler.can_handle(parsed):
                 continue
             status = await git_handler.get_status(parsed, cache_dir)
-        except Exception as exc:  # noqa: BLE001 - one bad source must not sink the report
-            logger.debug(f"Status check failed for {source.uri}: {exc}")
-            continue
+        except Exception:  # noqa: BLE001 - one bad source must not sink the report
+            logger.debug("Transitive source status check failed")
+            status = SourceStatus(
+                source_uri=source.uri,
+                is_cached=False,
+                has_update=None,
+                error="status check failed",
+            )
         results[key] = TransitiveStatus(source=source, status=status)
 
     return results
@@ -312,9 +410,11 @@ async def refresh_transitive_source(uri: str, *, cache_dir: Path) -> None:
 
 
 __all__ = [
+    "DeclaredModuleSource",
     "TransitiveSource",
     "TransitiveStatus",
     "check_transitive_sources",
+    "collect_declared_module_sources",
     "collect_transitive_git_sources",
     "refresh_transitive_source",
     "strip_uri_fragment",