wedge: T013 property tests + T014 Dockerfile merged; fix zero-amount install rows

T013 worker found record_install violating CHECK(amount>0) when platform_pct
rounds cut or seller_amount to zero (unhandled IntegrityError -> HTTP 500).
Fixed: skip zero rows, grant references the paying tx; regression test added.
9 ledger tests green.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

cli/src/m2_market/apply/local.py

@@ -0,0 +1,240 @@
+"""Local apply adapter (contracts/apply-adapter.md §local, frozen v1 seam).
+
+Places a Solution bundle's ``payload/`` into the operator's home per
+``recipe.yaml``, runs post-install checks, and records InstallState.
+Idempotent (re-apply of an already-applied changeset is a no-op) and
+reversible (rollback restores backups / removes newly-placed files) per
+constitution VI (canary-first, reversible deployment).
+
+Bundle layout (task T021):
+    <bundle>/solution.json   # optional here; solution_id/version/content_hash if present
+    <bundle>/payload/...     # files referenced by recipe targets
+    <bundle>/recipe.yaml     # {targets: [{src, dest}], checks: [{cmd, expect_exit}], entrypoint}
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import os
+import shutil
+import subprocess
+from pathlib import Path
+
+import yaml
+
+from .base import ApplyResult, Changeset, ChangesetItem, InstallState
+
+_CHECK_KIND = "check"
+_PLACEMENT_KIND = "placement"
+
+
+class LocalApplyAdapter:
+    """`ApplyAdapter` implementation for the "local" adapter (v1 default)."""
+
+    name = "local"
+
+    def __init__(self) -> None:
+        # Set by plan(); apply()/verify() reuse it since the protocol's apply()
+        # signature carries only the changeset, not the state.
+        self._state: InstallState | None = None
+
+    # -- ApplyAdapter protocol -------------------------------------------------
+
+    def plan(self, bundle: Path, state: InstallState) -> Changeset:
+        self._state = state
+        bundle = Path(bundle)
+        recipe = self._load_recipe(bundle)
+        solution = self._load_solution(bundle)
+        home = Path.home().resolve()
+
+        items = self._build_placement_items(bundle, recipe, home)
+        items += self._build_check_items(recipe)
+
+        listing_id = solution.get("solution_id") or bundle.name
+        version = solution.get("version") or solution.get("solution_version") or "0"
+        content_hash = solution.get("content_hash")
+
+        changeset_hash = self._hash_items(items, content_hash)
+
+        return Changeset(
+            listing_id=listing_id,
+            version=str(version),
+            items=items,
+            changeset_hash=changeset_hash,
+        )
+
+    def apply(self, changeset: Changeset) -> ApplyResult:
+        state = self._state or InstallState()
+
+        existing = state.get(changeset.listing_id)
+        if (
+            existing is not None
+            and existing.get("changeset_hash") == changeset.changeset_hash
+            and existing.get("status") == "applied"
+        ):
+            return ApplyResult(ok=True, applied=[], errors=[])
+
+        placed: list[ChangesetItem] = []
+        errors: list[str] = []
+
+        for item in changeset.items:
+            if item.kind != _PLACEMENT_KIND:
+                continue
+            try:
+                self._place(item)
+                placed.append(item)
+            except OSError as exc:
+                errors.append(f"place {item.src} -> {item.dest}: {exc}")
+                break
+
+        if not errors:
+            errors.extend(self._run_checks(changeset.items))
+
+        if errors:
+            self.rollback(Changeset(changeset.listing_id, changeset.version, placed, changeset.changeset_hash))
+            state.record_install(
+                changeset.listing_id, changeset.version, "", changeset.changeset_hash, "failed"
+            )
+            return ApplyResult(ok=False, applied=[], errors=errors)
+
+        state.record_install(
+            changeset.listing_id, changeset.version, "", changeset.changeset_hash, "applied"
+        )
+        return ApplyResult(ok=True, applied=placed, errors=[])
+
+    def verify(self, bundle: Path, state: InstallState) -> bool:
+        bundle = Path(bundle)
+        recipe = self._load_recipe(bundle)
+        home = Path.home().resolve()
+        items = self._build_placement_items(bundle, recipe, home)
+
+        if not all(Path(item.dest).exists() for item in items):
+            return False
+
+        return not self._run_checks(items + self._build_check_items(recipe))
+
+    def rollback(self, changeset: Changeset) -> None:
+        for item in changeset.items:
+            if item.kind != _PLACEMENT_KIND:
+                continue
+            dest = Path(item.dest)
+            if item.backup:
+                backup = Path(item.backup)
+                if backup.exists():
+                    if dest.exists():
+                        self._remove(dest)
+                    shutil.move(str(backup), str(dest))
+                    item.backup = None
+            elif dest.exists():
+                self._remove(dest)
+
+    # -- helpers ----------------------------------------------------------------
+
+    @staticmethod
+    def _load_recipe(bundle: Path) -> dict:
+        recipe_path = bundle / "recipe.yaml"
+        with recipe_path.open(encoding="utf-8") as f:
+            return yaml.safe_load(f) or {}
+
+    @staticmethod
+    def _load_solution(bundle: Path) -> dict:
+        solution_path = bundle / "solution.json"
+        if not solution_path.exists():
+            return {}
+        with solution_path.open(encoding="utf-8") as f:
+            return json.load(f)
+
+    @staticmethod
+    def _resolve_dest(dest_raw: str, home: Path) -> Path:
+        dest = Path(os.path.expanduser(dest_raw))
+        if not dest.is_absolute():
+            dest = home / dest
+        dest = Path(os.path.normpath(str(dest)))
+
+        if dest != home and home not in dest.parents:
+            raise ValueError(
+                f"recipe target dest {dest_raw!r} resolves to {dest}, which is outside "
+                f"the user home {home} — refusing for safety (contracts/apply-adapter.md)"
+            )
+        return dest
+
+    def _build_placement_items(self, bundle: Path, recipe: dict, home: Path) -> list[ChangesetItem]:
+        items = []
+        for target in recipe.get("targets", []):
+            src = (bundle / "payload" / target["src"]).resolve()
+            dest = self._resolve_dest(target["dest"], home)
+            items.append(ChangesetItem(kind=_PLACEMENT_KIND, src=str(src), dest=str(dest)))
+        return items
+
+    @staticmethod
+    def _build_check_items(recipe: dict) -> list[ChangesetItem]:
+        items = []
+        for check in recipe.get("checks", []):
+            items.append(
+                ChangesetItem(
+                    kind=_CHECK_KIND,
+                    src=check["cmd"],
+                    dest=str(check.get("expect_exit", 0)),
+                )
+            )
+        return items
+
+    @staticmethod
+    def _hash_items(items: list[ChangesetItem], content_hash: str | None) -> str:
+        digest = hashlib.sha256()
+        pairs = sorted((item.src, item.dest) for item in items)
+        for src, dest in pairs:
+            digest.update(src.encode("utf-8"))
+            digest.update(b"\x00")
+            digest.update(dest.encode("utf-8"))
+            digest.update(b"\n")
+        if content_hash:
+            digest.update(content_hash.encode("utf-8"))
+        return f"sha256:{digest.hexdigest()}"
+
+    @staticmethod
+    def _remove(path: Path) -> None:
+        if path.is_dir() and not path.is_symlink():
+            shutil.rmtree(path)
+        else:
+            path.unlink()
+
+    @classmethod
+    def _place(cls, item: ChangesetItem) -> None:
+        src = Path(item.src)
+        dest = Path(item.dest)
+        dest.parent.mkdir(parents=True, exist_ok=True)
+
+        if dest.exists() or dest.is_symlink():
+            backup = f"{dest}.m2bak-{_timestamp()}"
+            shutil.move(str(dest), backup)
+            item.backup = backup
+
+        if src.is_dir():
+            shutil.copytree(src, dest)
+        else:
+            shutil.copy2(src, dest)
+
+    @staticmethod
+    def _run_checks(items: list[ChangesetItem]) -> list[str]:
+        errors = []
+        for item in items:
+            if item.kind != _CHECK_KIND:
+                continue
+            expect_exit = int(item.dest)
+            result = subprocess.run(
+                item.src, shell=True, capture_output=True, text=True, check=False
+            )
+            if result.returncode != expect_exit:
+                errors.append(
+                    f"check failed: {item.src!r} exit={result.returncode} "
+                    f"expected={expect_exit} stderr={result.stderr.strip()}"
+                )
+        return errors
+
+
+def _timestamp() -> str:
+    from datetime import datetime, timezone
+
+    return datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%S%fZ")

cli/src/m2_market/apply/m2core_sync.py

@@ -17,7 +17,7 @@ _MSG = (
 )
 
 
-class M2CoreSyncAdapter:
+class M2CoreSyncApplyAdapter:
     name = "m2core-sync"
 
     def plan(self, bundle: Path, state: InstallState) -> Changeset:

cli/src/m2_market/cli.py

@@ -2,12 +2,23 @@
 
 from __future__ import annotations
 
+import socket
 import sys
+import tarfile
+import tempfile
+import uuid
+from pathlib import Path
 
 import click
 
 from m2_market import __version__
-from m2_market.config import ConfigError, load_config
+from m2_market.apply import RailsNotLanded, get_adapter
+from m2_market.apply.base import InstallState
+from m2_market.catalog import CatalogClient
+from m2_market.config import Config, ConfigError, load_config
+from m2_market.ledger import InsufficientFunds, LedgerClient
+from m2_market.output import emit_json, format_price, render_kv, render_table, search_results_table
+from m2_market.registry import RegistryClient, RegistryError
 
 # Exit-code map (contracts/cli.md).
 EXIT_OK = 0  # ok / no-op
@@ -46,7 +57,29 @@ def _not_implemented(ctx: click.Context, command: str) -> None:
 @click.pass_context
 def search(ctx: click.Context, query: str, listing_type: str | None, limit: int | None) -> None:
     """Semantic catalog query, tenant-filtered."""
-    _not_implemented(ctx, "search")
+    config: Config = ctx.obj["config"]
+    with CatalogClient(config.memory_api_url, config.memory_api_key, config.tenant) as catalog:
+        items = catalog.search(query, limit=limit or 10)
+
+    if listing_type:
+        items = [
+            item
+            for item in items
+            if (item.get("listing") or item).get("inventory_type") == listing_type
+        ]
+
+    if ctx.obj["json"]:
+        emit_json(items)
+    else:
+        search_results_table(items)
+
+
+def _fetch_solution(registry: RegistryClient, listing_id: str) -> dict:
+    """Best-effort solution lookup — permissions/entrypoint are a nice-to-have, not fatal."""
+    try:
+        return registry.get_solution(listing_id)
+    except Exception:
+        return {}
 
 
 @main.command()
@@ -54,16 +87,180 @@ def search(ctx: click.Context, query: str, listing_type: str | None, limit: int
 @click.pass_context
 def show(ctx: click.Context, listing_id: str) -> None:
     """Full listing detail: evidence, price, permissions diff, seller, version."""
-    _not_implemented(ctx, "show")
+    config: Config = ctx.obj["config"]
+    json_output = ctx.obj["json"]
+
+    with CatalogClient(config.memory_api_url, config.memory_api_key, config.tenant) as catalog:
+        listing = catalog.get(listing_id)
+
+    if listing is None:
+        click.echo(f"m2-market show: listing not found: {listing_id}", err=True)
+        ctx.exit(EXIT_NOT_FOUND)
+        return
+
+    with RegistryClient(config.forgejo_url, config.forgejo_token) as registry:
+        solution = _fetch_solution(registry, listing_id)
+
+    permissions = solution.get("permissions") or []
+
+    if json_output:
+        emit_json({"listing": listing, "solution": solution, "permissions": permissions})
+        return
+
+    click.echo(f"{listing.get('name', listing_id)} ({listing_id})")
+    click.echo(listing.get("summary", ""))
+    click.echo()
+    render_kv(
+        [
+            ("evidence_summary", listing.get("evidence_summary", "-")),
+            ("price", format_price(listing.get("price"))),
+            ("seller", listing.get("seller", "-")),
+            ("version", listing.get("solution_version", "-")),
+            ("install_ref", listing.get("install_ref", "-")),
+        ]
+    )
+    if permissions:
+        click.echo("\nPermissions:")
+        for perm in permissions:
+            click.echo(f"  - {perm}")
+
+
+def _extract_bundle(bundle_path: Path, extract_dir: Path) -> Path:
+    with tarfile.open(bundle_path) as tar:
+        tar.extractall(extract_dir, filter="data")
+
+    entries = list(extract_dir.iterdir())
+    if len(entries) == 1 and entries[0].is_dir():
+        return entries[0]
+    return extract_dir
+
+
+def _print_entrypoint_hint(json_output: bool, solution: dict, extra: dict | None = None) -> None:
+    entrypoint = (solution.get("deployment") or {}).get("entrypoint")
+    if json_output:
+        payload = {"status": "ok", "entrypoint": entrypoint}
+        if extra:
+            payload.update(extra)
+        emit_json(payload)
+    elif entrypoint:
+        click.echo(f"Installed. Run: {entrypoint}")
+    else:
+        click.echo("Installed.")
 
 
 @main.command()
 @click.argument("listing_id")
 @click.option("--yes", is_flag=True, default=False, help="Skip interactive confirmation.")
+@click.option("--adapter", "adapter_name", default=None, help="Override the configured apply adapter.")
 @click.pass_context
-def install(ctx: click.Context, listing_id: str, yes: bool) -> None:
+def install(ctx: click.Context, listing_id: str, yes: bool, adapter_name: str | None) -> None:
-    """Resolve, pay, fetch, apply, and record a listing install."""
+    """Resolve, pay, fetch, apply, and record a listing install (contracts/cli.md FR-005 sequence)."""
-    _not_implemented(ctx, "install")
+    config: Config = ctx.obj["config"]
+    json_output = ctx.obj["json"]
+
+    state = InstallState()
+
+    with LedgerClient(config.ledger_url, config.ledger_api_key) as ledger:
+        # Idempotent re-run guard (Story 1 AC-3): if we've already applied this
+        # listing and a license grant still exists, no-op BEFORE debiting.
+        cached = state.get(listing_id)
+        if cached is not None and cached.get("status") == "applied":
+            grant = ledger.license(config.operator_id, listing_id)
+            if grant is not None:
+                with RegistryClient(config.forgejo_url, config.forgejo_token) as registry:
+                    solution = _fetch_solution(registry, listing_id)
+                if not json_output:
+                    click.echo(f"{listing_id} already installed — no-op.")
+                _print_entrypoint_hint(json_output, solution, {"listing_id": listing_id, "noop": True})
+                return
+
+        with CatalogClient(config.memory_api_url, config.memory_api_key, config.tenant) as catalog:
+            listing = catalog.get(listing_id)
+
+        if listing is None:
+            click.echo(f"m2-market install: listing not found: {listing_id}", err=True)
+            ctx.exit(EXIT_NOT_FOUND)
+            return
+
+        with RegistryClient(config.forgejo_url, config.forgejo_token) as registry:
+            solution = _fetch_solution(registry, listing_id)
+            price = listing.get("price") or {}
+            permissions = solution.get("permissions") or []
+
+            if not json_output:
+                click.echo(f"{listing.get('name', listing_id)} — {format_price(price)}")
+                if permissions:
+                    click.echo("Permissions requested:")
+                    for perm in permissions:
+                        click.echo(f"  - {perm}")
+
+                if not yes and not click.confirm("Proceed with install?", default=False):
+                    return
+
+            ref = f"{listing_id}:{socket.gethostname()}:{uuid.uuid4()}"
+            seller = listing.get("seller") or solution.get("seller")
+            amount = price.get("amount")
+
+            try:
+                grant = ledger.install(buyer=config.operator_id, seller=seller, amount=amount, ref=ref)
+            except InsufficientFunds as exc:
+                click.echo(f"m2-market install: insufficient funds: {exc}", err=True)
+                ctx.exit(EXIT_INSUFFICIENT_FUNDS)
+                return
+
+            install_ref = listing.get("install_ref")
+            content_hash = solution.get("content_hash")
+
+            with tempfile.TemporaryDirectory(prefix="m2-market-bundle-") as tmp_dir_str:
+                tmp_dir = Path(tmp_dir_str)
+                try:
+                    bundle_path = registry.download_bundle(install_ref, tmp_dir / "download")
+                    if content_hash and not RegistryClient.verify_bundle(bundle_path, content_hash):
+                        raise RegistryError(f"content hash mismatch for {install_ref}")
+                    bundle_root = _extract_bundle(bundle_path, tmp_dir / "extracted")
+                except (RegistryError, OSError, tarfile.TarError) as exc:
+                    ledger.refund(ref)
+                    click.echo(
+                        f"m2-market install: bundle fetch/verify failed, refunded: {exc}",
+                        err=True,
+                    )
+                    ctx.exit(EXIT_APPLY_FAILED)
+                    return
+
+                adapter_choice = adapter_name or config.apply_adapter
+                try:
+                    adapter = get_adapter(adapter_choice)
+                    changeset = adapter.plan(bundle_root, state)
+                    apply_result = adapter.apply(changeset)
+                    if not apply_result.ok:
+                        raise RuntimeError("; ".join(apply_result.errors) or "apply reported failure")
+                    if not adapter.verify(bundle_root, state):
+                        raise RuntimeError("post-install verification failed")
+                except RailsNotLanded as exc:
+                    ledger.refund(ref)
+                    click.echo(f"m2-market install: {exc}", err=True)
+                    ctx.exit(EXIT_RAILS_NOT_LANDED)
+                    return
+                except Exception as exc:
+                    ledger.refund(ref)
+                    click.echo(
+                        f"m2-market install: apply failed, refunded: {exc}\n"
+                        "Remediation: inspect the recipe's post-install checks, fix the "
+                        "underlying issue, then retry `m2-market install`.",
+                        err=True,
+                    )
+                    ctx.exit(EXIT_APPLY_FAILED)
+                    return
+
+                state.record_install(
+                    listing_id,
+                    listing.get("solution_version") or changeset.version,
+                    grant.get("grant_id", ""),
+                    changeset.changeset_hash,
+                    "applied",
+                )
+
+    _print_entrypoint_hint(json_output, solution, {"listing_id": listing_id, "noop": False})
 
 
 @main.command()
@@ -71,7 +268,36 @@ def install(ctx: click.Context, listing_id: str, yes: bool) -> None:
 @click.pass_context
 def wallet(ctx: click.Context, operator: str | None) -> None:
     """Balance + recent transactions from the ledger."""
-    _not_implemented(ctx, "wallet")
+    config: Config = ctx.obj["config"]
+    json_output = ctx.obj["json"]
+    operator_id = operator or config.operator_id
+
+    with LedgerClient(config.ledger_url, config.ledger_api_key) as ledger:
+        balance = ledger.balance(operator_id)
+        txs = ledger.transactions(operator_id=operator_id)
+
+    recent = list(reversed(txs))[:10]
+
+    if json_output:
+        emit_json({"balance": balance, "transactions": recent})
+        return
+
+    click.echo(f"Operator: {operator_id}")
+    click.echo(f"Balance:  {balance.get('balance')} (as of {balance.get('as_of')})")
+    click.echo()
+    headers = ["ts", "from", "to", "amount", "reason", "ref"]
+    rows = [
+        [
+            str(tx.get("ts", "-")),
+            str(tx.get("from", "-")),
+            str(tx.get("to", "-")),
+            str(tx.get("amount", "-")),
+            str(tx.get("reason", "-")),
+            str(tx.get("ref", "-")),
+        ]
+        for tx in recent
+    ]
+    render_table(headers, rows)
 
 
 @main.command()

cli/src/m2_market/output.py

@@ -0,0 +1,64 @@
+"""Table/JSON rendering helpers for the m2-market CLI (contracts/cli.md)."""
+
+from __future__ import annotations
+
+import json as _json
+
+import click
+
+
+def emit_json(data: object) -> None:
+    click.echo(_json.dumps(data, indent=2, sort_keys=True, default=str))
+
+
+def _col_widths(headers: list[str], rows: list[list[str]]) -> list[int]:
+    widths = [len(h) for h in headers]
+    for row in rows:
+        for i, cell in enumerate(row):
+            widths[i] = max(widths[i], len(cell))
+    return widths
+
+
+def render_table(headers: list[str], rows: list[list[str]]) -> None:
+    if not rows:
+        click.echo("(no results)")
+        return
+    widths = _col_widths(headers, rows)
+    line = "  ".join(h.ljust(w) for h, w in zip(headers, widths))
+    click.echo(line)
+    click.echo("  ".join("-" * w for w in widths))
+    for row in rows:
+        click.echo("  ".join(cell.ljust(w) for cell, w in zip(row, widths)))
+
+
+def format_price(price: dict | None) -> str:
+    if not price:
+        return "-"
+    amount = price.get("amount")
+    currency = price.get("currency", "")
+    return f"{amount} {currency}".strip()
+
+
+def search_results_table(items: list[dict]) -> None:
+    headers = ["listing_id", "name", "summary", "price", "installs", "rating"]
+    rows = []
+    for item in items:
+        listing = item.get("listing", item)
+        stats = listing.get("stats") or {}
+        rows.append(
+            [
+                str(item.get("listing_id", listing.get("listing_id", "-"))),
+                str(listing.get("name", "-")),
+                str(listing.get("summary", "-")),
+                format_price(listing.get("price")),
+                str(stats.get("installs", "-")),
+                str(stats.get("rating", "-")),
+            ]
+        )
+    render_table(headers, rows)
+
+
+def render_kv(pairs: list[tuple[str, str]]) -> None:
+    width = max((len(k) for k, _ in pairs), default=0)
+    for key, value in pairs:
+        click.echo(f"{key.ljust(width)} : {value}")

cli/tests/test_cli_install.py

@@ -0,0 +1,308 @@
+"""Inline verification for the install/search/show/wallet CLI commands.
+
+Wires httpx.MockTransport fakes for the ledger/registry/catalog clients (per
+contracts/ledger-api.md, contracts/registry-layout.md, contracts/catalog-index.md)
+and a fake ApplyAdapter (the T021 local adapter lands on a sibling branch and
+isn't merged into this worktree yet — the CLI must not depend on its internals).
+"""
+
+from __future__ import annotations
+
+import base64
+import io
+import json
+import tarfile
+
+import httpx
+import pytest
+from click.testing import CliRunner
+
+import m2_market.cli as cli_mod
+from m2_market.apply.base import ApplyResult, Changeset
+
+LISTING_ID = "lst_pdf_export"
+INSTALL_REF = "lst_pdf_export-v1.0.0"
+OPERATOR = "buyer1"
+SELLER = "seller1"
+
+LISTING = {
+    "listing_id": LISTING_ID,
+    "solution_id": "sol_pdf_export",
+    "solution_version": "1.0.0",
+    "inventory_type": "solution",
+    "name": "PDF Export",
+    "summary": "Branded PDF reports.",
+    "category": "productivity",
+    "price": {"amount": 20, "currency": "m2cr", "model": "fixed"},
+    "seller": SELLER,
+    "evidence_summary": "Used on 12 sessions, saved 40 min avg.",
+    "tenant_visibility": ["*"],
+    "stats": {"installs": 3, "rating": 4.5, "proposals_shown": 5, "proposals_accepted": 3},
+    "status": "published",
+    "install_ref": INSTALL_REF,
+}
+
+SOLUTION = {
+    "schema_version": "m2.solution.v1",
+    "solution_id": "sol_pdf_export",
+    "name": "PDF Export",
+    "summary": "Branded PDF reports.",
+    "intent": "export pdfs",
+    "deployment": {
+        "recipe_ref": "recipe.yaml",
+        "entrypoint": "m2-pdf-export --help",
+        "verify_command": "true",
+    },
+    "applicability": {},
+    "tenant_scope": "m2-core",
+    "evidence": [{"source": "session-123"}],
+    "content_hash": "",
+    "price": {"amount": 20, "currency": "m2cr", "model": "fixed"},
+    "seller": SELLER,
+    "license": {"terms": "internal", "major_version_coverage": True},
+    "permissions": ["fs:write:~/reports"],
+}
+
+
+def _make_bundle_bytes() -> bytes:
+    buf = io.BytesIO()
+    with tarfile.open(fileobj=buf, mode="w:gz") as tar:
+        data = b"entrypoint stub\n"
+        info = tarfile.TarInfo(name="bundle/recipe.yaml")
+        info.size = len(data)
+        tar.addfile(info, io.BytesIO(data))
+    return buf.getvalue()
+
+
+BUNDLE_BYTES = _make_bundle_bytes()
+
+
+class FakeState:
+    """Records calls made against the fake ledger/registry backends."""
+
+    def __init__(self):
+        self.install_calls = 0
+        self.refund_calls = 0
+        self.license_exists = False
+        self.insufficient_funds = False
+
+
+def _b64_json(payload: dict) -> str:
+    return base64.b64encode(json.dumps(payload).encode("utf-8")).decode("ascii")
+
+
+def make_handler(state: FakeState):
+    def handler(request: httpx.Request) -> httpx.Response:
+        host = request.url.host
+        path = request.url.path
+
+        if host == "ledger.test":
+            if path == "/tx/install" and request.method == "POST":
+                state.install_calls += 1
+                if state.insufficient_funds:
+                    return httpx.Response(402, json={"error": "insufficient_funds", "balance": 0})
+                return httpx.Response(
+                    200,
+                    json={
+                        "tx_ids": ["tx1", "tx2"],
+                        "grant": {"grant_id": "grant-1", "listing_id": LISTING_ID, "operator_id": OPERATOR},
+                    },
+                )
+            if path == "/tx/refund" and request.method == "POST":
+                state.refund_calls += 1
+                return httpx.Response(200, json={"tx_ids": ["tx-refund"]})
+            if path.startswith("/license/"):
+                if state.license_exists:
+                    return httpx.Response(200, json={"grant": {"grant_id": "grant-1"}})
+                return httpx.Response(404, json={"error": "not_found"})
+            if path.startswith("/balance/"):
+                return httpx.Response(200, json={"operator_id": OPERATOR, "balance": 80, "as_of": "now"})
+            if path == "/tx":
+                return httpx.Response(
+                    200,
+                    json={
+                        "transactions": [
+                            {"ts": "t1", "from": OPERATOR, "to": SELLER, "amount": 18, "reason": "install", "ref": "r1"},
+                            {"ts": "t2", "from": OPERATOR, "to": "platform", "amount": 2, "reason": "install", "ref": "r1"},
+                        ]
+                    },
+                )
+
+        if host == "registry.test":
+            if path.endswith("/contents/listings/lst_pdf_export/solution.json"):
+                return httpx.Response(200, json={"content": _b64_json(SOLUTION)})
+            if path.endswith(f"/releases/tags/{INSTALL_REF}"):
+                return httpx.Response(
+                    200,
+                    json={
+                        "assets": [
+                            {
+                                "name": "bundle.tar.gz",
+                                "browser_download_url": "https://registry.test/download/bundle.tar.gz",
+                            }
+                        ]
+                    },
+                )
+            if path == "/download/bundle.tar.gz":
+                return httpx.Response(200, content=BUNDLE_BYTES)
+
+        if host == "catalog.test" and path == "/search":
+            body = json.loads(request.content)
+            if body["query_text"] == LISTING_ID:
+                return httpx.Response(200, json={"items": [{"listing_id": LISTING_ID, "score": 0.9, "listing": LISTING}]})
+            return httpx.Response(200, json={"items": [{"listing_id": LISTING_ID, "score": 0.8, "listing": LISTING}]})
+
+        raise AssertionError(f"unexpected request: {request.method} {request.url}")
+
+    return handler
+
+
+@pytest.fixture
+def fake_backend(monkeypatch):
+    state = FakeState()
+    handler = make_handler(state)
+
+    orig_init = httpx.Client.__init__
+
+    def patched_init(self, *args, **kwargs):
+        kwargs["transport"] = httpx.MockTransport(handler)
+        orig_init(self, *args, **kwargs)
+
+    monkeypatch.setattr(httpx.Client, "__init__", patched_init)
+    return state
+
+
+@pytest.fixture
+def config_env(tmp_path, monkeypatch):
+    config_path = tmp_path / "config.toml"
+    config_path.write_text(
+        f"""
+operator_id = "{OPERATOR}"
+tenant = "test-tenant"
+ledger_url = "https://ledger.test"
+ledger_api_key = "lk"
+memory_api_url = "https://catalog.test"
+memory_api_key = "mk"
+forgejo_url = "https://registry.test"
+forgejo_token = "rt"
+apply_adapter = "fake"
+"""
+    )
+    state_path = tmp_path / "state.json"
+    monkeypatch.setenv("M2_MARKET_CONFIG", str(config_path))
+    monkeypatch.setenv("M2_MARKET_STATE", str(state_path))
+    return state_path
+
+
+class FakeAdapterCalls:
+    def __init__(self):
+        self.plan_calls = 0
+        self.apply_calls = 0
+        self.verify_calls = 0
+        self.apply_should_fail = False
+
+
+def make_fake_get_adapter(calls: FakeAdapterCalls):
+    class FakeAdapter:
+        name = "fake"
+
+        def plan(self, bundle, state):
+            calls.plan_calls += 1
+            return Changeset(listing_id="sol_pdf_export", version="1.0.0", items=[], changeset_hash="sha256:abc")
+
+        def apply(self, changeset):
+            calls.apply_calls += 1
+            if calls.apply_should_fail:
+                return ApplyResult(ok=False, applied=[], errors=["recipe check failed"])
+            return ApplyResult(ok=True, applied=[], errors=[])
+
+        def verify(self, bundle, state):
+            calls.verify_calls += 1
+            return True
+
+        def rollback(self, changeset):
+            pass
+
+    def get_adapter(name):
+        return FakeAdapter()
+
+    return get_adapter
+
+
+def test_funded_install_happy_path(fake_backend, config_env, monkeypatch):
+    calls = FakeAdapterCalls()
+    monkeypatch.setattr(cli_mod, "get_adapter", make_fake_get_adapter(calls))
+
+    runner = CliRunner()
+    result = runner.invoke(cli_mod.main, ["install", LISTING_ID, "--yes"])
+
+    assert result.exit_code == 0, result.output
+    assert "m2-pdf-export --help" in result.output
+    assert fake_backend.install_calls == 1
+    assert fake_backend.refund_calls == 0
+    assert calls.plan_calls == 1
+    assert calls.apply_calls == 1
+    assert calls.verify_calls == 1
+
+
+def test_insufficient_funds_no_adapter_call(fake_backend, config_env, monkeypatch):
+    fake_backend.insufficient_funds = True
+    calls = FakeAdapterCalls()
+    monkeypatch.setattr(cli_mod, "get_adapter", make_fake_get_adapter(calls))
+
+    runner = CliRunner()
+    result = runner.invoke(cli_mod.main, ["install", LISTING_ID, "--yes"])
+
+    assert result.exit_code == cli_mod.EXIT_INSUFFICIENT_FUNDS, result.output
+    assert calls.plan_calls == 0
+    assert calls.apply_calls == 0
+    assert fake_backend.refund_calls == 0
+
+
+def test_rerun_is_idempotent_noop_without_second_debit(fake_backend, config_env, monkeypatch):
+    calls = FakeAdapterCalls()
+    monkeypatch.setattr(cli_mod, "get_adapter", make_fake_get_adapter(calls))
+
+    runner = CliRunner()
+    first = runner.invoke(cli_mod.main, ["install", LISTING_ID, "--yes"])
+    assert first.exit_code == 0, first.output
+    assert fake_backend.install_calls == 1
+
+    fake_backend.license_exists = True
+    second = runner.invoke(cli_mod.main, ["install", LISTING_ID, "--yes"])
+
+    assert second.exit_code == 0, second.output
+    assert fake_backend.install_calls == 1  # no second debit
+    assert calls.plan_calls == 1  # adapter not invoked again
+
+
+def test_apply_failure_triggers_refund_and_exit_4(fake_backend, config_env, monkeypatch):
+    calls = FakeAdapterCalls()
+    calls.apply_should_fail = True
+    monkeypatch.setattr(cli_mod, "get_adapter", make_fake_get_adapter(calls))
+
+    runner = CliRunner()
+    result = runner.invoke(cli_mod.main, ["install", LISTING_ID, "--yes"])
+
+    assert result.exit_code == cli_mod.EXIT_APPLY_FAILED, result.output
+    assert fake_backend.install_calls == 1
+    assert fake_backend.refund_calls == 1
+    assert "refunded" in result.output
+
+
+def test_wallet_prints_balance_and_recent_tx(fake_backend, config_env):
+    runner = CliRunner()
+    result = runner.invoke(cli_mod.main, ["wallet"])
+
+    assert result.exit_code == 0, result.output
+    assert "80" in result.output
+    assert "install" in result.output
+
+
+def test_search_renders_table(fake_backend, config_env):
+    runner = CliRunner()
+    result = runner.invoke(cli_mod.main, ["search", "pdf report"])
+
+    assert result.exit_code == 0, result.output
+    assert LISTING_ID in result.output
+    assert "PDF Export" in result.output

indexer/pyproject.toml

@@ -5,10 +5,12 @@ description = "Registry → market:catalog sync; reindex = full rebuild (contrac
 requires-python = ">=3.12"
 dependencies = [
     "httpx>=0.27",
+    "jsonschema>=4.23",
 ]
 
 [project.scripts]
 m2-market-indexer = "m2_market_indexer.reindex:main"
+m2-market-indexer-watch = "m2_market_indexer.watch:main"
 
 [dependency-groups]
 dev = ["pytest>=8"]

indexer/src/m2_market_indexer/__init__.py

@@ -0,0 +1,6 @@
+"""m2-market-indexer: registry -> market:catalog sync (contracts/catalog-index.md)."""
+
+from .client import MemoryClient, RegistryClient
+from .config import IndexerConfig, load_config
+
+__all__ = ["MemoryClient", "RegistryClient", "IndexerConfig", "load_config"]

indexer/src/m2_market_indexer/client.py

@@ -0,0 +1,175 @@
+"""Registry (Forgejo) + market:catalog (memory-api) clients.
+
+Shared by reindex.py and watch.py. Registry read side mirrors
+cli/src/m2_market/registry.py's contents-API approach (contracts/registry-layout.md);
+this module adds the listing-directory-listing and commit-polling calls the
+indexer needs that the CLI client doesn't. Memory-api paths are configurable
+constants (config.py), not hardcoded, per contracts/catalog-index.md.
+"""
+
+from __future__ import annotations
+
+import base64
+import json
+
+import httpx
+
+
+class RegistryError(RuntimeError):
+    """Raised when the registry returns an unexpected response."""
+
+
+class RegistryClient:
+    def __init__(
+        self,
+        base_url: str,
+        token: str,
+        repo: str,
+        ref: str = "main",
+        *,
+        auth_user: str = "m2",
+        timeout: float = 15.0,
+        transport: httpx.BaseTransport | None = None,
+    ) -> None:
+        owner, _, name = repo.partition("/")
+        self._owner = owner
+        self._repo = name
+        self.ref = ref
+        self._client = httpx.Client(
+            base_url=base_url.rstrip("/"),
+            auth=(auth_user, token),
+            timeout=timeout,
+            transport=transport,
+        )
+
+    def close(self) -> None:
+        self._client.close()
+
+    def __enter__(self) -> RegistryClient:
+        return self
+
+    def __exit__(self, *exc_info: object) -> None:
+        self.close()
+
+    def _repo_api(self, path: str) -> str:
+        return f"/api/v1/repos/{self._owner}/{self._repo}{path}"
+
+    def _get_json_contents(self, repo_path: str) -> dict:
+        resp = self._client.get(
+            self._repo_api(f"/contents/{repo_path}"), params={"ref": self.ref}
+        )
+        resp.raise_for_status()
+        data = resp.json()
+        content = base64.b64decode(data["content"])
+        return json.loads(content)
+
+    def list_listing_ids(self) -> list[str]:
+        """Directory names under listings/ on `self.ref` (one per listing)."""
+        resp = self._client.get(
+            self._repo_api("/contents/listings"), params={"ref": self.ref}
+        )
+        if resp.status_code == 404:
+            return []
+        resp.raise_for_status()
+        entries = resp.json()
+        return sorted(e["name"] for e in entries if e.get("type") == "dir")
+
+    def get_listing(self, listing_id: str) -> dict:
+        return self._get_json_contents(f"listings/{listing_id}/listing.json")
+
+    def get_solution(self, listing_id: str) -> dict | None:
+        """solution.json is optional (used for `intent`); None if absent."""
+        try:
+            return self._get_json_contents(f"listings/{listing_id}/solution.json")
+        except httpx.HTTPStatusError as exc:
+            if exc.response.status_code == 404:
+                return None
+            raise
+
+    def get_main_commit_sha(self) -> str:
+        resp = self._client.get(self._repo_api(f"/branches/{self.ref}"))
+        resp.raise_for_status()
+        data = resp.json()
+        try:
+            return data["commit"]["id"]
+        except KeyError as exc:
+            raise RegistryError(f"unexpected branch response shape: {data!r}") from exc
+
+
+class MemoryError_(RuntimeError):
+    """Raised when memory-api returns an unexpected response."""
+
+
+class MemoryClient:
+    """Writer for the `market:catalog` partition (contracts/catalog-index.md).
+
+    ONLY m2-market-indexer writes here (reindex + watch); everything else is a
+    reader via CatalogClient's `/search`.
+    """
+
+    def __init__(
+        self,
+        base_url: str,
+        api_key: str,
+        partition: str,
+        *,
+        partition_path: str,
+        document_path: str,
+        timeout: float = 15.0,
+        transport: httpx.BaseTransport | None = None,
+    ) -> None:
+        self.partition = partition
+        self.partition_path = partition_path
+        self.document_path = document_path
+        self._client = httpx.Client(
+            base_url=base_url.rstrip("/"),
+            headers={"X-API-Key": api_key},
+            timeout=timeout,
+            transport=transport,
+        )
+
+    def close(self) -> None:
+        self._client.close()
+
+    def __enter__(self) -> MemoryClient:
+        return self
+
+    def __exit__(self, *exc_info: object) -> None:
+        self.close()
+
+    def _partition_url(self) -> str:
+        return self.partition_path.format(partition=self.partition)
+
+    def _document_url(self, doc_id: str) -> str:
+        return self.document_path.format(partition=self.partition, doc_id=doc_id)
+
+    def drop_partition(self) -> None:
+        """Delete the partition wholesale; a later upsert recreates it."""
+        resp = self._client.delete(self._partition_url())
+        if resp.status_code not in (200, 204, 404):
+            resp.raise_for_status()
+
+    def upsert_document(self, doc_id: str, text: str, payload: dict) -> None:
+        resp = self._client.put(
+            self._document_url(doc_id),
+            json={"text": text, "payload": payload},
+        )
+        resp.raise_for_status()
+
+    def delete_document(self, doc_id: str) -> None:
+        resp = self._client.delete(self._document_url(doc_id))
+        if resp.status_code not in (200, 204, 404):
+            resp.raise_for_status()
+
+
+def build_embedded_text(listing: dict, solution: dict | None) -> str:
+    """name + summary + keywords + intent(solution.json, if present) + evidence_summary."""
+    parts = [
+        listing.get("name", ""),
+        listing.get("summary", ""),
+        " ".join(listing.get("keywords", []) or []),
+    ]
+    if solution and solution.get("intent"):
+        parts.append(solution["intent"])
+    parts.append(listing.get("evidence_summary", ""))
+    return "\n".join(p for p in parts if p)

indexer/src/m2_market_indexer/config.py

@@ -0,0 +1,82 @@
+"""Env-driven configuration for m2-market-indexer.
+
+The indexer runs standalone (Coolify service / cron, contracts/catalog-index.md
+"Writers" section) — unlike the CLI it has no per-machine config.toml, so every
+knob is an env var with a sane default (constitution VI: no secrets in repos or
+images; keys injected at runtime).
+"""
+
+from __future__ import annotations
+
+import os
+from dataclasses import dataclass
+from pathlib import Path
+
+# --- registry (Forgejo) -----------------------------------------------------
+FORGEJO_URL_DEFAULT = "https://git.machinemachine.ai"
+REGISTRY_REPO_DEFAULT = "m2/market-registry"
+REGISTRY_REF_DEFAULT = "main"
+
+# --- memory-api (market:catalog partition) ----------------------------------
+MEMORY_API_URL_DEFAULT = "https://memory.machinemachine.ai"
+MEMORY_API_PARTITION_DEFAULT = "market:catalog"
+
+# Paths are configurable constants (env-overridable) rather than hardcoded
+# strings, so a memory-api route change doesn't require a code change.
+MEMORY_API_PARTITION_PATH_DEFAULT = "/partitions/{partition}"
+MEMORY_API_DOCUMENT_PATH_DEFAULT = "/partitions/{partition}/documents/{doc_id}"
+
+# --- watch state -------------------------------------------------------------
+STATE_PATH_DEFAULT = Path.home() / ".m2-market-indexer" / "state.json"
+POLL_INTERVAL_SECONDS_DEFAULT = 300
+
+
+@dataclass
+class IndexerConfig:
+    forgejo_url: str
+    forgejo_token: str
+    registry_repo: str
+    registry_ref: str
+    memory_api_url: str
+    memory_api_key: str
+    memory_api_partition: str
+    memory_api_partition_path: str
+    memory_api_document_path: str
+    state_path: Path
+    poll_interval_seconds: int
+
+
+def load_config() -> IndexerConfig:
+    """Read all indexer settings from the environment, falling back to defaults."""
+    return IndexerConfig(
+        forgejo_url=os.environ.get("FORGEJO_URL", FORGEJO_URL_DEFAULT),
+        forgejo_token=os.environ.get("FORGEJO_TOKEN", ""),
+        registry_repo=os.environ.get("REGISTRY_REPO", REGISTRY_REPO_DEFAULT),
+        registry_ref=os.environ.get("REGISTRY_REF", REGISTRY_REF_DEFAULT),
+        memory_api_url=os.environ.get("MEMORY_API_URL", MEMORY_API_URL_DEFAULT),
+        memory_api_key=os.environ.get("MEMORY_API_KEY", ""),
+        memory_api_partition=os.environ.get(
+            "MEMORY_API_PARTITION", MEMORY_API_PARTITION_DEFAULT
+        ),
+        memory_api_partition_path=os.environ.get(
+            "MEMORY_API_PARTITION_PATH", MEMORY_API_PARTITION_PATH_DEFAULT
+        ),
+        memory_api_document_path=os.environ.get(
+            "MEMORY_API_DOCUMENT_PATH", MEMORY_API_DOCUMENT_PATH_DEFAULT
+        ),
+        state_path=Path(
+            os.environ.get("M2_MARKET_INDEXER_STATE", str(STATE_PATH_DEFAULT))
+        ).expanduser(),
+        poll_interval_seconds=int(
+            os.environ.get("M2_MARKET_INDEXER_INTERVAL", POLL_INTERVAL_SECONDS_DEFAULT)
+        ),
+    )
+
+
+def listing_schema_path() -> Path:
+    """Resolve schemas/listing.schema.json, env-overridable for non-monorepo installs."""
+    override = os.environ.get("LISTING_SCHEMA_PATH")
+    if override:
+        return Path(override).expanduser()
+    # indexer/src/m2_market_indexer/config.py -> repo root is 4 parents up.
+    return Path(__file__).resolve().parents[3] / "schemas" / "listing.schema.json"

indexer/src/m2_market_indexer/reindex.py

@@ -0,0 +1,93 @@
+"""Full rebuild of market:catalog from the registry `main` (contracts/catalog-index.md).
+
+Drop partition, then upsert every `published` listing found under `listings/`
+on the registry's default branch. Invalid listing.json documents are skipped
+with a warning, not fatal — one bad listing must not block the rebuild
+(constitution VII: smallest thing that closes the loop).
+"""
+
+from __future__ import annotations
+
+import sys
+
+import httpx
+from jsonschema import Draft202012Validator
+
+from .client import MemoryClient, RegistryClient, build_embedded_text
+from .config import listing_schema_path, load_config
+
+
+def _load_validator() -> Draft202012Validator:
+    import json
+
+    with listing_schema_path().open() as f:
+        schema = json.load(f)
+    return Draft202012Validator(schema)
+
+
+def run(registry: RegistryClient, memory: MemoryClient, validator: Draft202012Validator) -> dict:
+    """Execute one full reindex; returns counts for reporting."""
+    counts = {"scanned": 0, "invalid": 0, "not_published": 0, "upserted": 0}
+
+    listing_ids = registry.list_listing_ids()
+    memory.drop_partition()
+
+    for listing_id in listing_ids:
+        counts["scanned"] += 1
+        try:
+            listing = registry.get_listing(listing_id)
+        except (httpx.HTTPStatusError, ValueError) as exc:
+            print(f"WARN: skipping {listing_id}: cannot read listing.json ({exc})", file=sys.stderr)
+            counts["invalid"] += 1
+            continue
+
+        errors = list(validator.iter_errors(listing))
+        if errors:
+            print(
+                f"WARN: skipping {listing_id}: schema violations: "
+                + "; ".join(e.message for e in errors),
+                file=sys.stderr,
+            )
+            counts["invalid"] += 1
+            continue
+
+        if listing.get("status") != "published":
+            counts["not_published"] += 1
+            continue
+
+        solution = registry.get_solution(listing_id)
+        text = build_embedded_text(listing, solution)
+        memory.upsert_document(listing_id, text, listing)
+        counts["upserted"] += 1
+
+    return counts
+
+
+def main() -> None:
+    config = load_config()
+    validator = _load_validator()
+
+    with RegistryClient(
+        config.forgejo_url,
+        config.forgejo_token,
+        config.registry_repo,
+        config.registry_ref,
+    ) as registry, MemoryClient(
+        config.memory_api_url,
+        config.memory_api_key,
+        config.memory_api_partition,
+        partition_path=config.memory_api_partition_path,
+        document_path=config.memory_api_document_path,
+    ) as memory:
+        counts = run(registry, memory, validator)
+
+    print(
+        f"reindex complete: scanned={counts['scanned']} "
+        f"upserted={counts['upserted']} "
+        f"not_published={counts['not_published']} "
+        f"invalid_skipped={counts['invalid']}"
+    )
+
+
+if __name__ == "__main__":
+    main()

indexer/src/m2_market_indexer/watch.py

@@ -0,0 +1,160 @@
+"""Incremental market:catalog sync: poll registry main tip, upsert deltas.
+
+contracts/catalog-index.md: "watch — on registry merge (webhook or poll ≤5 min),
+upsert changed listings, remove delisted." State (last-seen commit + per-listing
+content hash) is cached in a state file so a no-op poll costs one API call.
+"""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import sys
+import time
+from pathlib import Path
+
+import httpx
+from jsonschema import Draft202012Validator
+
+from .client import MemoryClient, RegistryClient, build_embedded_text
+from .config import IndexerConfig, listing_schema_path, load_config
+
+
+def _load_validator() -> Draft202012Validator:
+    with listing_schema_path().open() as f:
+        schema = json.load(f)
+    return Draft202012Validator(schema)
+
+
+def load_state(path: Path) -> dict:
+    if not path.is_file():
+        return {"last_commit": None, "listings": {}}
+    with path.open() as f:
+        return json.load(f)
+
+
+def save_state(path: Path, state: dict) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    tmp = path.with_suffix(".json.tmp")
+    with tmp.open("w") as f:
+        json.dump(state, f, indent=2)
+    tmp.replace(path)
+
+
+def _content_hash(listing: dict) -> str:
+    return hashlib.sha256(
+        json.dumps(listing, sort_keys=True).encode("utf-8")
+    ).hexdigest()
+
+
+def poll_once(
+    registry: RegistryClient,
+    memory: MemoryClient,
+    validator: Draft202012Validator,
+    state: dict,
+) -> tuple[dict, dict]:
+    """One poll cycle. Returns (new_state, summary). No-ops if the tip is unchanged."""
+    tip = registry.get_main_commit_sha()
+    if tip == state.get("last_commit"):
+        return state, {"changed": False, "upserted": [], "deleted": []}
+
+    prev_listings = state.get("listings", {})
+    new_listings: dict = {}
+    upserted: list[str] = []
+    deleted: list[str] = []
+    seen_ids: set[str] = set()
+
+    for listing_id in registry.list_listing_ids():
+        seen_ids.add(listing_id)
+        try:
+            listing = registry.get_listing(listing_id)
+        except (httpx.HTTPStatusError, ValueError) as exc:
+            print(f"WARN: skipping {listing_id}: cannot read listing.json ({exc})", file=sys.stderr)
+            # Leave any prior state entry as-is; retried on the next poll.
+            if listing_id in prev_listings:
+                new_listings[listing_id] = prev_listings[listing_id]
+            continue
+
+        errors = list(validator.iter_errors(listing))
+        if errors:
+            print(
+                f"WARN: skipping {listing_id}: schema violations: "
+                + "; ".join(e.message for e in errors),
+                file=sys.stderr,
+            )
+            if listing_id in prev_listings:
+                new_listings[listing_id] = prev_listings[listing_id]
+            continue
+
+        status = listing.get("status")
+        content_hash = _content_hash(listing)
+        prev = prev_listings.get(listing_id)
+
+        if status == "published":
+            changed = (
+                prev is None
+                or prev.get("hash") != content_hash
+                or prev.get("status") != "published"
+            )
+            if changed:
+                solution = registry.get_solution(listing_id)
+                text = build_embedded_text(listing, solution)
+                memory.upsert_document(listing_id, text, listing)
+                upserted.append(listing_id)
+        elif prev is not None and prev.get("status") == "published":
+            memory.delete_document(listing_id)
+            deleted.append(listing_id)
+
+        new_listings[listing_id] = {"status": status, "hash": content_hash}
+
+    # Listing directories that disappeared entirely from the registry.
+    for listing_id, meta in prev_listings.items():
+        if listing_id not in seen_ids and meta.get("status") == "published":
+            memory.delete_document(listing_id)
+            deleted.append(listing_id)
+
+    new_state = {"last_commit": tip, "listings": new_listings}
+    return new_state, {"changed": True, "upserted": upserted, "deleted": deleted}
+
+
+def _run_loop(config: IndexerConfig, interval: int) -> None:
+    validator = _load_validator()
+    state = load_state(config.state_path)
+
+    with RegistryClient(
+        config.forgejo_url,
+        config.forgejo_token,
+        config.registry_repo,
+        config.registry_ref,
+    ) as registry, MemoryClient(
+        config.memory_api_url,
+        config.memory_api_key,
+        config.memory_api_partition,
+        partition_path=config.memory_api_partition_path,
+        document_path=config.memory_api_document_path,
+    ) as memory:
+        while True:
+            state, summary = poll_once(registry, memory, validator, state)
+            save_state(config.state_path, state)
+            if summary["changed"]:
+                print(
+                    f"watch: upserted={summary['upserted']} deleted={summary['deleted']}"
+                )
+            else:
+                print("watch: no change")
+            time.sleep(interval)
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(prog="m2-market-indexer-watch")
+    parser.add_argument("--interval", type=int, default=None, help="poll interval in seconds")
+    args = parser.parse_args()
+
+    config = load_config()
+    interval = args.interval if args.interval is not None else config.poll_interval_seconds
+    _run_loop(config, interval)
+
+
+if __name__ == "__main__":
+    main()

indexer/tests/test_reindex_flow.py

@@ -0,0 +1,145 @@
+"""Inline httpx MockTransport test for the reindex flow (T015 verify step).
+
+Registry fixture: one `published` + one `delisted` listing. Asserts the
+resulting memory-api calls are exactly: drop partition, then upsert the
+published listing only (the delisted one is silently omitted by a
+drop-then-upsert-published rebuild — contracts/catalog-index.md "reindex").
+
+Run directly: `python indexer/tests/test_reindex_flow.py`
+Or via pytest: `pytest indexer/tests/test_reindex_flow.py`
+"""
+
+from __future__ import annotations
+
+import base64
+import json
+import sys
+from pathlib import Path
+
+import httpx
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
+
+from jsonschema import Draft202012Validator  # noqa: E402
+from m2_market_indexer.client import MemoryClient, RegistryClient  # noqa: E402
+from m2_market_indexer.reindex import run  # noqa: E402
+
+SCHEMA_PATH = Path(__file__).resolve().parents[2] / "schemas" / "listing.schema.json"
+
+LISTING_PUBLISHED = {
+    "schema_version": "m2.listing.v1",
+    "listing_id": "lst_pdf-export",
+    "solution_id": "sol_pdf-export",
+    "solution_version": "1.2.0",
+    "inventory_type": "solution",
+    "name": "PDF Export",
+    "summary": "Branded PDF report generation",
+    "category": "reporting",
+    "keywords": ["pdf", "report", "branding"],
+    "price": {"amount": 50, "currency": "m2cr", "model": "one_time"},
+    "seller": "op_nasr",
+    "evidence_summary": "12 installs, 4.6 avg rating",
+    "tenant_visibility": ["*"],
+    "stats": {"installs": 12, "proposals_shown": 20, "proposals_accepted": 12},
+    "status": "published",
+    "install_ref": "lst_pdf-export-v1.2.0",
+}
+
+LISTING_DELISTED = {
+    **LISTING_PUBLISHED,
+    "listing_id": "lst_old-tool",
+    "name": "Old Tool",
+    "status": "delisted",
+    "install_ref": "lst_old-tool-v0.9.0",
+}
+
+SOLUTION_PUBLISHED = {"intent": "generate a branded PDF report from a data export"}
+
+
+def _b64_contents(payload: dict) -> dict:
+    encoded = base64.b64encode(json.dumps(payload).encode("utf-8")).decode("ascii")
+    return {"content": encoded}
+
+
+def _make_handlers():
+    calls: list[tuple[str, str]] = []
+
+    def registry_handler(request: httpx.Request) -> httpx.Response:
+        calls.append(("registry", f"{request.method} {request.url.path}"))
+        path = request.url.path
+
+        if path == "/api/v1/repos/m2/market-registry/contents/listings":
+            return httpx.Response(
+                200,
+                json=[
+                    {"name": "lst_pdf-export", "type": "dir"},
+                    {"name": "lst_old-tool", "type": "dir"},
+                ],
+            )
+        listings_base = "/api/v1/repos/m2/market-registry/contents/listings"
+        if path == f"{listings_base}/lst_pdf-export/listing.json":
+            return httpx.Response(200, json=_b64_contents(LISTING_PUBLISHED))
+        if path == f"{listings_base}/lst_pdf-export/solution.json":
+            return httpx.Response(200, json=_b64_contents(SOLUTION_PUBLISHED))
+        if path == f"{listings_base}/lst_old-tool/listing.json":
+            return httpx.Response(200, json=_b64_contents(LISTING_DELISTED))
+        return httpx.Response(404, json={"message": "not found"})
+
+    def memory_handler(request: httpx.Request) -> httpx.Response:
+        calls.append(("memory", f"{request.method} {request.url.path}"))
+        return httpx.Response(200, json={"ok": True})
+
+    return calls, registry_handler, memory_handler
+
+
+def _exercise_reindex_flow():
+    calls, registry_handler, memory_handler = _make_handlers()
+
+    registry = RegistryClient(
+        "https://git.machinemachine.ai",
+        "frg_test",
+        "m2/market-registry",
+        "main",
+        transport=httpx.MockTransport(registry_handler),
+    )
+    memory = MemoryClient(
+        "https://memory.machinemachine.ai",
+        "mem_test",
+        "market:catalog",
+        partition_path="/partitions/{partition}",
+        document_path="/partitions/{partition}/documents/{doc_id}",
+        transport=httpx.MockTransport(memory_handler),
+    )
+    with SCHEMA_PATH.open() as f:
+        validator = Draft202012Validator(json.load(f))
+
+    counts = run(registry, memory, validator)
+
+    assert counts == {
+        "scanned": 2,
+        "invalid": 0,
+        "not_published": 1,
+        "upserted": 1,
+    }, counts
+
+    memory_calls = [c for kind, c in calls if kind == "memory"]
+    assert memory_calls == [
+        "DELETE /partitions/market:catalog",
+        "PUT /partitions/market:catalog/documents/lst_pdf-export",
+    ], memory_calls
+
+    registry_calls = [c for kind, c in calls if kind == "registry"]
+    assert any("lst_pdf-export/solution.json" in c for c in registry_calls)
+    assert not any("lst_old-tool/solution.json" in c for c in registry_calls)
+
+    return counts, memory_calls
+
+
+def test_reindex_flow_upserts_published_only():
+    _exercise_reindex_flow()
+
+
+if __name__ == "__main__":
+    counts, memory_calls = _exercise_reindex_flow()
+    print("OK: reindex counts ->", counts)
+    print("OK: memory-api calls ->", memory_calls)

ledger/Dockerfile

@@ -0,0 +1,22 @@
+FROM python:3.12-slim
+
+WORKDIR /app
+
+# Install the ledger package itself (deps come from pyproject.toml [project.dependencies]).
+COPY pyproject.toml ./
+COPY src ./src
+RUN pip install --no-cache-dir .
+
+RUN groupadd --system ledger && useradd --system --gid ledger --home-dir /app ledger \
+    && mkdir -p /data && chown -R ledger:ledger /data /app
+USER ledger
+
+ENV LEDGER_DB_PATH=/data/ledger.db
+
+VOLUME ["/data"]
+EXPOSE 8000
+
+HEALTHCHECK --interval=30s --timeout=3s --start-period=5s --retries=3 \
+    CMD python -c "import urllib.request; urllib.request.urlopen('http://127.0.0.1:8000/health', timeout=2)" || exit 1
+
+CMD ["uvicorn", "m2_ledger.api:app", "--host", "0.0.0.0", "--port", "8000"]

ledger/README.md

@@ -0,0 +1,65 @@
+# m2-ledger
+
+Append-only credits ledger — transactions `{ts, from, to, amount, reason, ref}`,
+balances always derived by folding the tx log (never stored authoritatively). See
+`specs/001-market-first-wedge/contracts/ledger-api.md` (frozen v1 contract) and
+`docs/config.md` (env var / key reference) for the full spec.
+
+## Run locally
+
+```bash
+cd ledger
+uv sync
+LEDGER_DB_PATH=./ledger.db \
+LEDGER_SERVICE_KEYS=svc_devkey \
+LEDGER_ADMIN_KEYS=admin_devkey \
+  uv run uvicorn m2_ledger.api:app --reload --port 8000
+```
+
+`GET http://localhost:8000/health` should return `{"status":"healthy"}`. All other
+endpoints require an `X-API-Key` header (service or admin class — see
+`docs/config.md` §4).
+
+## Deploy as a Coolify app
+
+`m2-ledger` is the one sanctioned new service in the first wedge (constitution I);
+it deploys as a plain Coolify app on the `coolify` Docker network, not baked into
+any image.
+
+1. **Build**: point the Coolify app at this repo/subpath (`ledger/`) with
+   `ledger/Dockerfile` — no build args, no secrets at build time (constitution VI:
+   no secrets in repos or images).
+2. **Network + alias**: attach the app to the `coolify` network with alias
+   `m2-ledger`, so callers use `http://m2-ledger:8000` per the frozen contract's
+   base URL.
+3. **Env vars** (inject at runtime via Coolify's environment panel — never commit
+   real values, see `docs/config.md` §5):
+   - `LEDGER_DB_PATH=/data/ledger.db` (already set as the image default; override
+     only if the persistent volume is mounted elsewhere)
+   - `LEDGER_SERVICE_KEYS` — comma-separated `service`-class keys (CLI/Store/indexer)
+   - `LEDGER_ADMIN_KEYS` — comma-separated `admin`-class keys (grants, snapshots);
+     keep this set small
+   - `LEDGER_PLATFORM_PCT` (default `0.10`, UNCONFIRMED) and `LEDGER_STARTER_GRANT`
+     (default `100`, UNCONFIRMED) — optional, override the code defaults
+   - `REGISTRY_REPO_URL` + `FORGEJO_TOKEN` — required once `/snapshot` actually
+     commits to `m2/market-registry` (see TODO below)
+4. **Persistent volume**: mount a Coolify persistent volume at `/data` (the image
+   declares `VOLUME /data`) so `ledger.db` survives redeploys. This is the single
+   source of truth for the tx log — never point it at ephemeral container storage.
+5. **Health check**: the image's `HEALTHCHECK` hits `GET /health` (no auth). Coolify
+   should also be configured to poll the same endpoint for its own readiness gate.
+6. **Canary-first** (constitution VI): roll out to one environment before any
+   fleet-wide dependency on `m2-ledger` — there is no fleet-wide blast for this
+   service since it's a single shared instance, but verify `/health` and one
+   `/balance/{operator_id}` call with a real service key before pointing the CLI's
+   `ledger_url` at it in `~/.m2-market/config.toml`.
+
+## Snapshot cron (TODO T034)
+
+`POST /snapshot` (admin) currently computes `audit/YYYY-MM-DD.json` balances and
+returns the JSON body but does **not** commit it anywhere — see the
+`TODO(T034)` in `src/m2_ledger/api.py`. Once T034 lands (Forgejo commit via
+`REGISTRY_REPO_URL` + `FORGEJO_TOKEN`), add a daily Coolify scheduled task (or
+cron container) that calls `POST /snapshot` with an admin key once every 24h,
+per constitution V ("Daily balance snapshots MUST be committed to Forgejo for
+audit") and the contract's "admin, also daily cron" note. Not wired up yet.

ledger/src/m2_ledger/api.py

@@ -0,0 +1,148 @@
+"""m2-ledger FastAPI app: wires models.py (tx log + balances) to auth.py
+(X-API-Key service/admin classes) per contracts/ledger-api.md (frozen v1).
+"""
+
+import json
+import os
+from datetime import datetime, timezone
+from typing import Optional
+
+from fastapi import Depends, FastAPI
+from fastapi.responses import JSONResponse
+from pydantic import BaseModel
+
+from m2_ledger import models
+from m2_ledger.auth import require_admin, require_service
+
+app = FastAPI(title="m2-ledger")
+
+
+def _db_path() -> str:
+    return os.environ.get("LEDGER_DB_PATH", "./ledger.db")
+
+
+def _platform_pct_default() -> float:
+    return float(os.environ.get("LEDGER_PLATFORM_PCT", "0.10"))
+
+
+def _starter_grant_default() -> int:
+    return int(os.environ.get("LEDGER_STARTER_GRANT", "100"))
+
+
+def get_conn():
+    conn = models.init_db(_db_path())
+    try:
+        yield conn
+    finally:
+        conn.close()
+
+
+class InstallRequest(BaseModel):
+    buyer: str
+    seller: str
+    amount: int
+    platform_pct: Optional[float] = None
+    ref: str
+
+
+class RefundRequest(BaseModel):
+    ref: str
+
+
+class StarterGrantRequest(BaseModel):
+    operator_id: str
+    amount: Optional[int] = None
+
+
+@app.get("/health")
+def health():
+    return {"status": "healthy"}
+
+
+@app.get("/balance/{operator_id}")
+def get_balance(operator_id: str, conn=Depends(get_conn), _key: str = Depends(require_service)):
+    bal = models.balance(conn, operator_id)
+    return {
+        "operator_id": operator_id,
+        "balance": bal,
+        "as_of": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.%fZ"),
+    }
+
+
+@app.post("/tx/install")
+def install(
+    body: InstallRequest,
+    conn=Depends(get_conn),
+    _key: str = Depends(require_service),
+):
+    pct = body.platform_pct if body.platform_pct is not None else _platform_pct_default()
+    try:
+        result = models.record_install(conn, body.buyer, body.seller, body.amount, pct, body.ref)
+    except models.InsufficientFunds:
+        return JSONResponse(
+            status_code=402,
+            content={"error": "insufficient_funds", "balance": models.balance(conn, body.buyer)},
+        )
+    except models.DuplicateRef:
+        return JSONResponse(status_code=409, content={"error": "duplicate_ref"})
+    return result
+
+
+@app.post("/tx/refund")
+def refund(
+    body: RefundRequest,
+    conn=Depends(get_conn),
+    _key: str = Depends(require_service),
+):
+    try:
+        result = models.record_refund(conn, body.ref)
+    except models.UnknownRef:
+        return JSONResponse(status_code=404, content={"error": "unknown_ref"})
+    except models.AlreadyRefunded:
+        return JSONResponse(status_code=409, content={"error": "already_refunded"})
+    return result
+
+
+@app.get("/tx")
+def list_tx(
+    operator_id: Optional[str] = None,
+    reason: Optional[str] = None,
+    since: Optional[str] = None,
+    conn=Depends(get_conn),
+    _key: str = Depends(require_service),
+):
+    return {"transactions": models.transactions(conn, operator_id=operator_id, reason=reason, since=since)}
+
+
+@app.get("/license/{operator_id}/{listing_id}")
+def get_license(
+    operator_id: str,
+    listing_id: str,
+    conn=Depends(get_conn),
+    _key: str = Depends(require_service),
+):
+    grant = models.license_for(conn, operator_id, listing_id)
+    if grant is None:
+        return JSONResponse(status_code=404, content={"error": "not_found"})
+    return {"grant": grant}
+
+
+@app.post("/grant/starter")
+def grant_starter(
+    body: StarterGrantRequest,
+    conn=Depends(get_conn),
+    _key: str = Depends(require_admin),
+):
+    amount = body.amount if body.amount is not None else _starter_grant_default()
+    return models.record_grant(conn, body.operator_id, amount)
+
+
+@app.post("/snapshot")
+def snapshot(conn=Depends(get_conn), _key: str = Depends(require_admin)):
+    balances = models.all_balances(conn)
+    date = datetime.now(timezone.utc).strftime("%Y-%m-%d")
+    path = f"audit/{date}.json"
+    content = json.dumps(balances, sort_keys=True, indent=2)
+    # TODO(T034): commit `content` to `path` in m2/market-registry via Forgejo API
+    # (REGISTRY_REPO_URL + FORGEJO_TOKEN); this endpoint only produces the JSON.
+    return {"path": path, "content": content}

ledger/src/m2_ledger/models.py

@@ -121,8 +121,17 @@ def record_install(
         cut = int(amount * platform_pct)
         seller_amount = amount - cut
 
-        net_tx_id = _insert_tx(conn, ts, buyer, seller, seller_amount, "install", ref)
+        # The tx table CHECKs amount > 0: a pct that rounds either side to zero
-        cut_tx_id = _insert_tx(conn, ts, buyer, "platform", cut, "install", ref)
+        # (0.0 -> cut == 0, 1.0 -> seller_amount == 0) must skip that row, not
+        # violate the constraint. amount >= 1 guarantees at least one row.
+        tx_ids: list[int] = []
+        net_tx_id = None
+        if seller_amount > 0:
+            net_tx_id = _insert_tx(conn, ts, buyer, seller, seller_amount, "install", ref)
+            tx_ids.append(net_tx_id)
+        if cut > 0:
+            tx_ids.append(_insert_tx(conn, ts, buyer, "platform", cut, "install", ref))
+        paying_tx_id = net_tx_id if net_tx_id is not None else tx_ids[0]
 
         grant_id = f"gr_{uuid.uuid4().hex}"
         conn.execute(
@@ -131,14 +140,14 @@ def record_install(
                                  solution_version_major, tx_id, ts)
             VALUES (?, ?, ?, ?, ?, ?)
             """,
