diff --git a/.github/copilot-instructions.md b/.github/copilot-instructions.md
index 147f496c2..40236b683 100644
--- a/.github/copilot-instructions.md
+++ b/.github/copilot-instructions.md
@@ -22,4 +22,5 @@ The architectural decisions and basis for the project in that document are only
 - The philosophy when architecting and implementing the project is to prime speed and simplicity over complexity. Do NOT over-engineer, but rather build a solid foundation that can be iterated on.
 - APM is an active OSS project under the `microsoft` org with a growing community (250+ stars, external contributors). Breaking changes should be communicated clearly (CHANGELOG.md), but we still favor shipping fast over lengthy deprecation cycles.
 - The goal is to deliver a solid and scalable architecture but simple starting implementation. Not building something complex from the start and then having to simplify it later. Remember we are delivering a new tool to the developer community and we will need to rapidly adapt to what's really useful, evolving standards, etc.
-- **Cross-platform encoding rule**: All source code and CLI output must stay within printable ASCII (U+0020–U+007E). Do NOT use emojis, Unicode symbols, box-drawing characters, em dashes, or any character outside the ASCII range in source files or CLI output strings. Use bracket notation for status symbols: `[+]` success, `[!]` warning, `[x]` error, `[i]` info, `[*]` action, `[>]` running. This is required to prevent `charmap` codec errors on Windows cp1252 terminals.
\ No newline at end of file
+- **Cross-platform encoding rule**: All source code and CLI output must stay within printable ASCII (U+0020–U+007E). Do NOT use emojis, Unicode symbols, box-drawing characters, em dashes, or any character outside the ASCII range in source files or CLI output strings. Use bracket notation for status symbols: `[+]` success, `[!]` warning, `[x]` error, `[i]` info, `[*]` action, `[>]` running. This is required to prevent `charmap` codec errors on Windows cp1252 terminals.
+- **Path safety rule**: Any code that builds filesystem paths from user input or external data (marketplace names, plugin paths, lockfile entries, bundle contents) **must** use the centralized guards in `src/apm_cli/utils/path_security.py`. Use `validate_path_segments(value, context=)` at parse time to reject traversal sequences (`.`, `..`) with cross-platform backslash normalization, and `ensure_path_within(path, base_dir)` after resolution to assert containment (resolves symlinks). Never write ad-hoc `".." in x` checks.
\ No newline at end of file
diff --git a/.gitignore b/.gitignore
index a055ad570..2dd0f1a4e 100644
--- a/.gitignore
+++ b/.gitignore
@@ -72,3 +72,4 @@ apm_modules/
 build/tmp/
 scout-pipeline-result.png
 .copilot/
