cli/src/m2_market/apply/local.py

@@ -1,240 +0,0 @@
-"""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 M2CoreSyncApplyAdapter:
+class M2CoreSyncAdapter:
     name = "m2core-sync"
 
     def plan(self, bundle: Path, state: InstallState) -> Changeset:

cli/src/m2_market/cli.py

@@ -2,23 +2,12 @@
 
 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.apply import RailsNotLanded, get_adapter
+from m2_market.config import ConfigError, load_config
-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
@@ -57,29 +46,7 @@ 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."""
-    config: Config = ctx.obj["config"]
+    _not_implemented(ctx, "search")
-    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()
@@ -87,180 +54,16 @@ def _fetch_solution(registry: RegistryClient, listing_id: str) -> dict:
 @click.pass_context
 def show(ctx: click.Context, listing_id: str) -> None:
     """Full listing detail: evidence, price, permissions diff, seller, version."""
-    config: Config = ctx.obj["config"]
+    _not_implemented(ctx, "show")
-    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, adapter_name: str | None) -> None:
+def install(ctx: click.Context, listing_id: str, yes: bool) -> None:
-    """Resolve, pay, fetch, apply, and record a listing install (contracts/cli.md FR-005 sequence)."""
+    """Resolve, pay, fetch, apply, and record a listing install."""
-    config: Config = ctx.obj["config"]
+    _not_implemented(ctx, "install")
-    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()
@@ -268,36 +71,7 @@ def install(ctx: click.Context, listing_id: str, yes: bool, adapter_name: str |
 @click.pass_context
 def wallet(ctx: click.Context, operator: str | None) -> None:
     """Balance + recent transactions from the ledger."""
-    config: Config = ctx.obj["config"]
+    _not_implemented(ctx, "wallet")
-    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

@@ -1,64 +0,0 @@
-"""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

@@ -1,308 +0,0 @@
-"""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,12 +5,10 @@ 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

@@ -1,6 +0,0 @@
-"""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

@@ -1,175 +0,0 @@
-"""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

@@ -1,82 +0,0 @@
-"""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

@@ -1,93 +0,0 @@
-"""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

@@ -1,160 +0,0 @@
-"""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

@@ -1,145 +0,0 @@
-"""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

@@ -1,22 +0,0 @@
-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

@@ -1,65 +0,0 @@
-# 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

@@ -1,148 +0,0 @@
-"""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,17 +121,8 @@ def record_install(
         cut = int(amount * platform_pct)
         seller_amount = amount - cut
 
-        # The tx table CHECKs amount > 0: a pct that rounds either side to zero
+        net_tx_id = _insert_tx(conn, ts, buyer, seller, seller_amount, "install", ref)
-        # (0.0 -> cut == 0, 1.0 -> seller_amount == 0) must skip that row, not
+        cut_tx_id = _insert_tx(conn, ts, buyer, "platform", cut, "install", ref)
-        # 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(
@@ -140,14 +131,14 @@ def record_install(
                                  solution_version_major, tx_id, ts)
             VALUES (?, ?, ?, ?, ?, ?)
             """,
-            (grant_id, buyer, ref, 1, paying_tx_id, ts),
+            (grant_id, buyer, ref, 1, net_tx_id, ts),
         )
 
         grant_row = conn.execute(
             "SELECT * FROM grant_ WHERE grant_id = ?", (grant_id,)
         ).fetchone()
 
-    return {"tx_ids": tx_ids, "grant": dict(grant_row)}
+    return {"tx_ids": [net_tx_id, cut_tx_id], "grant": dict(grant_row)}
 
 
 def record_refund(conn: sqlite3.Connection, ref: str) -> dict:

ledger/tests/conftest.py

@@ -1,40 +0,0 @@
-"""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

@@ -1,271 +0,0 @@
-"""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`
-- [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)
+- [ ] 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] 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
+- [ ] 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
-- [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)
+- [ ] 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)
-- [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
+- [ ] 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)
-- [x] T023 [US1] `search` + `show` commands in `cli/src/m2_market/cli.py` (catalog query → table/JSON; show renders evidence/price/permissions diff)
+- [ ] T023 [US1] `search` + `show` commands in `cli/src/m2_market/cli.py` (catalog query → table/JSON; show renders evidence/price/permissions diff)
-- [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
+- [ ] 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] T025 [US1] `wallet` command in `cli/src/m2_market/cli.py` (balance + recent tx)
+- [ ] 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,7 +252,6 @@ version = "0.1.0"
 source = { editable = "indexer" }
 dependencies = [
     { name = "httpx" },
-    { name = "jsonschema" },
 ]
 
 [package.dev-dependencies]
@@ -261,10 +260,7 @@ dev = [
 ]
 
 [package.metadata]
-requires-dist = [
+requires-dist = [{ name = "httpx", specifier = ">=0.27" }]
-    { name = "httpx", specifier = ">=0.27" },
-    { name = "jsonschema", specifier = ">=4.23" },
-]
 
 [package.metadata.requires-dev]
 dev = [{ name = "pytest", specifier = ">=8" }]