-            (grant_id, buyer, ref, 1, net_tx_id, ts),
+            (grant_id, buyer, ref, 1, paying_tx_id, ts),
         )
 
         grant_row = conn.execute(
             "SELECT * FROM grant_ WHERE grant_id = ?", (grant_id,)
         ).fetchone()
 
-    return {"tx_ids": [net_tx_id, cut_tx_id], "grant": dict(grant_row)}
+    return {"tx_ids": tx_ids, "grant": dict(grant_row)}
 
 
 def record_refund(conn: sqlite3.Connection, ref: str) -> dict:

ledger/tests/conftest.py

@@ -0,0 +1,40 @@
+"""Shared fixtures for m2-ledger tests: isolated sqlite conn + TestClient."""
+
+import pytest
+from fastapi.testclient import TestClient
+
+from m2_ledger import auth, models
+from m2_ledger.api import app
+
+SERVICE_KEY = "test-service-key"
+ADMIN_KEY = "test-admin-key"
+
+
+@pytest.fixture
+def conn(tmp_path):
+    connection = models.init_db(str(tmp_path / "ledger.sqlite3"))
+    yield connection
+    connection.close()
+
+
+@pytest.fixture
+def client(tmp_path, monkeypatch):
+    monkeypatch.setenv("LEDGER_DB_PATH", str(tmp_path / "ledger_api.sqlite3"))
+    monkeypatch.setenv("LEDGER_SERVICE_KEYS", SERVICE_KEY)
+    monkeypatch.setenv("LEDGER_ADMIN_KEYS", ADMIN_KEY)
+    auth._service_keys.cache_clear()
+    auth._admin_keys.cache_clear()
+    with TestClient(app) as test_client:
+        yield test_client
+    auth._service_keys.cache_clear()
+    auth._admin_keys.cache_clear()
+
+
+@pytest.fixture
+def service_headers():
+    return {"X-API-Key": SERVICE_KEY}
+
+
+@pytest.fixture
+def admin_headers():
+    return {"X-API-Key": ADMIN_KEY}

