T015: catalog indexer (reindex + watch)

Registry (Forgejo) + memory-api clients under m2_market_indexer.client, with
memory-api paths/creds as env-overridable constants (config.py). reindex.py
drops market:catalog and rebuilds from published listings on main, skipping
schema-invalid ones. watch.py polls the registry tip commit against a cached
state.json and upserts changed/new published listings, deletes delisted ones.
This commit is contained in:
m2 (AI Agent) 2026-07-02 02:51:30 +02:00
parent 8a8db9f2cc
commit 1d10589cef
8 changed files with 668 additions and 1 deletions

View file

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

View file

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

View file

@ -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)

View file

@ -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"

View file

@ -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()

View file

@ -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()

View file

@ -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)

View file

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