+.playwright-mcp/
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 6787971bf..7e0d900d9 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -8,6 +8,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+
+- Marketplace integration for plugin discovery and governance: `apm marketplace add/list/browse/update/remove` commands, `apm search QUERY@MARKETPLACE` scoped search, `apm install NAME@MARKETPLACE` syntax for installing plugins from marketplace registries; lockfile provenance fields `discovered_via` and `marketplace_plugin_name` to track marketplace origin; support for Copilot CLI and Claude Code `marketplace.json` formats with 4 source types (github, url, git-subdir, relative path) (#503)
 ### Changed
 
 - `apm deps update` now skips download and integration for packages whose resolved SHA matches the lockfile SHA, making the common "nothing changed" case near-instant (#495)
diff --git a/docs/src/content/docs/guides/marketplaces.md b/docs/src/content/docs/guides/marketplaces.md
new file mode 100644
index 000000000..e906e8b36
--- /dev/null
+++ b/docs/src/content/docs/guides/marketplaces.md
@@ -0,0 +1,155 @@
+---
+title: "Marketplaces"
+sidebar:
+  order: 5
+---
+
+Marketplaces are curated indexes of plugins hosted as GitHub repositories. Each marketplace contains a `marketplace.json` file that maps plugin names to source locations. APM resolves these entries to Git URLs, so plugins installed from marketplaces get the same version locking, security scanning, and governance as any other APM dependency.
+
+## How marketplaces work
+
+A marketplace is a GitHub repository with a `marketplace.json` at its root. The file lists plugins with their source type and location:
+
+```json
+{
+  "name": "Acme Plugins",
+  "plugins": [
+    {
+      "name": "code-review",
+      "description": "Automated code review agent",
+      "source": { "type": "github", "repo": "acme/code-review-plugin" }
+    },
+    {
+      "name": "style-guide",
+      "source": { "type": "url", "url": "https://github.com/acme/style-guide.git" }
+    },
+    {
+      "name": "eslint-rules",
+      "source": { "type": "git-subdir", "repo": "acme/monorepo", "subdir": "plugins/eslint-rules" }
+    },
+    {
+      "name": "local-tools",
+      "source": "./tools/local-plugin"
+    }
+  ]
+}
+```
+
+Both Copilot CLI and Claude Code `marketplace.json` formats are supported. Copilot CLI uses `"repository"` and `"ref"` fields; Claude Code uses `"source"` (string or object). APM normalizes entries from either format into its canonical dependency representation.
+
+### Supported source types
+
+| Type | Description | Example |
+|------|-------------|---------|
+| `github` | GitHub `owner/repo` shorthand | `acme/code-review-plugin` |
+| `url` | Full HTTPS or SSH Git URL | `https://github.com/acme/style-guide.git` |
+| `git-subdir` | Subdirectory within a Git repository (`repo` + `subdir`) | `acme/monorepo` + `plugins/eslint-rules` |
+| String `source` | Subdirectory within the marketplace repository itself | `./tools/local-plugin` |
+
+npm sources are not supported. Copilot CLI format uses `"repository"` and optional `"ref"` fields instead of `"source"`.
+
+## Register a marketplace
+
+```bash
+apm marketplace add acme/plugin-marketplace
+```
+
+This registers the marketplace and fetches its `marketplace.json`. By default APM tracks the `main` branch.
+
+**Options:**
+- `--name/-n` -- Custom display name for the marketplace
+- `--branch/-b` -- Branch to track (default: `main`)
+
+```bash
+# Register with a custom name on a specific branch
+apm marketplace add acme/plugin-marketplace --name acme-plugins --branch release
+```
+
+## List registered marketplaces
+
+```bash
+apm marketplace list
+```
+
+Shows all registered marketplaces with their source repository and branch.
+
+## Browse plugins
+
+View all plugins available in a specific marketplace:
+
+```bash
+apm marketplace browse acme-plugins
+```
+
+## Search a marketplace
+
+Search plugins by name or description in a specific marketplace using `QUERY@MARKETPLACE`:
+
+```bash
+apm search "code review@skills"
+```
+
+**Options:**
+- `--limit` -- Maximum results to return (default: 20)
+
+```bash
+apm search "linting@awesome-copilot" --limit 5
+```
+
+The `@MARKETPLACE` scope is required -- this avoids name collisions when different
+marketplaces contain plugins with the same name. To see everything in a marketplace,
+use `apm marketplace browse <name>` instead.
+
+## Install from a marketplace
+
+Use the `NAME@MARKETPLACE` syntax to install a plugin from a specific marketplace:
+
+```bash
+apm install code-review@acme-plugins
+```
+
+APM resolves the plugin name against the marketplace index, fetches the underlying Git repository, and installs it as a standard APM dependency. The resolved source appears in `apm.yml` and `apm.lock.yaml` just like any direct dependency.
+
+For full `apm install` options, see [CLI Commands](../../reference/cli-commands/).
+
+## Provenance tracking
+
+Marketplace-resolved plugins are tracked in `apm.lock.yaml` with full provenance:
+
+```yaml
+apm_modules:
+  acme/code-review-plugin:
+    resolved: https://github.com/acme/code-review-plugin#main
+    commit: abc123def456789
+    discovered_via: acme-plugins
+    marketplace_plugin_name: code-review
+```
+
+The `discovered_via` field records which marketplace was used for discovery. `marketplace_plugin_name` stores the original plugin name from the index. The `resolved` URL and `commit` pin the exact version, so builds remain reproducible regardless of marketplace availability.
+
+## Cache behavior
+
+APM caches marketplace indexes locally with a 1-hour TTL. Within that window, commands like `search` and `browse` use the cached index. After expiry, APM fetches a fresh copy from the network. If the network request fails, APM falls back to the expired cache (stale-if-error) so commands still work offline.
+
+Force a cache refresh:
+
+```bash
+# Refresh a specific marketplace
+apm marketplace update acme-plugins
+
+# Refresh all registered marketplaces
+apm marketplace update
+```
+
+## Manage marketplaces
+
+Remove a registered marketplace:
+
+```bash
+apm marketplace remove acme-plugins
+
+# Skip confirmation prompt
+apm marketplace remove acme-plugins --yes
+```
+
+Removing a marketplace does not uninstall plugins previously installed from it. Those plugins remain pinned in `apm.lock.yaml` to their resolved Git sources.
diff --git a/docs/src/content/docs/guides/plugins.md b/docs/src/content/docs/guides/plugins.md
index ec133d9f2..b3dbf3c59 100644
--- a/docs/src/content/docs/guides/plugins.md
+++ b/docs/src/content/docs/guides/plugins.md
@@ -340,11 +340,19 @@ Use the [hybrid authoring workflow](#hybrid-authoring-workflow) to develop plugi
 ## Finding Plugins
 
 Plugins can be found through:
+- **Marketplaces** -- curated `marketplace.json` indexes browsable with `apm marketplace browse` and searchable with `apm search QUERY@MARKETPLACE`. See the [Marketplaces guide](../marketplaces/) for setup.
 - GitHub repositories (search for repos with `plugin.json`)
 - Organization-specific plugin repositories
-- Community plugin collections
 
-Once found, install them using the standard `apm install owner/repo/plugin-name` command.
+Install by name from a registered marketplace:
+
+```bash
+apm install code-review@acme-plugins
+```
+
+APM resolves marketplace entries to Git URLs, so marketplace-installed plugins get full version locking, security scanning, and governance. See [Marketplaces](../marketplaces/) for details.
+
+For direct installs, use the standard `apm install owner/repo/plugin-name` command.
 
 ## Troubleshooting
 
diff --git a/docs/src/content/docs/reference/cli-commands.md b/docs/src/content/docs/reference/cli-commands.md
index c84f08098..65ad338b0 100644
--- a/docs/src/content/docs/reference/cli-commands.md
+++ b/docs/src/content/docs/reference/cli-commands.md
@@ -81,7 +81,7 @@ apm install [PACKAGES...] [OPTIONS]
 ```
 
 **Arguments:**
-- `PACKAGES` - Optional APM packages to add and install. Accepts shorthand (`owner/repo`), HTTPS URLs, SSH URLs, FQDN shorthand (`host/owner/repo`), or local filesystem paths (`./path`, `../path`, `/absolute/path`, `~/path`). All forms are normalized to canonical format in `apm.yml`.
+- `PACKAGES` - Optional APM packages to add and install. Accepts shorthand (`owner/repo`), HTTPS URLs, SSH URLs, FQDN shorthand (`host/owner/repo`), local filesystem paths (`./path`, `../path`, `/absolute/path`, `~/path`), or marketplace references (`NAME@MARKETPLACE`). All forms are normalized to canonical format in `apm.yml`.
 
 **Options:**
 - `--runtime TEXT` - Target specific runtime only (copilot, codex, vscode)
@@ -151,6 +151,9 @@ apm install --dev owner/test-helpers
 # Install from a local path (copies to apm_modules/_local/)
 apm install ./packages/my-shared-skills
 apm install /home/user/repos/my-ai-package
+
+# Install a plugin from a registered marketplace
+apm install code-review@acme-plugins
 ```
 
 **Auto-Bootstrap Behavior:**
@@ -832,6 +835,148 @@ apm mcp show a5e8a7f0-d4e4-4a1d-b12f-2896a23fd4f1
 - Available installation packages
 - Installation instructions
 
+### `apm marketplace` - Plugin marketplace management
+
+Register, browse, and manage plugin marketplaces. Marketplaces are GitHub repositories containing a `marketplace.json` index of plugins.
+
+> See the [Marketplaces guide](../../guides/marketplaces/) for concepts and workflows.
+
+```bash
+apm marketplace COMMAND [OPTIONS]
+```
+
+#### `apm marketplace add` - Register a marketplace
+
+Register a GitHub repository as a plugin marketplace.
+
+```bash
+apm marketplace add OWNER/REPO [OPTIONS]
+```
+
+**Arguments:**
+- `OWNER/REPO` - GitHub repository containing `marketplace.json`
+
+**Options:**
+- `-n, --name TEXT` - Custom display name for the marketplace
+- `-b, --branch TEXT` - Branch to track (default: main)
+- `-v, --verbose` - Show detailed output
+
+**Examples:**
+```bash
+# Register a marketplace
+apm marketplace add acme/plugin-marketplace
+
+# Register with a custom name and branch
+apm marketplace add acme/plugin-marketplace --name acme-plugins --branch release
+```
+
+#### `apm marketplace list` - List registered marketplaces
+
+List all registered marketplaces with their source repository and branch.
+
+```bash
+apm marketplace list [OPTIONS]
+```
+
+**Options:**
+- `-v, --verbose` - Show detailed output
+
+**Examples:**
+```bash
+apm marketplace list
+```
+
+#### `apm marketplace browse` - Browse marketplace plugins
+
+List all plugins available in a registered marketplace.
+
+```bash
+apm marketplace browse NAME [OPTIONS]
+```
+
+**Arguments:**
+- `NAME` - Name of the registered marketplace
+
+**Options:**
+- `-v, --verbose` - Show detailed output
+
+**Examples:**
+```bash
+# Browse all plugins in a marketplace
+apm marketplace browse acme-plugins
+```
+
+#### `apm marketplace update` - Refresh marketplace cache
+
+Refresh the cached `marketplace.json` for one or all registered marketplaces.
+
+```bash
+apm marketplace update [NAME] [OPTIONS]
+```
+
+**Arguments:**
+- `NAME` - Optional marketplace name. Omit to refresh all.
+
+**Options:**
+- `-v, --verbose` - Show detailed output
+
+**Examples:**
+```bash
+# Refresh a specific marketplace
+apm marketplace update acme-plugins
+
+# Refresh all marketplaces
+apm marketplace update
+```
+
+#### `apm marketplace remove` - Remove a registered marketplace
+
+Unregister a marketplace. Plugins previously installed from it remain pinned in `apm.lock.yaml`.
+
+```bash
+apm marketplace remove NAME [OPTIONS]
+```
+
+**Arguments:**
+- `NAME` - Name of the marketplace to remove
+
+**Options:**
+- `-y, --yes` - Skip confirmation prompt
+- `-v, --verbose` - Show detailed output
+
+**Examples:**
+```bash
+# Remove with confirmation prompt
+apm marketplace remove acme-plugins
+
+# Remove without confirmation
+apm marketplace remove acme-plugins --yes
+```
+
+### `apm search` - Search plugins in a marketplace
+
+Search for plugins by name or description within a specific marketplace.
+
+```bash
+apm search QUERY@MARKETPLACE [OPTIONS]
+```
+
+**Arguments:**
+- `QUERY@MARKETPLACE` - Search term scoped to a marketplace (e.g., `security@skills`)
+
+**Options:**
+- `--limit INTEGER` - Maximum results to return (default: 20)
+- `-v, --verbose` - Show detailed output
+
+**Examples:**
+```bash
+# Search for code review plugins in a marketplace
+apm search "code review@skills"
+
+# Limit results
+apm search "linting@awesome-copilot" --limit 5
+```
+
 ### `apm run` (Experimental) - Execute prompts
 
 Execute a script defined in your apm.yml with parameters and real-time output streaming.
diff --git a/src/apm_cli/cli.py b/src/apm_cli/cli.py
index 2a958159d..5263aafbd 100644
--- a/src/apm_cli/cli.py
+++ b/src/apm_cli/cli.py
@@ -22,6 +22,7 @@
 from apm_cli.commands.init import init
 from apm_cli.commands.install import install
 from apm_cli.commands.list_cmd import list as list_cmd
+from apm_cli.commands.marketplace import marketplace, search as marketplace_search
 from apm_cli.commands.mcp import mcp
 from apm_cli.commands.pack import pack_cmd, unpack_cmd
 from apm_cli.commands.prune import prune
@@ -69,6 +70,8 @@ def cli(ctx):
 cli.add_command(config)
 cli.add_command(runtime)
 cli.add_command(mcp)
+cli.add_command(marketplace)
+cli.add_command(marketplace_search, name="search")
 
 
 def _configure_encoding() -> None:
diff --git a/src/apm_cli/commands/install.py b/src/apm_cli/commands/install.py
index 674a71a2a..aaa1535c7 100644
--- a/src/apm_cli/commands/install.py
+++ b/src/apm_cli/commands/install.py
@@ -123,18 +123,64 @@ def _validate_and_add_packages_to_apm_yml(packages, dry_run=False, dev=False, lo
     # First, validate all packages
     valid_outcomes = []  # (canonical, already_present) tuples
     invalid_outcomes = []  # (package, reason) tuples
+    _marketplace_provenance = {}  # canonical -> {discovered_via, marketplace_plugin_name}
 
     if logger:
         logger.validation_start(len(packages))
 
     for package in packages:
-        # Validate package format (should be owner/repo, a git URL, or a local path)
+        # --- Marketplace pre-parse intercept ---
+        # If input has no slash and is not a local path, check if it is a
+        # marketplace ref (NAME@MARKETPLACE).  If so, resolve it to a
+        # canonical owner/repo[#ref] string before entering the standard
+        # parse path.  Anything that doesn't match is rejected as an
+        # invalid format.
+        marketplace_provenance = None
         if "/" not in package and not DependencyReference.is_local_path(package):
-            reason = "invalid format -- use 'owner/repo'"
-            invalid_outcomes.append((package, reason))
-            if logger:
-                logger.validation_fail(package, reason)
-            continue
+            try:
+                from ..marketplace.resolver import (
+                    parse_marketplace_ref,
+                    resolve_marketplace_plugin,
+                )
+
+                mkt_ref = parse_marketplace_ref(package)
+            except ImportError:
+                mkt_ref = None
+
+            if mkt_ref is not None:
+                plugin_name, marketplace_name = mkt_ref
+                try:
+                    if logger:
+                        logger.verbose_detail(
+                            f"    Resolving {plugin_name}@{marketplace_name} via marketplace..."
+                        )
+                    canonical_str, resolved_plugin = resolve_marketplace_plugin(
+                        plugin_name,
+                        marketplace_name,
+                        auth_resolver=auth_resolver,
+                    )
+                    if logger:
+                        logger.verbose_detail(
+                            f"    Resolved to: {canonical_str}"
+                        )
+                    marketplace_provenance = {
+                        "discovered_via": marketplace_name,
+                        "marketplace_plugin_name": plugin_name,
+                    }
+                    package = canonical_str
+                except Exception as mkt_err:
+                    reason = str(mkt_err)
+                    invalid_outcomes.append((package, reason))
+                    if logger:
+                        logger.validation_fail(package, reason)
+                    continue
+            else:
+                # No slash, not a local path, and not a marketplace ref
+                reason = "invalid format -- use 'owner/repo' or 'plugin-name@marketplace'"
+                invalid_outcomes.append((package, reason))
+                if logger:
+                    logger.validation_fail(package, reason)
+                continue
 
         # Canonicalize input
         try:
@@ -161,6 +207,8 @@ def _validate_and_add_packages_to_apm_yml(packages, dry_run=False, dev=False, lo
             if not already_in_deps:
                 validated_packages.append(canonical)
                 existing_identities.add(identity)  # prevent duplicates within batch
+            if marketplace_provenance:
+                _marketplace_provenance[identity] = marketplace_provenance
         else:
             reason = _local_path_failure_reason(dep_ref)
             if not reason:
@@ -171,7 +219,11 @@ def _validate_and_add_packages_to_apm_yml(packages, dry_run=False, dev=False, lo
             if logger:
                 logger.validation_fail(package, reason)
 
-    outcome = _ValidationOutcome(valid=valid_outcomes, invalid=invalid_outcomes)
+    outcome = _ValidationOutcome(
+        valid=valid_outcomes,
+        invalid=invalid_outcomes,
+        marketplace_provenance=_marketplace_provenance or None,
+    )
 
     # Let the logger emit a summary and decide whether to continue
     if logger:
@@ -692,14 +744,20 @@ def install(ctx, packages, runtime, exclude, only, update, dry_run, force, verbo
 
             try:
                 # If specific packages were requested, only install those
-                # Otherwise install all from apm.yml
-                only_pkgs = builtins.list(packages) if packages else None
+                # Otherwise install all from apm.yml.
+                # Use validated_packages (canonical strings) instead of
+                # raw packages (which may contain marketplace refs like
+                # NAME@MARKETPLACE that don't match resolved dep identities).
+                only_pkgs = builtins.list(validated_packages) if packages else None
                 install_result = _install_apm_dependencies(
                     apm_package, update, verbose, only_pkgs, force=force,
                     parallel_downloads=parallel_downloads,
                     logger=logger,
                     auth_resolver=auth_resolver,
                     target=target,
+                    marketplace_provenance=(
+                        outcome.marketplace_provenance if packages and outcome else None
+                    ),
                 )
                 apm_count = install_result.installed_count
                 prompt_count = install_result.prompts_integrated
@@ -1023,6 +1081,7 @@ def _install_apm_dependencies(
     logger: "InstallLogger" = None,
     auth_resolver: "AuthResolver" = None,
     target: str = None,
+    marketplace_provenance: dict = None,
 ):
     """Install APM package dependencies.
 
@@ -2198,6 +2257,12 @@ def _collect_descendants(node, visited=None):
                 for dep_key, locked_dep in lockfile.dependencies.items():
                     if dep_key in _package_hashes:
                         locked_dep.content_hash = _package_hashes[dep_key]
+                # Attach marketplace provenance if available
+                if marketplace_provenance:
+                    for dep_key, prov in marketplace_provenance.items():
+                        if dep_key in lockfile.dependencies:
+                            lockfile.dependencies[dep_key].discovered_via = prov.get("discovered_via")
+                            lockfile.dependencies[dep_key].marketplace_plugin_name = prov.get("marketplace_plugin_name")
                 # Selectively merge entries from the existing lockfile:
                 #   - For partial installs (only_packages): preserve all old entries
                 #     (sequential install — only the specified package was processed).
diff --git a/src/apm_cli/commands/marketplace.py b/src/apm_cli/commands/marketplace.py
new file mode 100644
index 000000000..a3d34fd1b
--- /dev/null
+++ b/src/apm_cli/commands/marketplace.py
@@ -0,0 +1,436 @@
+"""APM marketplace command group.
+
+Manages plugin marketplace discovery and governance. Follows the same
+Click group pattern as ``mcp.py``.
+"""
+
+import builtins
+import sys
+
+import click
+
+from ..core.command_logger import CommandLogger
+from ._helpers import _get_console
+
+# Restore builtins shadowed by subcommand names
+list = builtins.list
+
+
+@click.group(help="Manage plugin marketplaces for discovery and governance")
+def marketplace():
+    """Register, browse, and search plugin marketplaces."""
+    pass
+
+
+# ---------------------------------------------------------------------------
+# marketplace add
+# ---------------------------------------------------------------------------
+
+
+@marketplace.command(help="Register a plugin marketplace")
+@click.argument("repo", required=True)
+@click.option("--name", "-n", default=None, help="Display name (defaults to repo name)")
+@click.option("--branch", "-b", default="main", show_default=True, help="Branch to use")
+@click.option("--verbose", "-v", is_flag=True, help="Show detailed output")
+def add(repo, name, branch, verbose):
+    """Register a marketplace from OWNER/REPO."""
+    logger = CommandLogger("marketplace-add", verbose=verbose)
+    try:
+        from ..marketplace.client import _auto_detect_path, fetch_marketplace
+        from ..marketplace.models import MarketplaceSource
+        from ..marketplace.registry import add_marketplace
+
+        # Parse OWNER/REPO
+        if "/" not in repo:
+            logger.error(
+                f"Invalid format: '{repo}'. Use 'OWNER/REPO' "
+                f"(e.g., 'acme-org/plugin-marketplace')"
+            )
+            sys.exit(1)
+
+        parts = repo.split("/")
+        if len(parts) != 2 or not parts[0] or not parts[1]:
+            logger.error(f"Invalid format: '{repo}'. Expected 'OWNER/REPO'")
+            sys.exit(1)
+
+        owner, repo_name = parts[0], parts[1]
+        display_name = name or repo_name
+
+        # Validate name is identifier-compatible for NAME@MARKETPLACE syntax
+        import re
+
+        if not re.match(r"^[a-zA-Z0-9._-]+$", display_name):
+            logger.error(
+                f"Invalid marketplace name: '{display_name}'. "
+                f"Names must only contain letters, digits, '.', '_', and '-' "
+                f"(required for 'apm install plugin@marketplace' syntax)."
+            )
+            sys.exit(1)
+
+        logger.start(f"Registering marketplace '{display_name}'...", symbol="gear")
+        logger.verbose_detail(f"    Repository: {owner}/{repo_name}")
+        logger.verbose_detail(f"    Branch: {branch}")
+
+        # Auto-detect marketplace.json location
+        probe_source = MarketplaceSource(
+            name=display_name,
+            owner=owner,
+            repo=repo_name,
+            branch=branch,
+        )
+        detected_path = _auto_detect_path(probe_source)
+
+        if detected_path is None:
+            logger.error(
+                f"No marketplace.json found in '{owner}/{repo_name}'. "
+                f"Checked: marketplace.json, .github/plugin/marketplace.json, "
+                f".claude-plugin/marketplace.json"
+            )
+            sys.exit(1)
+
+        logger.verbose_detail(f"    Detected path: {detected_path}")
+
+        # Create source with detected path
+        source = MarketplaceSource(
+            name=display_name,
+            owner=owner,
+            repo=repo_name,
+            branch=branch,
+            path=detected_path,
+        )
+
+        # Fetch and validate
+        manifest = fetch_marketplace(source, force_refresh=True)
+        plugin_count = len(manifest.plugins)
+
+        # Register
+        add_marketplace(source)
+
+        logger.success(
+            f"Marketplace '{display_name}' registered ({plugin_count} plugins)",
+            symbol="check",
+        )
+        if manifest.description:
+            logger.verbose_detail(f"    {manifest.description}")
+
+    except Exception as e:
+        logger.error(f"Failed to register marketplace: {e}")
+        sys.exit(1)
+
+
+# ---------------------------------------------------------------------------
+# marketplace list
+# ---------------------------------------------------------------------------
+
+
+@marketplace.command(name="list", help="List registered marketplaces")
+@click.option("--verbose", "-v", is_flag=True, help="Show detailed output")
+def list_cmd(verbose):
+    """Show all registered marketplaces."""
+    logger = CommandLogger("marketplace-list", verbose=verbose)
+    try:
+        from ..marketplace.registry import get_registered_marketplaces
+
+        sources = get_registered_marketplaces()
+
+        if not sources:
+            logger.progress(
+                "No marketplaces registered. "
+                "Use 'apm marketplace add OWNER/REPO' to register one.",
+                symbol="info",
+            )
+            return
+
+        console = _get_console()
+        if not console:
+            # Colorama fallback
+            logger.progress(
+                f"{len(sources)} marketplace(s) registered:", symbol="info"
+            )
+            for s in sources:
+                click.echo(f"  {s.name}  ({s.owner}/{s.repo})")
+            return
+
+        from rich.table import Table
+
+        table = Table(
+            title="Registered Marketplaces",
+            show_header=True,
+            header_style="bold cyan",
+            border_style="cyan",
+        )
+        table.add_column("Name", style="bold white", no_wrap=True)
+        table.add_column("Repository", style="white")
+        table.add_column("Branch", style="cyan")
+        table.add_column("Path", style="dim")
+
+        for s in sources:
+            table.add_row(s.name, f"{s.owner}/{s.repo}", s.branch, s.path)
+
+        console.print()
+        console.print(table)
+        console.print(
+            f"\n[dim]Use 'apm marketplace browse <name>' to see plugins[/dim]"
+        )
+
+    except Exception as e:
+        logger.error(f"Failed to list marketplaces: {e}")
+        sys.exit(1)
+
+
+# ---------------------------------------------------------------------------
+# marketplace browse
+# ---------------------------------------------------------------------------
+
+
+@marketplace.command(help="Browse plugins in a marketplace")
+@click.argument("name", required=True)
+@click.option("--verbose", "-v", is_flag=True, help="Show detailed output")
+def browse(name, verbose):
+    """Show available plugins in a marketplace."""
+    logger = CommandLogger("marketplace-browse", verbose=verbose)
+    try:
+        from ..marketplace.client import fetch_marketplace
+        from ..marketplace.registry import get_marketplace_by_name
+
+        source = get_marketplace_by_name(name)
+        logger.start(f"Fetching plugins from '{name}'...", symbol="search")
+
+        manifest = fetch_marketplace(source, force_refresh=True)
+
+        if not manifest.plugins:
+            logger.warning(f"Marketplace '{name}' has no plugins")
+            return
+
+        console = _get_console()
+        if not console:
+            # Colorama fallback
+            logger.success(
+                f"{len(manifest.plugins)} plugin(s) in '{name}':", symbol="check"
+            )
+            for p in manifest.plugins:
+                desc = f" -- {p.description}" if p.description else ""
+                click.echo(f"  {p.name}{desc}")
+            click.echo(
+                f"\n  Install: apm install <plugin-name>@{name}"
+            )
+            return
+
+        from rich.table import Table
+
+        table = Table(
+            title=f"Plugins in '{name}'",
+            show_header=True,
+            header_style="bold cyan",
+            border_style="cyan",
+        )
+        table.add_column("Plugin", style="bold white", no_wrap=True)
+        table.add_column("Description", style="white", ratio=1)
+        table.add_column("Version", style="cyan", justify="center")
+        table.add_column("Install", style="green")
+
+        for p in manifest.plugins:
+            desc = p.description or "--"
+            ver = p.version or "--"
+            table.add_row(p.name, desc, ver, f"{p.name}@{name}")
+
+        console.print()
+        console.print(table)
+        console.print(
+            f"\n[dim]Install a plugin: apm install <plugin-name>@{name}[/dim]"
+        )
+
+    except Exception as e:
+        logger.error(f"Failed to browse marketplace: {e}")
+        sys.exit(1)
+
+
+# ---------------------------------------------------------------------------
+# marketplace update
+# ---------------------------------------------------------------------------
+
+
+@marketplace.command(help="Refresh marketplace cache")
+@click.argument("name", required=False, default=None)
+@click.option("--verbose", "-v", is_flag=True, help="Show detailed output")
+def update(name, verbose):
+    """Refresh cached marketplace data (one or all)."""
+    logger = CommandLogger("marketplace-update", verbose=verbose)
+    try:
+        from ..marketplace.client import clear_marketplace_cache, fetch_marketplace
+        from ..marketplace.registry import (
+            get_marketplace_by_name,
+            get_registered_marketplaces,
+        )
+
+        if name:
+            source = get_marketplace_by_name(name)
+            logger.start(f"Refreshing marketplace '{name}'...", symbol="gear")
+            clear_marketplace_cache(name)
+            manifest = fetch_marketplace(source, force_refresh=True)
+            logger.success(
+                f"Marketplace '{name}' updated ({len(manifest.plugins)} plugins)",
+                symbol="check",
+            )
+        else:
+            sources = get_registered_marketplaces()
+            if not sources:
+                logger.progress(
+                    "No marketplaces registered.", symbol="info"
+                )
+                return
+            logger.start(
+                f"Refreshing {len(sources)} marketplace(s)...", symbol="gear"
+            )
+            for s in sources:
+                try:
+                    clear_marketplace_cache(s.name)
+                    manifest = fetch_marketplace(s, force_refresh=True)
+                    logger.tree_item(
+                        f"  {s.name} ({len(manifest.plugins)} plugins)"
+                    )
+                except Exception as exc:
+                    logger.warning(f"  {s.name}: {exc}")
+            logger.success("Marketplace cache refreshed", symbol="check")
+
+    except Exception as e:
+        logger.error(f"Failed to update marketplace: {e}")
+        sys.exit(1)
+
+
+# ---------------------------------------------------------------------------
+# marketplace remove
+# ---------------------------------------------------------------------------
+
+
+@marketplace.command(help="Remove a registered marketplace")
+@click.argument("name", required=True)
+@click.option("--yes", "-y", is_flag=True, help="Skip confirmation")
+@click.option("--verbose", "-v", is_flag=True, help="Show detailed output")
+def remove(name, yes, verbose):
+    """Unregister a marketplace."""
+    logger = CommandLogger("marketplace-remove", verbose=verbose)
+    try:
+        from ..marketplace.client import clear_marketplace_cache
+        from ..marketplace.registry import get_marketplace_by_name, remove_marketplace
+
+        # Verify it exists first
+        source = get_marketplace_by_name(name)
+
+        if not yes:
+            confirmed = click.confirm(
+                f"Remove marketplace '{source.name}' ({source.owner}/{source.repo})?",
+                default=False,
+            )
+            if not confirmed:
+                logger.progress("Cancelled", symbol="info")
+                return
+
+        remove_marketplace(name)
+        clear_marketplace_cache(name)
+        logger.success(f"Marketplace '{name}' removed", symbol="check")
+
+    except Exception as e:
+        logger.error(f"Failed to remove marketplace: {e}")
+        sys.exit(1)
+
+
+# ---------------------------------------------------------------------------
+# Top-level search command (registered separately in cli.py)
+# ---------------------------------------------------------------------------
+
+
+@click.command(
+    name="search",
+    help="Search plugins in a marketplace (QUERY@MARKETPLACE)",
+)
+@click.argument("expression", required=True)
+@click.option("--limit", default=20, show_default=True, help="Max results to show")
+@click.option("--verbose", "-v", is_flag=True, help="Show detailed output")
+def search(expression, limit, verbose):
+    """Search for plugins in a specific marketplace.
+
+    Use QUERY@MARKETPLACE format, e.g.:  apm marketplace search security@skills
+    """
+    logger = CommandLogger("marketplace-search", verbose=verbose)
+    try:
+        from ..marketplace.client import search_marketplace
+        from ..marketplace.registry import get_marketplace_by_name
+
+        if "@" not in expression:
+            logger.error(
+                f"Invalid format: '{expression}'. "
+                "Use QUERY@MARKETPLACE, e.g.: apm marketplace search security@skills"
+            )
+            sys.exit(1)
+
+        query, marketplace_name = expression.rsplit("@", 1)
+        if not query or not marketplace_name:
+            logger.error(
+                "Both QUERY and MARKETPLACE are required. "
+                "Use QUERY@MARKETPLACE, e.g.: apm marketplace search security@skills"
+            )
+            sys.exit(1)
+
+        try:
+            source = get_marketplace_by_name(marketplace_name)
+        except Exception:
+            logger.error(
+                f"Marketplace '{marketplace_name}' is not registered. "
+                "Use 'apm marketplace list' to see registered marketplaces."
+            )
+            sys.exit(1)
+
+        logger.start(
+            f"Searching '{marketplace_name}' for '{query}'...", symbol="search"
+        )
+        results = search_marketplace(query, source)[:limit]
+
+        if not results:
+            logger.warning(
+                f"No plugins found matching '{query}' in '{marketplace_name}'. "
+                f"Try 'apm marketplace browse {marketplace_name}' to see all plugins."
+            )
+            return
+
+        console = _get_console()
+        if not console:
+            # Colorama fallback
+            logger.success(f"Found {len(results)} plugin(s):", symbol="check")
+            for p in results:
+                desc = f" -- {p.description}" if p.description else ""
+                click.echo(f"  {p.name}@{marketplace_name}{desc}")
+            click.echo(
+                f"\n  Install: apm install <plugin-name>@{marketplace_name}"
+            )
+            return
+
+        from rich.table import Table
+
+        table = Table(
+            title=f"Search Results: '{query}' in {marketplace_name}",
+            show_header=True,
+            header_style="bold cyan",
+            border_style="cyan",
+        )
+        table.add_column("Plugin", style="bold white", no_wrap=True)
+        table.add_column("Description", style="white", ratio=1)
+        table.add_column("Install", style="green")
+
+        for p in results:
+            desc = p.description or "--"
+            if len(desc) > 60:
+                desc = desc[:57] + "..."
+            table.add_row(p.name, desc, f"{p.name}@{marketplace_name}")
+
+        console.print()
+        console.print(table)
+        console.print(
+            f"\n[dim]Install: apm install <plugin-name>@{marketplace_name}[/dim]"
+        )
+
+    except SystemExit:
+        raise
+    except Exception as e:
+        logger.error(f"Search failed: {e}")
+        sys.exit(1)
diff --git a/src/apm_cli/core/command_logger.py b/src/apm_cli/core/command_logger.py
index 3ff1436b5..d3f7e61ff 100644
--- a/src/apm_cli/core/command_logger.py
+++ b/src/apm_cli/core/command_logger.py
@@ -22,6 +22,7 @@ class _ValidationOutcome:
 
     valid: list  # List of (canonical_name, already_present: bool) tuples
     invalid: list  # List of (package_name, reason: str) tuples
+    marketplace_provenance: dict = None  # canonical -> {discovered_via, marketplace_plugin_name}
 
     @property
     def all_failed(self) -> bool:
diff --git a/src/apm_cli/deps/lockfile.py b/src/apm_cli/deps/lockfile.py
index f9992fbbe..7702bf7d9 100644
--- a/src/apm_cli/deps/lockfile.py
+++ b/src/apm_cli/deps/lockfile.py
@@ -36,6 +36,8 @@ class LockedDependency:
     local_path: Optional[str] = None  # Original local path (relative to project root)
     content_hash: Optional[str] = None  # SHA-256 of package file tree
     is_dev: bool = False  # True for devDependencies
+    discovered_via: Optional[str] = None  # Marketplace name (provenance)
+    marketplace_plugin_name: Optional[str] = None  # Plugin name in marketplace
 
     def get_unique_key(self) -> str:
         """Returns unique key for this dependency."""
@@ -78,6 +80,10 @@ def to_dict(self) -> Dict[str, Any]:
             result["content_hash"] = self.content_hash
         if self.is_dev:
             result["is_dev"] = True
+        if self.discovered_via:
+            result["discovered_via"] = self.discovered_via
+        if self.marketplace_plugin_name:
+            result["marketplace_plugin_name"] = self.marketplace_plugin_name
         return result
 
     @classmethod
@@ -114,6 +120,8 @@ def from_dict(cls, data: Dict[str, Any]) -> "LockedDependency":
             local_path=data.get("local_path"),
             content_hash=data.get("content_hash"),
             is_dev=data.get("is_dev", False),
+            discovered_via=data.get("discovered_via"),
+            marketplace_plugin_name=data.get("marketplace_plugin_name"),
         )
 
     @classmethod
diff --git a/src/apm_cli/marketplace/__init__.py b/src/apm_cli/marketplace/__init__.py
new file mode 100644
index 000000000..e7b28c1ae
--- /dev/null
+++ b/src/apm_cli/marketplace/__init__.py
@@ -0,0 +1,28 @@
+"""Marketplace integration for plugin discovery and governance."""
+
+from .errors import (
+    MarketplaceError,
+    MarketplaceFetchError,
+    MarketplaceNotFoundError,
+    PluginNotFoundError,
+)
+from .models import (
+    MarketplaceManifest,
+    MarketplacePlugin,
+    MarketplaceSource,
+    parse_marketplace_json,
+)
+from .resolver import parse_marketplace_ref, resolve_marketplace_plugin
+
+__all__ = [
+    "MarketplaceError",
+    "MarketplaceFetchError",
+    "MarketplaceNotFoundError",
+    "PluginNotFoundError",
+    "MarketplaceManifest",
+    "MarketplacePlugin",
+    "MarketplaceSource",
+    "parse_marketplace_json",
+    "parse_marketplace_ref",
+    "resolve_marketplace_plugin",
+]
diff --git a/src/apm_cli/marketplace/client.py b/src/apm_cli/marketplace/client.py
new file mode 100644
index 000000000..cee2d6ebf
--- /dev/null
+++ b/src/apm_cli/marketplace/client.py
@@ -0,0 +1,302 @@
+"""Fetch, parse, and cache marketplace.json from GitHub repositories.
+
+Uses ``AuthResolver.try_with_fallback(unauth_first=True)`` for public-first
+access with automatic credential fallback for private marketplace repos.
+Cache lives at ``~/.apm/cache/marketplace/`` with a 1-hour TTL.
+"""
+
+import json
+import logging
+import os
+import time
+from typing import Dict, List, Optional
+
+import requests
+
+from .errors import MarketplaceFetchError
+from .models import MarketplaceManifest, MarketplacePlugin, MarketplaceSource, parse_marketplace_json
+from .registry import get_registered_marketplaces
+
+logger = logging.getLogger(__name__)
+
+_CACHE_TTL_SECONDS = 3600  # 1 hour
+_CACHE_DIR_NAME = os.path.join("cache", "marketplace")
+
+# Candidate locations for marketplace.json in a repository (priority order)
+_MARKETPLACE_PATHS = [
+    "marketplace.json",
+    ".github/plugin/marketplace.json",
+    ".claude-plugin/marketplace.json",
+]
+
+
+def _cache_dir() -> str:
+    """Return the cache directory, creating it if needed."""
+    from ..config import CONFIG_DIR
+
+    d = os.path.join(CONFIG_DIR, _CACHE_DIR_NAME)
+    os.makedirs(d, exist_ok=True)
+    return d
+
+
+def _sanitize_cache_name(name: str) -> str:
+    """Sanitize marketplace name for safe use in file paths."""
+    import re
+
+    from ..utils.path_security import PathTraversalError, validate_path_segments
+
+    safe = re.sub(r"[^a-zA-Z0-9._-]", "_", name)
+    # Prevent path traversal even after sanitization
+    safe = safe.strip(".").strip("_") or "unnamed"
+    # Defense-in-depth: validate with centralized path security
+    try:
+        validate_path_segments(safe, context="cache name")
+    except PathTraversalError:
+        safe = "unnamed"
+    return safe
+
+
+def _cache_data_path(name: str) -> str:
+    return os.path.join(_cache_dir(), f"{_sanitize_cache_name(name)}.json")
+
+
+def _cache_meta_path(name: str) -> str:
+    return os.path.join(_cache_dir(), f"{_sanitize_cache_name(name)}.meta.json")
+
+
+def _read_cache(name: str) -> Optional[Dict]:
+    """Read cached marketplace data if valid (not expired)."""
+    data_path = _cache_data_path(name)
+    meta_path = _cache_meta_path(name)
+    if not os.path.exists(data_path) or not os.path.exists(meta_path):
+        return None
+    try:
+        with open(meta_path, "r") as f:
+            meta = json.load(f)
+        fetched_at = meta.get("fetched_at", 0)
+        ttl = meta.get("ttl_seconds", _CACHE_TTL_SECONDS)
+        if time.time() - fetched_at > ttl:
+            return None  # Expired
+        with open(data_path, "r") as f:
+            return json.load(f)
+    except (json.JSONDecodeError, OSError, KeyError) as exc:
+        logger.debug("Cache read failed for '%s': %s", name, exc)
+        return None
+
+
+def _read_stale_cache(name: str) -> Optional[Dict]:
+    """Read cached data even if expired (stale-while-revalidate)."""
+    data_path = _cache_data_path(name)
+    if not os.path.exists(data_path):
+        return None
+    try:
+        with open(data_path, "r") as f:
+            return json.load(f)
+    except (json.JSONDecodeError, OSError):
+        return None
+
+
+def _write_cache(name: str, data: Dict) -> None:
+    """Write marketplace data and metadata to cache."""
+    data_path = _cache_data_path(name)
+    meta_path = _cache_meta_path(name)
+    try:
+        with open(data_path, "w") as f:
+            json.dump(data, f, indent=2)
+        with open(meta_path, "w") as f:
+            json.dump(
+                {"fetched_at": time.time(), "ttl_seconds": _CACHE_TTL_SECONDS},
+                f,
+            )
+    except OSError as exc:
+        logger.debug("Cache write failed for '%s': %s", name, exc)
+
+
+def _clear_cache(name: str) -> None:
+    """Remove cached data for a marketplace."""
+    for path in (_cache_data_path(name), _cache_meta_path(name)):
+        try:
+            os.remove(path)
+        except OSError:
+            pass
+
+
+# ---------------------------------------------------------------------------
+# Network fetch
+# ---------------------------------------------------------------------------
+
+
+def _github_contents_url(source: MarketplaceSource, file_path: str) -> str:
+    """Build the GitHub Contents API URL for a file."""
+    from ..core.auth import AuthResolver
+
+    host_info = AuthResolver.classify_host(source.host)
+    api_base = host_info.api_base
+    return f"{api_base}/repos/{source.owner}/{source.repo}/contents/{file_path}?ref={source.branch}"
+
+
+def _fetch_file(
+    source: MarketplaceSource,
+    file_path: str,
+    auth_resolver: Optional[object] = None,
+) -> Optional[Dict]:
+    """Fetch a JSON file from a GitHub repo via the Contents API.
+
+    Returns parsed JSON or ``None`` if the file does not exist (404).
+    Raises ``MarketplaceFetchError`` on unexpected failures.
+    """
+    url = _github_contents_url(source, file_path)
+
+    def _do_fetch(token, _git_env):
+        headers = {
+            "Accept": "application/vnd.github.v3.raw",
+            "User-Agent": "apm-cli",
+        }
+        if token:
+            headers["Authorization"] = f"token {token}"
+        resp = requests.get(url, headers=headers, timeout=30)
+        if resp.status_code == 404:
+            return None
+        resp.raise_for_status()
+        return resp.json()
+
+    if auth_resolver is None:
+        from ..core.auth import AuthResolver
+
+        auth_resolver = AuthResolver()
+
+    try:
+        return auth_resolver.try_with_fallback(
+            source.host,
+            _do_fetch,
+            org=source.owner,
+            unauth_first=True,
+        )
+    except Exception as exc:
+        raise MarketplaceFetchError(source.name, str(exc)) from exc
+
+
+def _auto_detect_path(
+    source: MarketplaceSource,
+    auth_resolver: Optional[object] = None,
+) -> Optional[str]:
+    """Probe candidate locations and return the first that exists.
+
+    Returns ``None`` if no location contains a marketplace.json.
+    Raises ``MarketplaceFetchError`` on non-404 failures (auth errors, etc.).
+    """
+    for candidate in _MARKETPLACE_PATHS:
+        data = _fetch_file(source, candidate, auth_resolver=auth_resolver)
+        if data is not None:
+            return candidate
+    return None
+
+
+# ---------------------------------------------------------------------------
+# Public API
+# ---------------------------------------------------------------------------
+
+
+def fetch_marketplace(
+    source: MarketplaceSource,
+    *,
+    force_refresh: bool = False,
+    auth_resolver: Optional[object] = None,
+) -> MarketplaceManifest:
+    """Fetch and parse a marketplace manifest.
+
+    Uses cache when available (1h TTL). Falls back to stale cache on
+    network errors.
+
+    Args:
+        source: Marketplace source to fetch.
+        force_refresh: Skip cache and re-fetch from network.
+        auth_resolver: Optional ``AuthResolver`` instance (created if None).
+
+    Returns:
+        MarketplaceManifest: Parsed manifest.
+
+    Raises:
+        MarketplaceFetchError: If fetch fails and no cache is available.
+    """
+    # Try fresh cache first
+    if not force_refresh:
+        cached = _read_cache(source.name)
+        if cached is not None:
+            logger.debug("Using cached marketplace data for '%s'", source.name)
+            return parse_marketplace_json(cached, source.name)
+
+    # Fetch from network
+    try:
+        data = _fetch_file(source, source.path, auth_resolver=auth_resolver)
+        if data is None:
+            raise MarketplaceFetchError(
+                source.name,
+                f"marketplace.json not found at '{source.path}' "
+                f"in {source.owner}/{source.repo}",
+            )
+        _write_cache(source.name, data)
+        return parse_marketplace_json(data, source.name)
+    except MarketplaceFetchError:
+        # Stale-while-revalidate: serve expired cache on network error
+        stale = _read_stale_cache(source.name)
+        if stale is not None:
+            logger.warning(
+                "Network error fetching '%s'; using stale cache", source.name
+            )
+            return parse_marketplace_json(stale, source.name)
+        raise
+
+
+def fetch_or_cache(
+    source: MarketplaceSource,
+    *,
+    auth_resolver: Optional[object] = None,
+) -> MarketplaceManifest:
+    """Convenience wrapper -- same as ``fetch_marketplace`` with defaults."""
+    return fetch_marketplace(source, auth_resolver=auth_resolver)
+
+
+def search_marketplace(
+    query: str,
+    source: MarketplaceSource,
+    *,
+    auth_resolver: Optional[object] = None,
+) -> List[MarketplacePlugin]:
+    """Search a single marketplace for plugins matching *query*."""
+    manifest = fetch_marketplace(source, auth_resolver=auth_resolver)
+    return manifest.search(query)
+
+
+def search_all_marketplaces(
+    query: str,
+    *,
+    auth_resolver: Optional[object] = None,
+) -> List[MarketplacePlugin]:
+    """Search across all registered marketplaces.
+
+    Returns plugins matching the query, annotated with their source marketplace.
+    """
+    results: List[MarketplacePlugin] = []
+    for source in get_registered_marketplaces():
+        try:
+            manifest = fetch_marketplace(source, auth_resolver=auth_resolver)
+            results.extend(manifest.search(query))
+        except MarketplaceFetchError as exc:
+            logger.warning("Skipping marketplace '%s': %s", source.name, exc)
+    return results
+
+
+def clear_marketplace_cache(name: Optional[str] = None) -> int:
+    """Clear cached data for one or all marketplaces.
+
+    Returns the number of caches cleared.
+    """
+    if name:
+        _clear_cache(name)
+        return 1
+    count = 0
+    for source in get_registered_marketplaces():
+        _clear_cache(source.name)
+        count += 1
+    return count
diff --git a/src/apm_cli/marketplace/errors.py b/src/apm_cli/marketplace/errors.py
new file mode 100644
index 000000000..c286db461
--- /dev/null
+++ b/src/apm_cli/marketplace/errors.py
@@ -0,0 +1,44 @@
+"""Marketplace-specific error hierarchy."""
+
+
+class MarketplaceError(Exception):
+    """Base class for marketplace errors."""
+
+    pass
+
+
+class MarketplaceNotFoundError(MarketplaceError):
+    """Raised when a registered marketplace cannot be found."""
+
+    def __init__(self, name: str):
+        self.name = name
+        super().__init__(
+            f"Marketplace '{name}' is not registered. "
+            f"Run 'apm marketplace add OWNER/REPO' to register it, "
+            f"or 'apm marketplace list' to see registered marketplaces."
+        )
+
+
+class PluginNotFoundError(MarketplaceError):
+    """Raised when a plugin is not found in a marketplace."""
+
+    def __init__(self, plugin_name: str, marketplace_name: str):
+        self.plugin_name = plugin_name
+        self.marketplace_name = marketplace_name
+        super().__init__(
+            f"Plugin '{plugin_name}' not found in marketplace '{marketplace_name}'. "
+            f"Run 'apm marketplace browse {marketplace_name}' to see available plugins."
+        )
+
+
+class MarketplaceFetchError(MarketplaceError):
+    """Raised when fetching marketplace data fails."""
+
+    def __init__(self, name: str, reason: str = ""):
+        self.name = name
+        self.reason = reason
+        detail = f": {reason}" if reason else ""
+        super().__init__(
+            f"Failed to fetch marketplace '{name}'{detail}. "
+            f"Run 'apm marketplace update {name}' to retry."
+        )
diff --git a/src/apm_cli/marketplace/models.py b/src/apm_cli/marketplace/models.py
new file mode 100644
index 000000000..478eaae1f
--- /dev/null
+++ b/src/apm_cli/marketplace/models.py
@@ -0,0 +1,219 @@
+"""Frozen dataclasses and JSON parser for marketplace manifests.
+
+Supports both Copilot CLI and Claude Code marketplace.json formats.
+All dataclasses are frozen for thread-safety.
+"""
+
+import logging
+from dataclasses import dataclass, field
+from typing import Any, Dict, List, Optional, Tuple
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass(frozen=True)
+class MarketplaceSource:
+    """A registered marketplace repository.
+
+    Stored in ``~/.apm/marketplaces.json``.
+    """
+
+    name: str  # Display name (e.g., "acme-tools")
+    owner: str  # GitHub owner
+    repo: str  # GitHub repo
+    host: str = "github.com"
+    branch: str = "main"
+    path: str = "marketplace.json"  # Detected on add
+
+    def to_dict(self) -> Dict[str, Any]:
+        """Serialize to dict for JSON storage."""
+        result: Dict[str, Any] = {
+            "name": self.name,
+            "owner": self.owner,
+            "repo": self.repo,
+        }
+        if self.host != "github.com":
+            result["host"] = self.host
+        if self.branch != "main":
+            result["branch"] = self.branch
+        if self.path != "marketplace.json":
+            result["path"] = self.path
+        return result
+
+    @classmethod
+    def from_dict(cls, data: Dict[str, Any]) -> "MarketplaceSource":
+        """Deserialize from JSON dict."""
+        return cls(
+            name=data["name"],
+            owner=data["owner"],
+            repo=data["repo"],
+            host=data.get("host", "github.com"),
+            branch=data.get("branch", "main"),
+            path=data.get("path", "marketplace.json"),
+        )
+
+
+@dataclass(frozen=True)
+class MarketplacePlugin:
+    """A single plugin entry inside a marketplace manifest."""
+
+    name: str  # Plugin name (unique within marketplace)
+    source: Any = None  # String (relative) or dict (github/url/git-subdir)
+    description: str = ""
+    version: str = ""
+    tags: Tuple[str, ...] = ()
+    source_marketplace: str = ""  # Populated during resolution
+
+    def matches_query(self, query: str) -> bool:
+        """Return True if the plugin matches a search query (case-insensitive)."""
+        q = query.lower()
+        return (
+            q in self.name.lower()
+            or q in self.description.lower()
+            or any(q in tag.lower() for tag in self.tags)
+        )
+
+
+@dataclass(frozen=True)
+class MarketplaceManifest:
+    """Parsed marketplace.json content."""
+
+    name: str
+    plugins: Tuple[MarketplacePlugin, ...] = ()
+    owner_name: str = ""
+    description: str = ""
+
+    def find_plugin(self, plugin_name: str) -> Optional[MarketplacePlugin]:
+        """Find a plugin by exact name (case-insensitive)."""
+        lower = plugin_name.lower()
+        for p in self.plugins:
+            if p.name.lower() == lower:
+                return p
+        return None
+
+    def search(self, query: str) -> List[MarketplacePlugin]:
+        """Search plugins matching a query."""
+        return [p for p in self.plugins if p.matches_query(query)]
+
+
+# ---------------------------------------------------------------------------
+# JSON parser -- handles Copilot CLI and Claude Code marketplace.json formats
+# ---------------------------------------------------------------------------
+
+# Copilot CLI format:
+#   { "name": "...", "plugins": [ { "name": "...", "repository": "owner/repo" } ] }
+#
+# Claude Code format:
+#   { "name": "...", "plugins": [ { "name": "...", "source": { "type": "github", ... } } ] }
+
+def _parse_plugin_entry(
+    entry: Dict[str, Any], source_name: str
+) -> Optional[MarketplacePlugin]:
+    """Parse a single plugin entry from either format."""
+    name = entry.get("name", "").strip()
+    if not name:
+        logger.debug("Skipping marketplace plugin entry without a name")
+        return None
+
+    description = entry.get("description", "")
+    version = entry.get("version", "")
+    raw_tags = entry.get("tags", [])
+    tags = tuple(raw_tags) if isinstance(raw_tags, list) else ()
+
+    # Determine source -- Copilot uses "repository", Claude uses "source"
+    source: Any = None
+
+    if "source" in entry:
+        raw = entry["source"]
+        if isinstance(raw, str):
+            # Relative path source (Claude shorthand)
+            source = raw
+        elif isinstance(raw, dict):
+            # Type discriminator: Copilot CLI uses "source" key, Claude uses "type"
+            source_type = raw.get("type", "") or raw.get("source", "")
+            if source_type == "npm":
+                logger.debug(
+                    "Skipping npm source type for plugin '%s' (unsupported)", name
+                )
+                return None
+            # Normalize: ensure "type" key is set for downstream resolvers
+            if source_type and "type" not in raw:
+                raw = {**raw, "type": source_type}
+            source = raw
+        else:
+            logger.debug(
+                "Skipping plugin '%s' with unrecognized source format", name
+            )
+            return None
+    elif "repository" in entry:
+        # Copilot CLI format: "repository": "owner/repo"
+        repo = entry["repository"]
+        ref = entry.get("ref", "")
+        if isinstance(repo, str) and "/" in repo:
+            source = {"type": "github", "repo": repo}
+            if ref:
+                source["ref"] = ref
+        else:
+            logger.debug(
+                "Skipping plugin '%s' with invalid repository field: %s",
+                name,
+                repo,
+            )
+            return None
+    else:
+        logger.debug("Plugin '%s' has no source or repository field", name)
+        return None
+
+    return MarketplacePlugin(
+        name=name,
+        source=source,
+        description=description,
+        version=version,
+        tags=tags,
+        source_marketplace=source_name,
+    )
+
+
+def parse_marketplace_json(
+    data: Dict[str, Any], source_name: str = ""
+) -> MarketplaceManifest:
+    """Parse a marketplace.json dict into a ``MarketplaceManifest``.
+
+    Accepts both Copilot CLI and Claude Code marketplace formats.
+    Invalid or unsupported entries are silently skipped with debug logging.
+
+    Args:
+        data: Parsed JSON content of marketplace.json.
+        source_name: Display name of the marketplace (for provenance).
+
+    Returns:
+        MarketplaceManifest: Parsed manifest with valid plugin entries.
+    """
+    manifest_name = data.get("name", source_name or "unknown")
+    description = data.get("description", "")
+    owner_name = data.get("owner", {}).get("name", "") if isinstance(
+        data.get("owner"), dict
+    ) else data.get("owner", "")
+
+    raw_plugins = data.get("plugins", [])
+    if not isinstance(raw_plugins, list):
+        logger.warning(
+            "marketplace.json 'plugins' field is not a list in '%s'",
+            source_name,
+        )
+        raw_plugins = []
+
+    plugins: List[MarketplacePlugin] = []
+    for entry in raw_plugins:
+        if not isinstance(entry, dict):
+            continue
+        plugin = _parse_plugin_entry(entry, source_name)
+        if plugin is not None:
+            plugins.append(plugin)
+
+    return MarketplaceManifest(
+        name=manifest_name,
+        plugins=tuple(plugins),
+        owner_name=owner_name,
+        description=description,
+    )
diff --git a/src/apm_cli/marketplace/registry.py b/src/apm_cli/marketplace/registry.py
new file mode 100644
index 000000000..3aa668d58
--- /dev/null
+++ b/src/apm_cli/marketplace/registry.py
@@ -0,0 +1,130 @@
+"""Manage registered marketplaces in ``~/.apm/marketplaces.json``."""
+
+import json
+import logging
+import os
+from typing import Dict, List, Optional
+
+from .errors import MarketplaceNotFoundError
+from .models import MarketplaceSource
+
+logger = logging.getLogger(__name__)
+
+_MARKETPLACES_FILENAME = "marketplaces.json"
+
+# Process-lifetime cache --------------------------------------------------
+_registry_cache: Optional[List[MarketplaceSource]] = None
+
+
+def _marketplaces_path() -> str:
+    """Return the full path to ``~/.apm/marketplaces.json``."""
+    from ..config import CONFIG_DIR
+
+    return os.path.join(CONFIG_DIR, _MARKETPLACES_FILENAME)
+
+
+def _ensure_file() -> str:
+    """Ensure the marketplaces file exists, creating it if needed."""
+    from ..config import ensure_config_exists
+
+    ensure_config_exists()
+    path = _marketplaces_path()
+    if not os.path.exists(path):
+        with open(path, "w") as f:
+            json.dump({"marketplaces": []}, f, indent=2)
+    return path
+
+
+def _invalidate_cache() -> None:
+    global _registry_cache
+    _registry_cache = None
+
+
+def _load() -> List[MarketplaceSource]:
+    """Load registered marketplaces from disk (cached per-process)."""
+    global _registry_cache
+    if _registry_cache is not None:
+        return list(_registry_cache)
+
+    path = _ensure_file()
+    try:
+        with open(path, "r") as f:
+            data = json.load(f)
+    except (json.JSONDecodeError, OSError) as exc:
+        logger.warning("Failed to read %s: %s", path, exc)
+        data = {"marketplaces": []}
+
+    sources: List[MarketplaceSource] = []
+    for entry in data.get("marketplaces", []):
+        try:
+            sources.append(MarketplaceSource.from_dict(entry))
+        except (KeyError, TypeError) as exc:
+            logger.debug("Skipping invalid marketplace entry: %s", exc)
+
+    _registry_cache = sources
+    return list(sources)
+
+
+def _save(sources: List[MarketplaceSource]) -> None:
+    """Write marketplace list to disk atomically."""
+    global _registry_cache
+    path = _ensure_file()
+    data = {"marketplaces": [s.to_dict() for s in sources]}
+    tmp = path + ".tmp"
+    with open(tmp, "w") as f:
+        json.dump(data, f, indent=2)
+    os.replace(tmp, path)
+    _registry_cache = list(sources)
+
+
+# Public API ---------------------------------------------------------------
+
+
+def get_registered_marketplaces() -> List[MarketplaceSource]:
+    """Return all registered marketplaces."""
+    return _load()
+
+
+def get_marketplace_by_name(name: str) -> MarketplaceSource:
+    """Return a marketplace by display name (case-insensitive).
+
+    Raises:
+        MarketplaceNotFoundError: If not found.
+    """
+    lower = name.lower()
+    for src in _load():
+        if src.name.lower() == lower:
+            return src
+    raise MarketplaceNotFoundError(name)
+
+
+def add_marketplace(source: MarketplaceSource) -> None:
+    """Register a marketplace (replaces if same name exists)."""
+    sources = [s for s in _load() if s.name.lower() != source.name.lower()]
+    sources.append(source)
+    _save(sources)
+    logger.debug("Registered marketplace '%s'", source.name)
+
+
+def remove_marketplace(name: str) -> None:
+    """Remove a marketplace by name.
+
+    Raises:
+        MarketplaceNotFoundError: If not found.
+    """
+    before = _load()
+    after = [s for s in before if s.name.lower() != name.lower()]
+    if len(after) == len(before):
+        raise MarketplaceNotFoundError(name)
+    _save(after)
+    logger.debug("Removed marketplace '%s'", name)
+
+
+def marketplace_names() -> List[str]:
+    """Return sorted list of registered marketplace names."""
+    return sorted(s.name for s in _load())
+
+
+def marketplace_count() -> int:
+    """Return the number of registered marketplaces."""
+    return len(_load())
diff --git a/src/apm_cli/marketplace/resolver.py b/src/apm_cli/marketplace/resolver.py
new file mode 100644
index 000000000..e8f5768c9
--- /dev/null
+++ b/src/apm_cli/marketplace/resolver.py
@@ -0,0 +1,229 @@
+"""Resolve ``NAME@MARKETPLACE`` specifiers to canonical ``owner/repo#ref`` strings.
+
+The ``@`` disambiguation rule:
+- If input matches ``^[a-zA-Z0-9._-]+@[a-zA-Z0-9._-]+$`` (no ``/``, no ``:``),
+  it is a marketplace ref.
+- Everything else goes to the existing ``DependencyReference.parse()`` path.
+- These inputs previously raised ``ValueError`` ("Use 'user/repo' format"),
+  so this is a backward-compatible grammar extension.
+"""
+
+import logging
+import re
+from typing import Optional, Tuple
+
+from ..utils.path_security import PathTraversalError, validate_path_segments
+from .client import fetch_or_cache
+from .errors import MarketplaceFetchError, PluginNotFoundError
+from .models import MarketplacePlugin
+from .registry import get_marketplace_by_name
+
+logger = logging.getLogger(__name__)
+
+_MARKETPLACE_RE = re.compile(r"^([a-zA-Z0-9._-]+)@([a-zA-Z0-9._-]+)$")
+
+
+def parse_marketplace_ref(specifier: str) -> Optional[Tuple[str, str]]:
+    """Parse a ``NAME@MARKETPLACE`` specifier.
+
+    Returns:
+        ``(plugin_name, marketplace_name)`` if the specifier matches,
+        or ``None`` if it does not look like a marketplace ref.
+    """
+    s = specifier.strip()
+    # Quick rejection: slashes and colons belong to other formats
+    if "/" in s or ":" in s:
+        return None
+    match = _MARKETPLACE_RE.match(s)
+    if match:
+        return (match.group(1), match.group(2))
+    return None
+
+
+def _resolve_github_source(source: dict) -> str:
+    """Resolve a ``github`` source type to ``owner/repo[/path][#ref]``.
+
+    Accepts ``path`` field (Copilot CLI format) as a virtual subdirectory.
+    """
+    repo = source.get("repo", "")
+    ref = source.get("ref", "")
+    path = source.get("path", "").strip("/")
+    if not repo or "/" not in repo:
+        raise ValueError(
+            f"Invalid github source: 'repo' field must be 'owner/repo', got '{repo}'"
+        )
+    if path:
+        try:
+            validate_path_segments(path, context="github source path")
+        except PathTraversalError as exc:
+            raise ValueError(str(exc)) from exc
+        base = f"{repo}/{path}"
+    else:
+        base = repo
+    if ref:
+        return f"{base}#{ref}"
+    return base
+
+
+def _resolve_url_source(source: dict) -> str:
+    """Resolve a ``url`` source type.
+
+    APM is Git-native -- URL sources that point to GitHub repos are
+    resolved to ``owner/repo``. Non-GitHub URLs are rejected.
+    """
+    url = source.get("url", "")
+    # Try to extract owner/repo from common GitHub URL patterns
+    for prefix in ("https://github.com/", "http://github.com/"):
+        if url.lower().startswith(prefix):
+            path = url[len(prefix) :].rstrip("/").split("?")[0]
+            # Remove .git suffix
+            if path.endswith(".git"):
+                path = path[:-4]
+            parts = path.split("/")
+            if len(parts) >= 2:
+                return f"{parts[0]}/{parts[1]}"
+
+    raise ValueError(
+        f"Cannot resolve URL source '{url}' to a Git coordinate. "
+        f"APM requires Git-based sources (owner/repo format)."
+    )
+
+
+def _resolve_git_subdir_source(source: dict) -> str:
+    """Resolve a ``git-subdir`` source type to ``owner/repo[/subdir][#ref]``."""
+    repo = source.get("repo", "")
+    ref = source.get("ref", "")
+    subdir = (source.get("subdir", "") or source.get("path", "")).strip("/")
+    if not repo or "/" not in repo:
+        raise ValueError(
+            f"Invalid git-subdir source: 'repo' must be 'owner/repo', got '{repo}'"
+        )
+    if subdir:
+        try:
+            validate_path_segments(subdir, context="git-subdir source path")
+        except PathTraversalError as exc:
+            raise ValueError(str(exc)) from exc
+        base = f"{repo}/{subdir}"
+    else:
+        base = repo
+    if ref:
+        return f"{base}#{ref}"
+    return base
+
+
+def _resolve_relative_source(source: str, marketplace_owner: str, marketplace_repo: str) -> str:
+    """Resolve a relative path source to ``owner/repo[/subdir]``.
+
+    Relative sources point to subdirectories within the marketplace repo itself.
+    """
+    # Normalize the relative path (strip leading ./ and trailing /)
+    rel = source.strip("/")
+    if rel.startswith("./"):
+        rel = rel[2:]
+    rel = rel.strip("/")
+    if rel and rel != ".":
+        try:
+            validate_path_segments(rel, context="relative source path")
+        except PathTraversalError as exc:
+            raise ValueError(str(exc)) from exc
+        return f"{marketplace_owner}/{marketplace_repo}/{rel}"
+    return f"{marketplace_owner}/{marketplace_repo}"
+
+
+def resolve_plugin_source(
+    plugin: MarketplacePlugin,
+    marketplace_owner: str = "",
+    marketplace_repo: str = "",
+) -> str:
+    """Resolve a plugin's source to a canonical ``owner/repo[#ref]`` string.
+
+    Handles 4 source types: relative, github, url, git-subdir.
+    NPM sources are rejected with a clear message.
+
+    Args:
+        plugin: The marketplace plugin to resolve.
+        marketplace_owner: Owner of the marketplace repo (for relative sources).
+        marketplace_repo: Repo name of the marketplace (for relative sources).
+
+    Returns:
+        Canonical ``owner/repo[#ref]`` string.
+
+    Raises:
+        ValueError: If the source type is unsupported or the source is invalid.
+    """
+    source = plugin.source
+    if source is None:
+        raise ValueError(f"Plugin '{plugin.name}' has no source defined")
+
+    # String source = relative path
+    if isinstance(source, str):
+        return _resolve_relative_source(source, marketplace_owner, marketplace_repo)
+
+    if not isinstance(source, dict):
+        raise ValueError(
+            f"Plugin '{plugin.name}' has unrecognized source format: {type(source).__name__}"
+        )
+
+    source_type = source.get("type", "")
+
+    if source_type == "github":
+        return _resolve_github_source(source)
+    elif source_type == "url":
+        return _resolve_url_source(source)
+    elif source_type == "git-subdir":
+        return _resolve_git_subdir_source(source)
+    elif source_type == "npm":
+        raise ValueError(
+            f"Plugin '{plugin.name}' uses npm source type which is not supported by APM. "
+            f"APM requires Git-based sources. "
+            f"Consider asking the marketplace maintainer to add a 'github' source."
+        )
+    else:
+        raise ValueError(
+            f"Plugin '{plugin.name}' has unsupported source type: '{source_type}'"
+        )
+
+
+def resolve_marketplace_plugin(
+    plugin_name: str,
+    marketplace_name: str,
+    *,
+    auth_resolver: Optional[object] = None,
+) -> Tuple[str, MarketplacePlugin]:
+    """Resolve a marketplace plugin reference to a canonical string.
+
+    Args:
+        plugin_name: Plugin name within the marketplace.
+        marketplace_name: Registered marketplace name.
+        auth_resolver: Optional ``AuthResolver`` instance.
+
+    Returns:
+        Tuple of (canonical ``owner/repo[#ref]`` string, resolved plugin).
+
+    Raises:
+        MarketplaceNotFoundError: If the marketplace is not registered.
+        PluginNotFoundError: If the plugin is not in the marketplace.
+        MarketplaceFetchError: If the marketplace cannot be fetched.
+        ValueError: If the plugin source cannot be resolved.
+    """
+    source = get_marketplace_by_name(marketplace_name)
+    manifest = fetch_or_cache(source, auth_resolver=auth_resolver)
+
+    plugin = manifest.find_plugin(plugin_name)
+    if plugin is None:
+        raise PluginNotFoundError(plugin_name, marketplace_name)
+
+    canonical = resolve_plugin_source(
+        plugin,
+        marketplace_owner=source.owner,
+        marketplace_repo=source.repo,
+    )
+
+    logger.debug(
+        "Resolved %s@%s -> %s",
+        plugin_name,
+        marketplace_name,
+        canonical,
+    )
+
+    return canonical, plugin
diff --git a/tests/unit/marketplace/__init__.py b/tests/unit/marketplace/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/tests/unit/marketplace/test_lockfile_provenance.py b/tests/unit/marketplace/test_lockfile_provenance.py
new file mode 100644
index 000000000..0600b3768
--- /dev/null
+++ b/tests/unit/marketplace/test_lockfile_provenance.py
@@ -0,0 +1,73 @@
+"""Tests for lockfile provenance fields -- serialization round-trip and backward compat."""
+
+import pytest
+
+from apm_cli.deps.lockfile import LockedDependency
+
+
+class TestLockedDependencyProvenance:
+    """Verify marketplace provenance fields round-trip correctly."""
+
+    def test_default_none(self):
+        dep = LockedDependency(repo_url="owner/repo")
+        assert dep.discovered_via is None
+        assert dep.marketplace_plugin_name is None
+
+    def test_to_dict_omits_none(self):
+        dep = LockedDependency(repo_url="owner/repo")
+        d = dep.to_dict()
+        assert "discovered_via" not in d
+        assert "marketplace_plugin_name" not in d
+
+    def test_to_dict_includes_values(self):
+        dep = LockedDependency(
+            repo_url="owner/repo",
+            discovered_via="acme-tools",
+            marketplace_plugin_name="security-checks",
+        )
+        d = dep.to_dict()
+        assert d["discovered_via"] == "acme-tools"
+        assert d["marketplace_plugin_name"] == "security-checks"
+
+    def test_from_dict_missing_fields(self):
+        """Old lockfiles without provenance fields still deserialize."""
+        dep = LockedDependency.from_dict({"repo_url": "owner/repo"})
+        assert dep.discovered_via is None
+        assert dep.marketplace_plugin_name is None
+
+    def test_from_dict_with_fields(self):
+        dep = LockedDependency.from_dict({
+            "repo_url": "owner/repo",
+            "discovered_via": "acme-tools",
+            "marketplace_plugin_name": "security-checks",
+        })
+        assert dep.discovered_via == "acme-tools"
+        assert dep.marketplace_plugin_name == "security-checks"
+
+    def test_roundtrip(self):
+        original = LockedDependency(
+            repo_url="owner/repo",
+            resolved_commit="abc123",
+            resolved_ref="v1.0",
+            discovered_via="acme-tools",
+            marketplace_plugin_name="security-checks",
+        )
+        restored = LockedDependency.from_dict(original.to_dict())
+        assert restored.discovered_via == "acme-tools"
+        assert restored.marketplace_plugin_name == "security-checks"
+        assert restored.resolved_commit == "abc123"
+        assert restored.resolved_ref == "v1.0"
+
+    def test_backward_compat_existing_fields(self):
+        """Ensure existing fields still work alongside new provenance fields."""
+        dep = LockedDependency.from_dict({
+            "repo_url": "owner/repo",
+            "resolved_commit": "abc123",
+            "content_hash": "sha256:def456",
+            "is_dev": True,
+            "discovered_via": "mkt",
+        })
+        assert dep.resolved_commit == "abc123"
+        assert dep.content_hash == "sha256:def456"
+        assert dep.is_dev is True
+        assert dep.discovered_via == "mkt"
diff --git a/tests/unit/marketplace/test_marketplace_client.py b/tests/unit/marketplace/test_marketplace_client.py
new file mode 100644
index 000000000..398d76cb5
--- /dev/null
+++ b/tests/unit/marketplace/test_marketplace_client.py
@@ -0,0 +1,198 @@
+"""Tests for marketplace client -- HTTP mock, caching, TTL, auth, auto-detection."""
+
+import json
+import time
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from apm_cli.marketplace.errors import MarketplaceFetchError
+from apm_cli.marketplace.models import MarketplaceSource
+from apm_cli.marketplace import client as client_mod
+
+
+@pytest.fixture(autouse=True)
+def _isolate_cache(tmp_path, monkeypatch):
+    """Point cache and config to temp directories."""
+    config_dir = str(tmp_path / ".apm")
+    monkeypatch.setattr("apm_cli.config.CONFIG_DIR", config_dir)
+    monkeypatch.setattr("apm_cli.config.CONFIG_FILE", str(tmp_path / ".apm" / "config.json"))
+    monkeypatch.setattr("apm_cli.config._config_cache", None)
+    monkeypatch.setattr("apm_cli.marketplace.registry._registry_cache", None)
+    yield
+
+
+def _make_source(name="acme"):
+    return MarketplaceSource(name=name, owner="acme-org", repo="plugins")
+
+
+class TestCache:
+    """Cache read/write with TTL."""
+
+    def test_write_and_read(self, tmp_path):
+        data = {"name": "Test", "plugins": []}
+        client_mod._write_cache("test-mkt", data)
+
+        cached = client_mod._read_cache("test-mkt")
+        assert cached is not None
+        assert cached["name"] == "Test"
+
+    def test_expired_cache(self, tmp_path, monkeypatch):
+        data = {"name": "Test", "plugins": []}
+        client_mod._write_cache("test-mkt", data)
+
+        # Make the cache appear old
+        meta_path = client_mod._cache_meta_path("test-mkt")
+        with open(meta_path, "w") as f:
+            json.dump({"fetched_at": time.time() - 7200, "ttl_seconds": 3600}, f)
+
+        assert client_mod._read_cache("test-mkt") is None
+
+    def test_stale_cache_still_readable(self, tmp_path):
+        data = {"name": "Stale", "plugins": []}
+        client_mod._write_cache("test-mkt", data)
+
+        # Make the cache appear old
+        meta_path = client_mod._cache_meta_path("test-mkt")
+        with open(meta_path, "w") as f:
+            json.dump({"fetched_at": time.time() - 7200, "ttl_seconds": 3600}, f)
+
+        stale = client_mod._read_stale_cache("test-mkt")
+        assert stale is not None
+        assert stale["name"] == "Stale"
+
+    def test_clear_cache(self, tmp_path):
+        data = {"name": "Test", "plugins": []}
+        client_mod._write_cache("test-mkt", data)
+        client_mod._clear_cache("test-mkt")
+        assert client_mod._read_cache("test-mkt") is None
+
+    def test_nonexistent_cache(self):
+        assert client_mod._read_cache("nonexistent") is None
+        assert client_mod._read_stale_cache("nonexistent") is None
+
+
+class TestFetchMarketplace:
+    """fetch_marketplace with mocked HTTP."""
+
+    def test_fetch_from_network(self, tmp_path):
+        source = _make_source()
+        raw_data = {
+            "name": "Acme Plugins",
+            "plugins": [
+                {"name": "tool-a", "repository": "acme-org/tool-a"},
+            ],
+        }
+        mock_resolver = MagicMock()
+        mock_resolver.try_with_fallback.return_value = raw_data
+        mock_resolver.classify_host.return_value = MagicMock(api_base="https://api.github.com")
+
+        manifest = client_mod.fetch_marketplace(
+            source, force_refresh=True, auth_resolver=mock_resolver
+        )
+        assert manifest.name == "Acme Plugins"
+        assert len(manifest.plugins) == 1
+
+    def test_serves_from_cache(self, tmp_path):
+        source = _make_source()
+        raw_data = {
+            "name": "Cached",
+            "plugins": [{"name": "cached-tool", "repository": "o/r"}],
+        }
+        client_mod._write_cache(source.name, raw_data)
+
+        # Should not hit network
+        manifest = client_mod.fetch_marketplace(source)
+        assert manifest.name == "Cached"
+        assert len(manifest.plugins) == 1
+
+    def test_force_refresh_bypasses_cache(self, tmp_path):
+        source = _make_source()
+        client_mod._write_cache(source.name, {"name": "Old", "plugins": []})
+
+        new_data = {"name": "Fresh", "plugins": [{"name": "new", "repository": "o/r"}]}
+        mock_resolver = MagicMock()
+        mock_resolver.try_with_fallback.return_value = new_data
+        mock_resolver.classify_host.return_value = MagicMock(api_base="https://api.github.com")
+
+        manifest = client_mod.fetch_marketplace(
+            source, force_refresh=True, auth_resolver=mock_resolver
+        )
+        assert manifest.name == "Fresh"
+
+    def test_stale_while_revalidate(self, tmp_path):
+        source = _make_source()
+        stale_data = {"name": "Stale", "plugins": []}
+        client_mod._write_cache(source.name, stale_data)
+
+        # Expire the cache
+        meta_path = client_mod._cache_meta_path(source.name)
+        with open(meta_path, "w") as f:
+            json.dump({"fetched_at": time.time() - 7200, "ttl_seconds": 3600}, f)
+
+        # Network fetch will fail
+        mock_resolver = MagicMock()
+        mock_resolver.try_with_fallback.side_effect = Exception("Network error")
+        mock_resolver.classify_host.return_value = MagicMock(api_base="https://api.github.com")
+
+        manifest = client_mod.fetch_marketplace(
+            source, auth_resolver=mock_resolver
+        )
+        assert manifest.name == "Stale"  # Falls back to stale cache
+
+    def test_no_cache_no_network_raises(self, tmp_path):
+        source = _make_source()
+        mock_resolver = MagicMock()
+        mock_resolver.try_with_fallback.side_effect = Exception("Network error")
+        mock_resolver.classify_host.return_value = MagicMock(api_base="https://api.github.com")
+
+        with pytest.raises(MarketplaceFetchError):
+            client_mod.fetch_marketplace(
+                source, force_refresh=True, auth_resolver=mock_resolver
+            )
+
+
+class TestAutoDetectPath:
+    """Auto-detect marketplace.json location in a repo."""
+
+    def test_found_at_root(self, tmp_path):
+        source = _make_source()
+        mock_resolver = MagicMock()
+
+        def mock_fetch(host, op, org=None, unauth_first=False):
+            # First probe: marketplace.json at root -- found
+            return {"name": "Test", "plugins": []}
+
+        mock_resolver.try_with_fallback.side_effect = mock_fetch
+        mock_resolver.classify_host.return_value = MagicMock(api_base="https://api.github.com")
+
+        path = client_mod._auto_detect_path(source, auth_resolver=mock_resolver)
+        assert path == "marketplace.json"
+
+    def test_found_at_github_plugin(self, tmp_path):
+        source = _make_source()
+        mock_resolver = MagicMock()
+        call_count = [0]
+
+        def mock_fetch(host, op, org=None, unauth_first=False):
+            call_count[0] += 1
+            if call_count[0] == 1:
+                # First probe: root -- not found (404)
+                return None
+            # Second probe: .github/plugin/ -- found
+            return {"name": "Test", "plugins": []}
+
+        mock_resolver.try_with_fallback.side_effect = mock_fetch
+        mock_resolver.classify_host.return_value = MagicMock(api_base="https://api.github.com")
+
+        path = client_mod._auto_detect_path(source, auth_resolver=mock_resolver)
+        assert path == ".github/plugin/marketplace.json"
+
+    def test_not_found_anywhere(self, tmp_path):
+        source = _make_source()
+        mock_resolver = MagicMock()
+        mock_resolver.try_with_fallback.return_value = None
+        mock_resolver.classify_host.return_value = MagicMock(api_base="https://api.github.com")
+
+        path = client_mod._auto_detect_path(source, auth_resolver=mock_resolver)
+        assert path is None
diff --git a/tests/unit/marketplace/test_marketplace_commands.py b/tests/unit/marketplace/test_marketplace_commands.py
new file mode 100644
index 000000000..7e77377d5
--- /dev/null
+++ b/tests/unit/marketplace/test_marketplace_commands.py
@@ -0,0 +1,216 @@
+"""Tests for marketplace CLI commands using CliRunner."""
+
+import json
+from unittest.mock import MagicMock, patch
+
+import pytest
+from click.testing import CliRunner
+
+from apm_cli.marketplace.models import (
+    MarketplaceManifest,
+    MarketplacePlugin,
+    MarketplaceSource,
+)
+
+
+@pytest.fixture
+def runner():
+    return CliRunner()
+
+
+@pytest.fixture(autouse=True)
+def _isolate_config(tmp_path, monkeypatch):
+    """Isolate filesystem writes."""
+    config_dir = str(tmp_path / ".apm")
+    monkeypatch.setattr("apm_cli.config.CONFIG_DIR", config_dir)
+    monkeypatch.setattr("apm_cli.config.CONFIG_FILE", str(tmp_path / ".apm" / "config.json"))
+    monkeypatch.setattr("apm_cli.config._config_cache", None)
+    monkeypatch.setattr("apm_cli.marketplace.registry._registry_cache", None)
+
+
+class TestMarketplaceAdd:
+    """marketplace add OWNER/REPO."""
+
+    def test_invalid_format_no_slash(self, runner):
+        from apm_cli.commands.marketplace import marketplace
+
+        result = runner.invoke(marketplace, ["add", "just-a-name"])
+        assert result.exit_code != 0
+        assert "OWNER/REPO" in result.output
+
+    @patch("apm_cli.marketplace.client.fetch_marketplace")
+    @patch("apm_cli.marketplace.client._auto_detect_path")
+    def test_successful_add(self, mock_detect, mock_fetch, runner):
+        from apm_cli.commands.marketplace import marketplace
+
+        mock_detect.return_value = "marketplace.json"
+        mock_fetch.return_value = MarketplaceManifest(
+            name="Test",
+            plugins=(MarketplacePlugin(name="p1"),),
+        )
+
+        result = runner.invoke(marketplace, ["add", "acme-org/plugins"])
+        assert result.exit_code == 0
+        assert "registered" in result.output.lower() or "1 plugin" in result.output
+
+    @patch("apm_cli.marketplace.client._auto_detect_path")
+    def test_no_marketplace_json_found(self, mock_detect, runner):
+        from apm_cli.commands.marketplace import marketplace
+
+        mock_detect.return_value = None
+        result = runner.invoke(marketplace, ["add", "acme-org/empty-repo"])
+        assert result.exit_code != 0
+        assert "marketplace.json" in result.output
+
+
+class TestMarketplaceList:
+    """marketplace list."""
+
+    def test_empty_list(self, runner):
+        from apm_cli.commands.marketplace import marketplace
+
+        result = runner.invoke(marketplace, ["list"])
+        assert result.exit_code == 0
+        assert "no marketplace" in result.output.lower() or "add" in result.output.lower()
+
+    @patch("apm_cli.marketplace.registry.get_registered_marketplaces")
+    def test_list_with_entries(self, mock_get, runner):
+        from apm_cli.commands.marketplace import marketplace
+
+        mock_get.return_value = [
+            MarketplaceSource(name="acme", owner="acme-org", repo="plugins"),
+        ]
+        result = runner.invoke(marketplace, ["list"])
+        assert result.exit_code == 0
+        assert "acme" in result.output
+
+
+class TestMarketplaceBrowse:
+    """marketplace browse NAME."""
+
+    @patch("apm_cli.marketplace.client.fetch_marketplace")
+    @patch("apm_cli.marketplace.registry.get_marketplace_by_name")
+    def test_browse_shows_plugins(self, mock_get, mock_fetch, runner):
+        from apm_cli.commands.marketplace import marketplace
+
+        mock_get.return_value = MarketplaceSource(
+            name="acme", owner="acme-org", repo="plugins"
+        )
+        mock_fetch.return_value = MarketplaceManifest(
+            name="Acme",
+            plugins=(
+                MarketplacePlugin(name="security-checks", description="Scans"),
+                MarketplacePlugin(name="code-review", description="Reviews"),
+            ),
+        )
+        result = runner.invoke(marketplace, ["browse", "acme"])
+        assert result.exit_code == 0
+        assert "security-checks" in result.output
+
+
+class TestMarketplaceUpdate:
+    """marketplace update [NAME]."""
+
+    @patch("apm_cli.marketplace.client.fetch_marketplace")
+    @patch("apm_cli.marketplace.client.clear_marketplace_cache")
+    @patch("apm_cli.marketplace.registry.get_marketplace_by_name")
+    def test_update_single(self, mock_get, mock_clear, mock_fetch, runner):
+        from apm_cli.commands.marketplace import marketplace
+
+        mock_get.return_value = MarketplaceSource(
+            name="acme", owner="acme-org", repo="plugins"
+        )
+        mock_fetch.return_value = MarketplaceManifest(
+            name="Acme", plugins=(MarketplacePlugin(name="p1"),)
+        )
+        result = runner.invoke(marketplace, ["update", "acme"])
+        assert result.exit_code == 0
+        assert "updated" in result.output.lower() or "1 plugin" in result.output
+
+
+class TestMarketplaceRemove:
+    """marketplace remove NAME."""
+
+    @patch("apm_cli.marketplace.client.clear_marketplace_cache")
+    @patch("apm_cli.marketplace.registry.remove_marketplace")
+    @patch("apm_cli.marketplace.registry.get_marketplace_by_name")
+    def test_remove_with_confirm(self, mock_get, mock_remove, mock_clear, runner):
+        from apm_cli.commands.marketplace import marketplace
+
+        mock_get.return_value = MarketplaceSource(
+            name="acme", owner="acme-org", repo="plugins"
+        )
+        result = runner.invoke(marketplace, ["remove", "acme", "--yes"])
+        assert result.exit_code == 0
+        mock_remove.assert_called_once()
+        assert "removed" in result.output.lower()
+
+
+class TestSearch:
+    """Top-level search command -- requires QUERY@MARKETPLACE format."""
+
+    def test_search_missing_at_symbol(self, runner):
+        from apm_cli.commands.marketplace import search
+
+        result = runner.invoke(search, ["security"])
+        assert result.exit_code != 0
+        assert "QUERY@MARKETPLACE" in result.output
+
+    def test_search_empty_query(self, runner):
+        from apm_cli.commands.marketplace import search
+
+        result = runner.invoke(search, ["@skills"])
+        assert result.exit_code != 0
+        assert "QUERY" in result.output and "MARKETPLACE" in result.output
+
+    def test_search_empty_marketplace(self, runner):
+        from apm_cli.commands.marketplace import search
+
+        result = runner.invoke(search, ["security@"])
+        assert result.exit_code != 0
+        assert "QUERY" in result.output and "MARKETPLACE" in result.output
+
+    @patch("apm_cli.marketplace.registry.get_marketplace_by_name")
+    def test_search_unknown_marketplace(self, mock_get, runner):
+        from apm_cli.commands.marketplace import search
+
+        mock_get.side_effect = Exception("not found")
+        result = runner.invoke(search, ["security@nonexistent"])
+        assert result.exit_code != 0
+        assert "not registered" in result.output.lower()
+
+    @patch("apm_cli.marketplace.client.search_marketplace")
+    @patch("apm_cli.marketplace.registry.get_marketplace_by_name")
+    def test_search_finds_results(self, mock_get, mock_search, runner):
+        from apm_cli.commands.marketplace import search
+
+        mock_get.return_value = MarketplaceSource(
+            name="skills", owner="anthropics", repo="anthropics/skills", path=".claude-plugin/marketplace.json"
+        )
+        mock_search.return_value = [
+            MarketplacePlugin(
+                name="security-scanner",
+                description="Scans code",
+                source_marketplace="skills",
+            ),
+        ]
+        result = runner.invoke(search, ["security@skills"])
+        assert result.exit_code == 0
+        assert "security-scanner" in result.output
+
+    @patch("apm_cli.marketplace.client.search_marketplace")
+    @patch("apm_cli.marketplace.registry.get_marketplace_by_name")
+    def test_search_no_results(self, mock_get, mock_search, runner):
+        from apm_cli.commands.marketplace import search
+
+        mock_get.return_value = MarketplaceSource(
+            name="skills", owner="anthropics", repo="anthropics/skills", path=".claude-plugin/marketplace.json"
+        )
+        mock_search.return_value = []
+        result = runner.invoke(search, ["zzz-nonexistent@skills"])
+        assert result.exit_code == 0
+        assert (
+            "no plugin" in result.output.lower()
+            or "not found" in result.output.lower()
+            or "browse" in result.output.lower()
+        )
diff --git a/tests/unit/marketplace/test_marketplace_errors.py b/tests/unit/marketplace/test_marketplace_errors.py
new file mode 100644
index 000000000..ff45fb1ce
--- /dev/null
+++ b/tests/unit/marketplace/test_marketplace_errors.py
@@ -0,0 +1,47 @@
+"""Tests for marketplace error hierarchy."""
+
+import pytest
+
+from apm_cli.marketplace.errors import (
+    MarketplaceError,
+    MarketplaceFetchError,
+    MarketplaceNotFoundError,
+    PluginNotFoundError,
+)
+
+
+class TestMarketplaceErrors:
+    """Error messages are actionable and include next-step commands."""
+
+    def test_hierarchy(self):
+        assert issubclass(MarketplaceNotFoundError, MarketplaceError)
+        assert issubclass(PluginNotFoundError, MarketplaceError)
+        assert issubclass(MarketplaceFetchError, MarketplaceError)
+        assert issubclass(MarketplaceError, Exception)
+
+    def test_not_found_message(self):
+        err = MarketplaceNotFoundError("acme")
+        assert "acme" in str(err)
+        assert "apm marketplace add" in str(err)
+        assert err.name == "acme"
+
+    def test_plugin_not_found_message(self):
+        err = PluginNotFoundError("my-plugin", "acme")
+        assert "my-plugin" in str(err)
+        assert "acme" in str(err)
+        assert "apm marketplace browse" in str(err)
+        assert err.plugin_name == "my-plugin"
+        assert err.marketplace_name == "acme"
+
+    def test_fetch_error_message(self):
+        err = MarketplaceFetchError("acme", "timeout")
+        assert "acme" in str(err)
+        assert "timeout" in str(err)
+        assert "apm marketplace update" in str(err)
+        assert err.name == "acme"
+        assert err.reason == "timeout"
+
+    def test_fetch_error_no_reason(self):
+        err = MarketplaceFetchError("acme")
+        assert "acme" in str(err)
+        assert "apm marketplace update" in str(err)
diff --git a/tests/unit/marketplace/test_marketplace_install_integration.py b/tests/unit/marketplace/test_marketplace_install_integration.py
new file mode 100644
index 000000000..6399e131f
--- /dev/null
+++ b/tests/unit/marketplace/test_marketplace_install_integration.py
@@ -0,0 +1,62 @@
+"""Tests for the install flow with mocked marketplace resolution."""
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from apm_cli.marketplace.resolver import parse_marketplace_ref
+
+
+class TestInstallMarketplacePreParse:
+    """The pre-parse intercept in _validate_and_add_packages_to_apm_yml."""
+
+    def test_marketplace_ref_detected(self):
+        """NAME@MARKETPLACE triggers marketplace resolution."""
+        result = parse_marketplace_ref("security-checks@acme-tools")
+        assert result == ("security-checks", "acme-tools")
+
+    def test_owner_repo_not_intercepted(self):
+        """owner/repo should NOT be intercepted."""
+        result = parse_marketplace_ref("owner/repo")
+        assert result is None
+
+    def test_owner_repo_at_alias_not_intercepted(self):
+        """owner/repo@alias should NOT be intercepted (has slash)."""
+        result = parse_marketplace_ref("owner/repo@alias")
+        assert result is None
+
+    def test_bare_name_not_intercepted(self):
+        """Just a name without @ should NOT be intercepted."""
+        result = parse_marketplace_ref("just-a-name")
+        assert result is None
+
+    def test_ssh_not_intercepted(self):
+        """SSH URLs should NOT be intercepted (has colon)."""
+        result = parse_marketplace_ref("git@github.com:o/r")
+        assert result is None
+
+
+class TestValidationOutcomeProvenance:
+    """Verify marketplace provenance is attached to ValidationOutcome."""
+
+    def test_outcome_has_provenance_field(self):
+        from apm_cli.core.command_logger import _ValidationOutcome
+
+        outcome = _ValidationOutcome(
+            valid=[("owner/repo", False)],
+            invalid=[],
+            marketplace_provenance={
+                "owner/repo": {
+                    "discovered_via": "acme-tools",
+                    "marketplace_plugin_name": "security-checks",
+                }
+            },
+        )
+        assert outcome.marketplace_provenance is not None
+        assert "owner/repo" in outcome.marketplace_provenance
+
+    def test_outcome_no_provenance(self):
+        from apm_cli.core.command_logger import _ValidationOutcome
+
+        outcome = _ValidationOutcome(valid=[], invalid=[])
+        assert outcome.marketplace_provenance is None
diff --git a/tests/unit/marketplace/test_marketplace_models.py b/tests/unit/marketplace/test_marketplace_models.py
new file mode 100644
index 000000000..1b4e7ba68
--- /dev/null
+++ b/tests/unit/marketplace/test_marketplace_models.py
@@ -0,0 +1,349 @@
+"""Tests for marketplace data models and JSON parser."""
+
+import pytest
+
+from apm_cli.marketplace.models import (
+    MarketplaceManifest,
+    MarketplacePlugin,
+    MarketplaceSource,
+    parse_marketplace_json,
+)
+
+
+class TestMarketplaceSource:
+    """Frozen dataclass for registered marketplace sources."""
+
+    def test_basic_creation(self):
+        src = MarketplaceSource(name="acme", owner="acme-org", repo="plugins")
+        assert src.name == "acme"
+        assert src.owner == "acme-org"
+        assert src.repo == "plugins"
+        assert src.host == "github.com"
+        assert src.branch == "main"
+        assert src.path == "marketplace.json"
+
+    def test_frozen(self):
+        src = MarketplaceSource(name="x", owner="o", repo="r")
+        with pytest.raises(AttributeError):
+            src.name = "y"
+
+    def test_to_dict_defaults(self):
+        src = MarketplaceSource(name="acme", owner="acme-org", repo="plugins")
+        d = src.to_dict()
+        assert d == {"name": "acme", "owner": "acme-org", "repo": "plugins"}
+        assert "host" not in d  # default omitted
+        assert "branch" not in d
+
+    def test_to_dict_non_defaults(self):
+        src = MarketplaceSource(
+            name="acme",
+            owner="acme-org",
+            repo="plugins",
+            host="ghe.corp.com",
+            branch="release",
+            path=".github/plugin/marketplace.json",
+        )
+        d = src.to_dict()
+        assert d["host"] == "ghe.corp.com"
+        assert d["branch"] == "release"
+        assert d["path"] == ".github/plugin/marketplace.json"
+
+    def test_from_dict_minimal(self):
+        src = MarketplaceSource.from_dict(
+            {"name": "acme", "owner": "acme-org", "repo": "plugins"}
+        )
+        assert src.name == "acme"
+        assert src.host == "github.com"
+
+    def test_from_dict_full(self):
+        src = MarketplaceSource.from_dict(
+            {
+                "name": "acme",
+                "owner": "acme-org",
+                "repo": "plugins",
+                "host": "ghe.corp.com",
+                "branch": "release",
+                "path": ".claude-plugin/marketplace.json",
+            }
+        )
+        assert src.host == "ghe.corp.com"
+        assert src.branch == "release"
+        assert src.path == ".claude-plugin/marketplace.json"
+
+    def test_roundtrip(self):
+        original = MarketplaceSource(
+            name="acme",
+            owner="acme-org",
+            repo="plugins",
+            host="ghe.corp.com",
+            branch="release",
+        )
+        restored = MarketplaceSource.from_dict(original.to_dict())
+        assert restored == original
+
+
+class TestMarketplacePlugin:
+    """Frozen dataclass for plugin entries."""
+
+    def test_basic_creation(self):
+        p = MarketplacePlugin(name="my-plugin", description="A plugin")
+        assert p.name == "my-plugin"
+        assert p.description == "A plugin"
+        assert p.tags == ()
+        assert p.source is None
+
+    def test_frozen(self):
+        p = MarketplacePlugin(name="x")
+        with pytest.raises(AttributeError):
+            p.name = "y"
+
+    def test_matches_query_name(self):
+        p = MarketplacePlugin(name="security-checks", description="Scan for vulns")
+        assert p.matches_query("security")
+        assert p.matches_query("SECURITY")
+
+    def test_matches_query_description(self):
+        p = MarketplacePlugin(name="x", description="Scan for vulnerabilities")
+        assert p.matches_query("vuln")
+
+    def test_matches_query_tags(self):
+        p = MarketplacePlugin(name="x", tags=("security", "audit"))
+        assert p.matches_query("audit")
+
+    def test_no_match(self):
+        p = MarketplacePlugin(name="x", description="desc", tags=("a",))
+        assert not p.matches_query("zzz")
+
+
+class TestMarketplaceManifest:
+    """Frozen dataclass for parsed marketplace content."""
+
+    def test_find_plugin(self):
+        plugins = (
+            MarketplacePlugin(name="alpha"),
+            MarketplacePlugin(name="beta"),
+        )
+        m = MarketplaceManifest(name="test", plugins=plugins)
+        assert m.find_plugin("alpha").name == "alpha"
+        assert m.find_plugin("BETA").name == "beta"
+        assert m.find_plugin("gamma") is None
+
+    def test_search(self):
+        plugins = (
+            MarketplacePlugin(name="security-scanner", description="Scans stuff"),
+            MarketplacePlugin(name="code-formatter", description="Formats code"),
+        )
+        m = MarketplaceManifest(name="test", plugins=plugins)
+        results = m.search("security")
+        assert len(results) == 1
+        assert results[0].name == "security-scanner"
+
+
+class TestParseMarketplaceJson:
+    """Parser for both Copilot CLI and Claude Code formats."""
+
+    def test_copilot_format(self):
+        data = {
+            "name": "Acme Tools",
+            "description": "Corporate plugins",
+            "plugins": [
+                {
+                    "name": "security-checks",
+                    "description": "Security scanning",
+                    "repository": "acme-org/security-plugin",
+                    "ref": "v1.3.0",
+                },
+                {
+                    "name": "code-review",
+                    "description": "Code review helper",
+                    "repository": "acme-org/review-plugin",
+                },
+            ],
+        }
+        manifest = parse_marketplace_json(data, "acme-tools")
+        assert manifest.name == "Acme Tools"
+        assert manifest.description == "Corporate plugins"
+        assert len(manifest.plugins) == 2
+        p1 = manifest.find_plugin("security-checks")
+        assert p1.source == {"type": "github", "repo": "acme-org/security-plugin", "ref": "v1.3.0"}
+        p2 = manifest.find_plugin("code-review")
+        assert p2.source == {"type": "github", "repo": "acme-org/review-plugin"}
+
+    def test_claude_format_github(self):
+        data = {
+            "name": "Claude Plugins",
+            "plugins": [
+                {
+                    "name": "my-plugin",
+                    "description": "A plugin",
+                    "source": {
+                        "type": "github",
+                        "repo": "owner/plugin-repo",
+                        "ref": "v2.0",
+                    },
+                }
+            ],
+        }
+        manifest = parse_marketplace_json(data, "claude-mkt")
+        assert len(manifest.plugins) == 1
+        p = manifest.plugins[0]
+        assert p.name == "my-plugin"
+        assert p.source["type"] == "github"
+        assert p.source["repo"] == "owner/plugin-repo"
+        assert p.source_marketplace == "claude-mkt"
+
+    def test_claude_format_relative(self):
+        data = {
+            "name": "Test",
+            "plugins": [
+                {"name": "local-plugin", "source": "./plugins/local"},
+            ],
+        }
+        manifest = parse_marketplace_json(data)
+        assert len(manifest.plugins) == 1
+        assert manifest.plugins[0].source == "./plugins/local"
+
+    def test_claude_format_url(self):
+        data = {
+            "name": "Test",
+            "plugins": [
+                {
+                    "name": "url-plugin",
+                    "source": {"type": "url", "url": "https://github.com/org/repo"},
+                }
+            ],
+        }
+        manifest = parse_marketplace_json(data)
+        assert len(manifest.plugins) == 1
+        assert manifest.plugins[0].source["type"] == "url"
+
+    def test_claude_format_git_subdir(self):
+        data = {
+            "name": "Test",
+            "plugins": [
+                {
+                    "name": "subdir-plugin",
+                    "source": {
+                        "type": "git-subdir",
+                        "repo": "owner/monorepo",
+                        "subdir": "packages/plugin-a",
+                        "ref": "main",
+                    },
+                }
+            ],
+        }
+        manifest = parse_marketplace_json(data)
+        assert len(manifest.plugins) == 1
+        assert manifest.plugins[0].source["type"] == "git-subdir"
+
+    def test_npm_source_skipped(self):
+        data = {
+            "name": "Test",
+            "plugins": [
+                {
+                    "name": "npm-plugin",
+                    "source": {"type": "npm", "package": "@scope/pkg"},
+                }
+            ],
+        }
+        manifest = parse_marketplace_json(data)
+        assert len(manifest.plugins) == 0
+
+    def test_copilot_cli_source_key_as_type(self):
+        """Copilot CLI format uses 'source' (not 'type') as type discriminator inside dict."""
+        data = {
+            "name": "Awesome Copilot",
+            "plugins": [
+                {
+                    "name": "azure",
+                    "description": "Azure skills",
+                    "source": {
+                        "source": "github",
+                        "repo": "microsoft/azure-skills",
+                        "path": ".github/plugins/azure-skills",
+                    },
+                }
+            ],
+        }
+        manifest = parse_marketplace_json(data, "awesome-copilot")
+        assert len(manifest.plugins) == 1
+        p = manifest.plugins[0]
+        assert p.name == "azure"
+        # Parser should normalize "source" key to "type" key
+        assert p.source["type"] == "github"
+        assert p.source["repo"] == "microsoft/azure-skills"
+        assert p.source["path"] == ".github/plugins/azure-skills"
+
+    def test_npm_via_source_key_skipped(self):
+        """npm source type should be skipped even when using 'source' key."""
+        data = {
+            "name": "Test",
+            "plugins": [
+                {
+                    "name": "npm-plugin",
+                    "source": {"source": "npm", "package": "@scope/pkg"},
+                }
+            ],
+        }
+        manifest = parse_marketplace_json(data)
+        assert len(manifest.plugins) == 0
+
+    def test_invalid_entries_skipped(self):
+        data = {
+            "name": "Test",
+            "plugins": [
+                {"name": "valid", "repository": "o/r"},
+                {"description": "no name"},  # Missing name
+                "not-a-dict",  # Non-dict
+                {"name": "no-source"},  # Missing source
+            ],
+        }
+        manifest = parse_marketplace_json(data)
+        assert len(manifest.plugins) == 1
+        assert manifest.plugins[0].name == "valid"
+
+    def test_empty_plugins_list(self):
+        data = {"name": "Empty"}
+        manifest = parse_marketplace_json(data)
+        assert len(manifest.plugins) == 0
+
+    def test_plugins_not_a_list(self):
+        data = {"name": "Bad", "plugins": "not-a-list"}
+        manifest = parse_marketplace_json(data)
+        assert len(manifest.plugins) == 0
+
+    def test_tags_preserved(self):
+        data = {
+            "name": "Test",
+            "plugins": [
+                {
+                    "name": "tagged",
+                    "repository": "o/r",
+                    "tags": ["security", "compliance"],
+                }
+            ],
+        }
+        manifest = parse_marketplace_json(data)
+        assert manifest.plugins[0].tags == ("security", "compliance")
+
+    def test_version_preserved(self):
+        data = {
+            "name": "Test",
+            "plugins": [
+                {"name": "versioned", "repository": "o/r", "version": "1.2.3"},
+            ],
+        }
+        manifest = parse_marketplace_json(data)
+        assert manifest.plugins[0].version == "1.2.3"
+
+    def test_owner_string(self):
+        """Owner can be a string (not a dict)."""
+        data = {"name": "Test", "owner": "John Doe", "plugins": []}
+        manifest = parse_marketplace_json(data)
+        assert manifest.owner_name == "John Doe"
+
+    def test_owner_dict(self):
+        """Owner can be a dict with 'name' key."""
+        data = {"name": "Test", "owner": {"name": "Jane"}, "plugins": []}
+        manifest = parse_marketplace_json(data)
+        assert manifest.owner_name == "Jane"
diff --git a/tests/unit/marketplace/test_marketplace_registry.py b/tests/unit/marketplace/test_marketplace_registry.py
new file mode 100644
index 000000000..3dedf85df
--- /dev/null
+++ b/tests/unit/marketplace/test_marketplace_registry.py
@@ -0,0 +1,98 @@
+"""Tests for marketplace registry CRUD with tmp_path isolation."""
+
+import json
+
+import pytest
+
+from apm_cli.marketplace.errors import MarketplaceNotFoundError
+from apm_cli.marketplace.models import MarketplaceSource
+from apm_cli.marketplace import registry as registry_mod
+
+
+@pytest.fixture(autouse=True)
+def _isolate_registry(tmp_path, monkeypatch):
+    """Isolate registry reads/writes to a temp directory."""
+    config_dir = str(tmp_path / ".apm")
+    monkeypatch.setattr("apm_cli.marketplace.registry._registry_cache", None)
+    monkeypatch.setattr("apm_cli.config.CONFIG_DIR", config_dir)
+    monkeypatch.setattr("apm_cli.config.CONFIG_FILE", str(tmp_path / ".apm" / "config.json"))
+    monkeypatch.setattr("apm_cli.config._config_cache", None)
+    yield
+
+
+class TestRegistryBasicOps:
+    """CRUD operations on marketplace registry."""
+
+    def test_empty_registry(self):
+        assert registry_mod.get_registered_marketplaces() == []
+        assert registry_mod.marketplace_count() == 0
+
+    def test_add_and_get(self):
+        src = MarketplaceSource(name="acme", owner="acme-org", repo="plugins")
+        registry_mod.add_marketplace(src)
+        assert registry_mod.marketplace_count() == 1
+
+        fetched = registry_mod.get_marketplace_by_name("acme")
+        assert fetched.name == "acme"
+        assert fetched.owner == "acme-org"
+
+    def test_add_replaces_same_name(self):
+        src1 = MarketplaceSource(name="acme", owner="old-org", repo="plugins")
+        src2 = MarketplaceSource(name="acme", owner="new-org", repo="plugins")
+        registry_mod.add_marketplace(src1)
+        registry_mod.add_marketplace(src2)
+        assert registry_mod.marketplace_count() == 1
+        assert registry_mod.get_marketplace_by_name("acme").owner == "new-org"
+
+    def test_add_case_insensitive_replace(self):
+        src1 = MarketplaceSource(name="Acme", owner="old", repo="r")
+        src2 = MarketplaceSource(name="acme", owner="new", repo="r")
+        registry_mod.add_marketplace(src1)
+        registry_mod.add_marketplace(src2)
+        assert registry_mod.marketplace_count() == 1
+
+    def test_remove(self):
+        src = MarketplaceSource(name="acme", owner="o", repo="r")
+        registry_mod.add_marketplace(src)
+        registry_mod.remove_marketplace("acme")
+        assert registry_mod.marketplace_count() == 0
+
+    def test_remove_not_found(self):
+        with pytest.raises(MarketplaceNotFoundError):
+            registry_mod.remove_marketplace("nonexistent")
+
+    def test_get_not_found(self):
+        with pytest.raises(MarketplaceNotFoundError):
+            registry_mod.get_marketplace_by_name("nonexistent")
+
+    def test_marketplace_names(self):
+        registry_mod.add_marketplace(
+            MarketplaceSource(name="beta", owner="o", repo="r")
+        )
+        registry_mod.add_marketplace(
+            MarketplaceSource(name="alpha", owner="o", repo="r")
+        )
+        assert registry_mod.marketplace_names() == ["alpha", "beta"]
+
+
+class TestRegistryPersistence:
+    """Verify data survives cache invalidation."""
+
+    def test_persists_to_disk(self, tmp_path):
+        src = MarketplaceSource(name="acme", owner="acme-org", repo="plugins")
+        registry_mod.add_marketplace(src)
+
+        # Invalidate cache
+        registry_mod._invalidate_cache()
+
+        # Should reload from disk
+        fetched = registry_mod.get_marketplace_by_name("acme")
+        assert fetched.owner == "acme-org"
+
+    def test_corrupted_file_returns_empty(self, tmp_path):
+        path = registry_mod._ensure_file()
+        with open(path, "w") as f:
+            f.write("not json")
+
+        registry_mod._invalidate_cache()
+        assert registry_mod.get_registered_marketplaces() == []
diff --git a/tests/unit/marketplace/test_marketplace_resolver.py b/tests/unit/marketplace/test_marketplace_resolver.py
new file mode 100644
index 000000000..9dbdda051
--- /dev/null
+++ b/tests/unit/marketplace/test_marketplace_resolver.py
@@ -0,0 +1,249 @@
+"""Tests for marketplace resolver -- regex and source type resolution."""
+
+import pytest
+
+from apm_cli.marketplace.models import MarketplacePlugin
+from apm_cli.marketplace.resolver import (
+    _resolve_github_source,
+    _resolve_git_subdir_source,
+    _resolve_relative_source,
+    _resolve_url_source,
+    parse_marketplace_ref,
+    resolve_plugin_source,
+)
+
+
+class TestParseMarketplaceRef:
+    """Regex positive/negative cases for NAME@MARKETPLACE detection."""
+
+    # Positive cases -- valid marketplace refs
+    def test_simple(self):
+        assert parse_marketplace_ref("security-checks@acme-tools") == (
+            "security-checks",
+            "acme-tools",
+        )
+
+    def test_dots(self):
+        assert parse_marketplace_ref("my.plugin@my.marketplace") == (
+            "my.plugin",
+            "my.marketplace",
+        )
+
+    def test_underscores(self):
+        assert parse_marketplace_ref("my_plugin@my_marketplace") == (
+            "my_plugin",
+            "my_marketplace",
+        )
+
+    def test_mixed(self):
+        assert parse_marketplace_ref("plugin-v2.0@corp_tools") == (
+            "plugin-v2.0",
+            "corp_tools",
+        )
+
+    def test_whitespace_stripped(self):
+        assert parse_marketplace_ref("  name@mkt  ") == ("name", "mkt")
+
+    # Negative cases -- not marketplace refs (should return None)
+    def test_owner_repo(self):
+        """owner/repo has slash -> rejected."""
+        assert parse_marketplace_ref("owner/repo") is None
+
+    def test_owner_repo_at_alias(self):
+        """owner/repo@alias has slash -> rejected."""
+        assert parse_marketplace_ref("owner/repo@alias") is None
+
+    def test_ssh_url(self):
+        """git@host:... has colon -> rejected."""
+        assert parse_marketplace_ref("git@github.com:o/r") is None
+
+    def test_https_url(self):
+        """https://... has slashes -> rejected."""
+        assert parse_marketplace_ref("https://github.com/o/r") is None
+
+    def test_no_at(self):
+        """Bare name without @ is NOT a marketplace ref."""
+        assert parse_marketplace_ref("just-a-name") is None
+
+    def test_empty(self):
+        assert parse_marketplace_ref("") is None
+
+    def test_only_at(self):
+        """Just @ with no name/marketplace."""
+        assert parse_marketplace_ref("@") is None
+
+    def test_at_prefix(self):
+        """@marketplace with no name."""
+        assert parse_marketplace_ref("@mkt") is None
+
+    def test_at_suffix(self):
+        """name@ with no marketplace."""
+        assert parse_marketplace_ref("name@") is None
+
+    def test_multiple_at(self):
+        """Multiple @ signs."""
+        assert parse_marketplace_ref("a@b@c") is None
+
+    def test_special_chars(self):
+        """Special characters that aren't in the allowed set."""
+        assert parse_marketplace_ref("name@mkt!") is None
+        assert parse_marketplace_ref("na me@mkt") is None
+
+
+class TestResolveGithubSource:
+    """Resolve github source type."""
+
+    def test_with_ref(self):
+        assert _resolve_github_source({"repo": "owner/repo", "ref": "v1.0"}) == "owner/repo#v1.0"
+
+    def test_without_ref(self):
+        assert _resolve_github_source({"repo": "owner/repo"}) == "owner/repo"
+
+    def test_with_path(self):
+        """Copilot CLI format uses 'path' for subdirectory."""
+        result = _resolve_github_source({
+            "repo": "microsoft/azure-skills",
+            "path": ".github/plugins/azure-skills",
+        })
+        assert result == "microsoft/azure-skills/.github/plugins/azure-skills"
+
+    def test_with_path_and_ref(self):
+        result = _resolve_github_source({
+            "repo": "owner/mono",
+            "path": "plugins/foo",
+            "ref": "v2.0",
+        })
+        assert result == "owner/mono/plugins/foo#v2.0"
+
+    def test_path_traversal_rejected(self):
+        with pytest.raises(ValueError, match="traversal sequence"):
+            _resolve_github_source({"repo": "owner/repo", "path": "../escape"})
+
+    def test_invalid_repo(self):
+        with pytest.raises(ValueError, match="owner/repo"):
+            _resolve_github_source({"repo": "just-a-name"})
+
+
+class TestResolveUrlSource:
+    """Resolve url source type."""
+
+    def test_github_https(self):
+        assert _resolve_url_source({"url": "https://github.com/owner/repo"}) == "owner/repo"
+
+    def test_github_https_with_git_suffix(self):
+        assert _resolve_url_source({"url": "https://github.com/owner/repo.git"}) == "owner/repo"
+
+    def test_non_github_url(self):
+        with pytest.raises(ValueError, match="Cannot resolve URL source"):
+            _resolve_url_source({"url": "https://gitlab.com/owner/repo"})
+
+
+class TestResolveGitSubdirSource:
+    """Resolve git-subdir source type."""
+
+    def test_with_ref(self):
+        result = _resolve_git_subdir_source({
+            "repo": "owner/monorepo",
+            "subdir": "packages/plugin-a",
+            "ref": "main",
+        })
+        assert result == "owner/monorepo/packages/plugin-a#main"
+
+    def test_without_ref(self):
+        result = _resolve_git_subdir_source({"repo": "owner/monorepo"})
+        assert result == "owner/monorepo"
+
+    def test_without_subdir(self):
+        result = _resolve_git_subdir_source({"repo": "owner/monorepo", "ref": "v1"})
+        assert result == "owner/monorepo#v1"
+
+    def test_invalid_repo(self):
+        with pytest.raises(ValueError, match="owner/repo"):
+            _resolve_git_subdir_source({"repo": "bad"})
+
+    def test_path_traversal_rejected(self):
+        with pytest.raises(ValueError, match="traversal sequence"):
+            _resolve_git_subdir_source({"repo": "owner/mono", "subdir": "../escape"})
+
+
+class TestResolveRelativeSource:
+    """Resolve relative path source type."""
+
+    def test_relative_path(self):
+        result = _resolve_relative_source("./plugins/my-plugin", "acme-org", "marketplace")
+        assert result == "acme-org/marketplace/plugins/my-plugin"
+
+    def test_root_relative(self):
+        result = _resolve_relative_source(".", "acme-org", "marketplace")
+        assert result == "acme-org/marketplace"
+
+    def test_path_traversal_rejected(self):
+        with pytest.raises(ValueError, match="traversal sequence"):
+            _resolve_relative_source("../escape", "acme-org", "marketplace")
+
+
+class TestResolvePluginSource:
+    """Integration of all source type resolvers."""
+
+    def test_github_source(self):
+        p = MarketplacePlugin(
+            name="test",
+            source={"type": "github", "repo": "owner/repo", "ref": "v1.0"},
+        )
+        assert resolve_plugin_source(p) == "owner/repo#v1.0"
+
+    def test_github_source_with_path(self):
+        """Copilot CLI format: github source with 'path' field."""
+        p = MarketplacePlugin(
+            name="azure",
+            source={
+                "type": "github",
+                "repo": "microsoft/azure-skills",
+                "path": ".github/plugins/azure-skills",
+            },
+        )
+        assert resolve_plugin_source(p) == "microsoft/azure-skills/.github/plugins/azure-skills"
+
+    def test_url_source(self):
+        p = MarketplacePlugin(
+            name="test",
+            source={"type": "url", "url": "https://github.com/owner/repo"},
+        )
+        assert resolve_plugin_source(p) == "owner/repo"
+
+    def test_git_subdir_source(self):
+        p = MarketplacePlugin(
+            name="test",
+            source={
+                "type": "git-subdir",
+                "repo": "owner/mono",
+                "subdir": "pkg/a",
+                "ref": "main",
+            },
+        )
+        assert resolve_plugin_source(p) == "owner/mono/pkg/a#main"
+
+    def test_relative_source(self):
+        p = MarketplacePlugin(name="test", source="./plugins/local")
+        assert resolve_plugin_source(p, "acme", "mkt") == "acme/mkt/plugins/local"
+
+    def test_npm_source_rejected(self):
+        p = MarketplacePlugin(
+            name="test",
+            source={"type": "npm", "package": "@scope/pkg"},
+        )
+        with pytest.raises(ValueError, match="npm source type"):
+            resolve_plugin_source(p)
+
+    def test_unknown_source_type_rejected(self):
+        p = MarketplacePlugin(
+            name="test",
+            source={"type": "unknown"},
+        )
+        with pytest.raises(ValueError, match="unsupported source type"):
+            resolve_plugin_source(p)
+
+    def test_no_source_rejected(self):
+        p = MarketplacePlugin(name="test", source=None)
+        with pytest.raises(ValueError, match="no source defined"):
+            resolve_plugin_source(p)