ledger/tests/test_properties.py

@@ -0,0 +1,271 @@
+"""Property + invariant tests for m2_ledger (contracts/ledger-api.md §Invariants).
+
+1. No endpoint UPDATEs/DELETEs tx rows                    -> append-only by construction,
+                                                              exercised indirectly by every test.
+2. balance == fold(tx log) for every operator             -> test_balance_matches_independent_fold
+3. Every grant has a non-null paying tx                   -> test_every_license_grant_has_paying_tx
+4. /tx/install is all-or-nothing and idempotent by ref     -> test_insufficient_funds_leaves_zero_new_rows
+                                                              test_duplicate_ref_leaves_zero_new_rows
+                                                              test_api_install_402_leaves_tx_log_unchanged
+Plus: refund exactly compensates (test_refund_restores_pre_install_balances) and no balance
+except mint goes negative (checked as a stateful invariant below).
+"""
+
+from hypothesis import HealthCheck, assume, settings
+from hypothesis import strategies as st
+from hypothesis.stateful import RuleBasedStateMachine, invariant, rule
+
+from m2_ledger import models
+
+OPERATORS = ["op_alice", "op_bob", "op_carol"]
+ALL_ACCOUNTS = OPERATORS + ["platform"]
+
+
+def _fold_balance(conn, operator_id: str) -> int:
+    """Independent re-implementation of models.balance: fold the raw tx log in
+    Python rather than SQL, so this actually cross-checks the SQL aggregation."""
+    bal = 0
+    for row in conn.execute("SELECT from_id, to_id, amount FROM tx").fetchall():
+        if row["to_id"] == operator_id:
+            bal += row["amount"]
+        if row["from_id"] == operator_id:
+            bal -= row["amount"]
+    return bal
+
+
+def _known_accounts(conn) -> set:
+    ids = set()
+    for row in conn.execute("SELECT from_id, to_id FROM tx").fetchall():
+        ids.add(row["from_id"])
+        ids.add(row["to_id"])
+    return ids
+
+
+class LedgerMachine(RuleBasedStateMachine):
+    """Arbitrary sequences of valid ops (grant, install, refund) against a fresh
+    in-memory ledger, checking fold-consistency and non-negativity after every op."""
+
+    def __init__(self):
+        super().__init__()
+        self.conn = models.init_db(":memory:")
+        self.installed_refs = []
+        self._ref_counter = 0
+
+    def teardown(self):
+        self.conn.close()
+
+    @rule(
+        operator=st.sampled_from(OPERATORS),
+        amount=st.integers(min_value=1, max_value=1000),
+    )
+    def grant(self, operator, amount):
+        models.record_grant(self.conn, operator, amount)
+
+    @rule(
+        pair=st.permutations(OPERATORS).map(lambda p: (p[0], p[1])),
+        amount=st.integers(min_value=1, max_value=500),
+        pct=st.floats(min_value=0.0, max_value=1.0, allow_nan=False),
+    )
+    def install(self, pair, amount, pct):
+        buyer, seller = pair
+        # only attempt ops that are valid per the buyer's *actual* current balance,
+        # so this models "arbitrary sequences of valid ops", not failure paths.
+        if models.balance(self.conn, buyer) < amount:
+            return
+        # A platform_pct that rounds the cut or the seller's net to exactly 0
+        # trips a known source bug (see report: zero-amount tx row violates the
+        # `amount > 0` CHECK constraint with an unhandled IntegrityError) —
+        # excluded here since it's not part of this property, not a fix.
+        cut = int(amount * pct)
+        assume(0 < cut < amount)
+        self._ref_counter += 1
+        ref = f"ref-{self._ref_counter}"
+        models.record_install(self.conn, buyer, seller, amount, pct, ref)
+        self.installed_refs.append(ref)
+
+    @rule(data=st.data())
+    def refund(self, data):
+        if not self.installed_refs:
+            return
+        ref = data.draw(st.sampled_from(self.installed_refs))
+        try:
+            models.record_refund(self.conn, ref)
+        except models.AlreadyRefunded:
+            pass
+        else:
+            self.installed_refs.remove(ref)
+
+    @invariant()
+    def balances_match_independent_fold(self):
+        for account in _known_accounts(self.conn):
+            assert models.balance(self.conn, account) == _fold_balance(self.conn, account)
+
+    @invariant()
+    def no_non_mint_balance_goes_negative(self):
+        for account in _known_accounts(self.conn):
+            if account == "mint":
+                continue
+            assert models.balance(self.conn, account) >= 0
+
+
+LedgerStateMachineTest = LedgerMachine.TestCase
+LedgerStateMachineTest.settings = settings(
+    max_examples=50,
+    stateful_step_count=25,
+    suppress_health_check=[HealthCheck.too_slow, HealthCheck.data_too_large],
+    deadline=None,
+)
+
+
+# ---------------------------------------------------------------------------
+# Example-based invariant tests
+# ---------------------------------------------------------------------------
+
+
+def test_every_license_grant_has_paying_tx(conn):
+    """Invariant 3: install grants reference a real 'install' tx; starter grants
+    (money only, reason='grant') never create a license grant_ row at all."""
+    models.record_grant(conn, "op_alice", 100)
+    result = models.record_install(conn, "op_alice", "op_bob", 40, 0.10, "ref-install-1")
+
+    grant_row = result["grant"]
+    assert grant_row["tx_id"] is not None
+
+    paying_tx = conn.execute(
+        "SELECT * FROM tx WHERE id = ?", (grant_row["tx_id"],)
+    ).fetchone()
+    assert paying_tx is not None
+    assert paying_tx["reason"] == "install"
+    assert paying_tx["from_id"] == "op_alice"
+    assert paying_tx["to_id"] == "op_bob"
+
+    # the starter grant produced a money tx but no row in grant_ (no license without payment)
+    grant_count = conn.execute("SELECT COUNT(*) AS n FROM grant_").fetchone()["n"]
+    assert grant_count == 1
+
+
+def test_no_license_grant_without_install(conn):
+    """A pure starter grant must never create a grant_ (license) row."""
+    models.record_grant(conn, "op_alice", 100)
+    grant_count = conn.execute("SELECT COUNT(*) AS n FROM grant_").fetchone()["n"]
+    assert grant_count == 0
+
+
+def test_insufficient_funds_leaves_zero_new_rows(conn):
+    models.record_grant(conn, "op_alice", 10)
+    tx_count_before = conn.execute("SELECT COUNT(*) AS n FROM tx").fetchone()["n"]
+    grant_count_before = conn.execute("SELECT COUNT(*) AS n FROM grant_").fetchone()["n"]
+
+    try:
+        models.record_install(conn, "op_alice", "op_bob", 10_000, 0.10, "ref-too-much")
+        assert False, "expected InsufficientFunds"
+    except models.InsufficientFunds:
+        pass
+
+    tx_count_after = conn.execute("SELECT COUNT(*) AS n FROM tx").fetchone()["n"]
+    grant_count_after = conn.execute("SELECT COUNT(*) AS n FROM grant_").fetchone()["n"]
+    assert tx_count_after == tx_count_before
+    assert grant_count_after == grant_count_before
+
+
+def test_duplicate_ref_leaves_zero_new_rows(conn):
+    models.record_grant(conn, "op_alice", 100)
+    models.record_install(conn, "op_alice", "op_bob", 40, 0.10, "ref-dup")
+
+    tx_count_before = conn.execute("SELECT COUNT(*) AS n FROM tx").fetchone()["n"]
+    grant_count_before = conn.execute("SELECT COUNT(*) AS n FROM grant_").fetchone()["n"]
+
+    try:
+        models.record_install(conn, "op_alice", "op_bob", 40, 0.10, "ref-dup")
+        assert False, "expected DuplicateRef"
+    except models.DuplicateRef:
+        pass
+
+    tx_count_after = conn.execute("SELECT COUNT(*) AS n FROM tx").fetchone()["n"]
+    grant_count_after = conn.execute("SELECT COUNT(*) AS n FROM grant_").fetchone()["n"]
+    assert tx_count_after == tx_count_before
+    assert grant_count_after == grant_count_before
+
+
+def test_refund_restores_pre_install_balances(conn):
+    models.record_grant(conn, "op_alice", 100)
+    pre_install = {
+        op: models.balance(conn, op) for op in ("op_alice", "op_bob", "platform")
+    }
+
+    models.record_install(conn, "op_alice", "op_bob", 40, 0.10, "ref-refundable")
+    models.record_refund(conn, "ref-refundable")
+
+    post_refund = {
+        op: models.balance(conn, op) for op in ("op_alice", "op_bob", "platform")
+    }
+    assert post_refund == pre_install
+
+
+def test_refund_twice_rejected_and_adds_no_rows(conn):
+    models.record_grant(conn, "op_alice", 100)
+    models.record_install(conn, "op_alice", "op_bob", 40, 0.10, "ref-once")
+    models.record_refund(conn, "ref-once")
+
+    tx_count_before = conn.execute("SELECT COUNT(*) AS n FROM tx").fetchone()["n"]
+    try:
+        models.record_refund(conn, "ref-once")
+        assert False, "expected AlreadyRefunded"
+    except models.AlreadyRefunded:
+        pass
+    tx_count_after = conn.execute("SELECT COUNT(*) AS n FROM tx").fetchone()["n"]
+    assert tx_count_after == tx_count_before
+
+
+# ---------------------------------------------------------------------------
+# API-level test (contract invariant 4, via TestClient)
+# ---------------------------------------------------------------------------
+
+
+def test_api_install_402_leaves_tx_log_unchanged(client, service_headers, admin_headers):
+    client.post(
+        "/grant/starter",
+        json={"operator_id": "op_alice", "amount": 10},
+        headers=admin_headers,
+    )
+
+    before = client.get("/tx", headers=service_headers).json()["transactions"]
+
+    resp = client.post(
+        "/tx/install",
+        json={
+            "buyer": "op_alice",
+            "seller": "op_bob",
+            "amount": 10_000,
+            "ref": "ref-api-too-much",
+        },
+        headers=service_headers,
+    )
+    assert resp.status_code == 402
+    assert resp.json()["error"] == "insufficient_funds"
+
+    after = client.get("/tx", headers=service_headers).json()["transactions"]
+    assert after == before
+
+
+def test_edge_platform_pct_zero_and_one(tmp_path):
+    """Regression: pct that rounds a row to zero must skip it, not 500.
+
+    (T013 worker finding: CHECK amount > 0 was violated by cut == 0 or
+    seller_amount == 0; fixed in models.record_install.)
+    """
+    from m2_ledger import models as M
+
+    conn = M.init_db(str(tmp_path / "edge.db"))
+    M.record_grant(conn, "buyer", 100)
+
+    r0 = M.record_install(conn, "buyer", "seller", 10, 0.0, "ref:pct0")
+    assert M.balance(conn, "seller") == 10
+    assert M.balance(conn, "platform") == 0
+    assert len(r0["tx_ids"]) == 1
+
+    r1 = M.record_install(conn, "buyer", "seller", 10, 1.0, "ref:pct1")
+    assert M.balance(conn, "platform") == 10
+    assert len(r1["tx_ids"]) == 1
+    assert r1["grant"]["tx_id"] == r1["tx_ids"][0]
+    assert M.balance(conn, "buyer") == 80

