merge: indexer (US1 wave 2)

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)

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" }]