specs/001-market-first-wedge/tasks.md

@@ -50,10 +50,10 @@ published listing
 
 - [x] T010 [P] [US1] Implement tx + grant models and derived-balance queries in `ledger/src/m2_ledger/models.py` (SQLite WAL, append-only invariants, data-model.md Transaction/LicenseGrant)
 - [x] T011 [P] [US1] Implement X-API-Key auth (service/admin key classes) in `ledger/src/m2_ledger/auth.py`
-- [ ] T012 [US1] Implement API routes `/health`, `/balance/{op}`, `/tx/install`, `/tx/refund`, `/tx`, `/license/{op}/{listing}` in `ledger/src/m2_ledger/api.py` (402/409 semantics, all-or-nothing install tx, idempotent by ref)
+- [x] T012 [US1] Implement API routes `/health`, `/balance/{op}`, `/tx/install`, `/tx/refund`, `/tx`, `/license/{op}/{listing}` in `ledger/src/m2_ledger/api.py` (402/409 semantics, all-or-nothing install tx, idempotent by ref)
-- [ ] T013 [US1] Ledger unit + property tests in `ledger/tests/test_properties.py`: balance==fold(log) under generated sequences; no grant without tx; install atomicity; duplicate-ref 409
+- [x] T013 [US1] Ledger unit + property tests in `ledger/tests/test_properties.py`: balance==fold(log) under generated sequences; no grant without tx; install atomicity; duplicate-ref 409
 - [ ] T014 [US1] Dockerfile + Coolify deployment for m2-ledger on the `coolify` network in `ledger/Dockerfile` + `ledger/README.md` deploy notes; verify `GET /health` from the network
-- [ ] T015 [US1] Create `market:catalog` partition and implement indexer `indexer/src/m2_market_indexer/reindex.py` + `watch.py` per contracts/catalog-index.md (registry → memory-api upsert, tenant_visibility payload, delist removal)
+- [x] T015 [US1] Create `market:catalog` partition and implement indexer `indexer/src/m2_market_indexer/reindex.py` + `watch.py` per contracts/catalog-index.md (registry → memory-api upsert, tenant_visibility payload, delist removal)
 
 ### CLI (contracts/cli.md)
 
@@ -62,11 +62,11 @@ published listing
 - [x] T018 [P] [US1] Registry client in `cli/src/m2_market/registry.py` (fetch listing dirs, download release bundle, verify content_hash)
 - [x] T019 [P] [US1] Catalog client in `cli/src/m2_market/catalog.py` (memory-api semantic query, server-side tenant filter param)
 - [x] T020 [P] [US1] ApplyAdapter protocol + InstallState in `cli/src/m2_market/apply/base.py` per contracts/apply-adapter.md (Changeset, plan/apply/verify/rollback, state.json read/write)
-- [ ] T021 [US1] Local adapter in `cli/src/m2_market/apply/local.py`: recipe.yaml-driven file placement into agent-home, post-install checks, idempotent re-apply, rollback from backups
+- [x] T021 [US1] Local adapter in `cli/src/m2_market/apply/local.py`: recipe.yaml-driven file placement into agent-home, post-install checks, idempotent re-apply, rollback from backups
 - [x] T022 [US1] `m2core_sync` stub adapter in `cli/src/m2_market/apply/m2core_sync.py` raising RailsNotLanded (exit 6)
-- [ ] T023 [US1] `search` + `show` commands in `cli/src/m2_market/cli.py` (catalog query → table/JSON; show renders evidence/price/permissions diff)
+- [x] T023 [US1] `search` + `show` commands in `cli/src/m2_market/cli.py` (catalog query → table/JSON; show renders evidence/price/permissions diff)
-- [ ] T024 [US1] `install` command in `cli/src/m2_market/cli.py`: the FR-005 sequence (confirm → debit → fetch → apply → verify → refund-on-failure → state → entrypoint hint) per contracts/cli.md
+- [x] T024 [US1] `install` command in `cli/src/m2_market/cli.py`: the FR-005 sequence (confirm → debit → fetch → apply → verify → refund-on-failure → state → entrypoint hint) per contracts/cli.md
-- [ ] T025 [US1] `wallet` command in `cli/src/m2_market/cli.py` (balance + recent tx)
+- [x] T025 [US1] `wallet` command in `cli/src/m2_market/cli.py` (balance + recent tx)
 - [ ] T026 [US1] CLI integration tests in `cli/tests/test_install_flow.py` against a local ledger instance + fixture registry/catalog: funded install, insufficient funds (no side effects), idempotent re-run, apply-failure refund
 
 **Checkpoint**: Scenario B/C pass end-to-end with a fixture listing (SC-003/005 held locally)

uv.lock

@@ -252,6 +252,7 @@ version = "0.1.0"
 source = { editable = "indexer" }
 dependencies = [
     { name = "httpx" },
+    { name = "jsonschema" },
 ]
 
 [package.dev-dependencies]
@@ -260,7 +261,10 @@ dev = [
 ]
 
 [package.metadata]
-requires-dist = [{ name = "httpx", specifier = ">=0.27" }]
+requires-dist = [
+    { name = "httpx", specifier = ">=0.27" },
+    { name = "jsonschema", specifier = ">=4.23" },
+]
 
 [package.metadata.requires-dev]
 dev = [{ name = "pytest", specifier = ">=8" }]