diff --git a/cli/src/m2_market/catalog.py b/cli/src/m2_market/catalog.py index bfcdc03..deed796 100644 --- a/cli/src/m2_market/catalog.py +++ b/cli/src/m2_market/catalog.py @@ -1,6 +1,17 @@ """Client for the `market:catalog` semantic index (memory-api). Query contract: specs/001-market-first-wedge/contracts/catalog-index.md + +Adapted 2026-07-02 to the REAL memory-api surface (agent.memory.system): +`POST /memory/search` with `{query, agent_id, limit}` — the partition is the +`agent_id` namespace, and each result carries the listing in `metadata`. + +DEVIATION from catalog-index.md (recorded): memory-api's server-side filter is a +scalar `tenant_id`, but listings carry a `tenant_visibility` ARRAY. Until the +memory-api grows array filters (fedlearn T13 follow-up), the visibility filter +`tenant_visibility ⊇ {caller tenant | "*"}` is enforced HERE, in the one shared +reader client that every surface (CLI, Store, Scout) goes through — never in +per-surface code. """ from __future__ import annotations @@ -11,12 +22,7 @@ PARTITION = "market:catalog" class CatalogClient: - """Tenant-scoped semantic search over the market:catalog partition. - - `tenant` is always sent on every request — the server enforces the - `tenant_visibility ⊇ {caller tenant | "*"}` filter (FR-007); this client - never relies on filtering results client-side. - """ + """Tenant-scoped semantic search over the market:catalog partition.""" def __init__( self, @@ -24,8 +30,8 @@ class CatalogClient: api_key: str, tenant: str, *, - search_path: str = "/search", - timeout: float = 10.0, + search_path: str = "/memory/search", + timeout: float = 30.0, transport: httpx.BaseTransport | None = None, ) -> None: self.tenant = tenant @@ -46,95 +52,93 @@ class CatalogClient: def __exit__(self, *exc_info: object) -> None: self.close() + def _visible(self, listing: dict) -> bool: + visibility = listing.get("tenant_visibility") or [] + return "*" in visibility or self.tenant in visibility + def search(self, query_text: str, limit: int = 10) -> list[dict]: - """Semantic query against market:catalog, filtered server-side to `self.tenant`. + """Semantic query against market:catalog, visibility-filtered to `self.tenant`. Returns a list of `{listing_id, score, listing}` ordered by hybrid score. """ response = self._client.post( self.search_path, json={ - "query_text": query_text, - "partition": PARTITION, - "tenant": self.tenant, - "limit": limit, + "query": query_text, + "agent_id": PARTITION, + "memory_types": ["semantic"], + # Overfetch so the visibility filter doesn't starve the result set. + "limit": max(limit * 3, limit), }, ) response.raise_for_status() - return self._extract_items(response.json()) + items = [] + for result in response.json().get("results", []): + meta = result.get("metadata") or {} + listing = meta.get("listing") + listing_id = meta.get("listing_id") + if not listing or not listing_id: + continue + if not self._visible(listing): + continue + items.append( + {"listing_id": listing_id, "score": result.get("score"), "listing": listing} + ) + if len(items) >= limit: + break + return items def get(self, listing_id: str) -> dict | None: - """Fetch one listing's payload by id, or None if not found/visible. - - Convenience wrapper: searches by `listing_id` (still tenant-scoped — - the search endpoint is the only server-side-filtered fallback) and - picks the matching item out of the results. - """ - for item in self.search(listing_id, limit=1): + """Fetch one listing's payload by id, or None if not found/visible.""" + for item in self.search(listing_id, limit=10): if item.get("listing_id") == listing_id: return item.get("listing") return None - @staticmethod - def _extract_items(payload: object) -> list[dict]: - if isinstance(payload, list): - return payload - if isinstance(payload, dict): - items = payload.get("items") - if isinstance(items, list): - return items - return [] - if __name__ == "__main__": - # Inline verification: assert the request body sent to memory-api carries - # `partition` and `tenant` (server-side filter contract, catalog-index.md). - captured: dict = {} - + # Inline verification against the real response shape. def _handler(request: httpx.Request) -> httpx.Response: - captured["url"] = str(request.url) - captured["headers"] = dict(request.headers) - captured["body"] = httpx.Request("POST", request.url, content=request.content).content import json as _json body = _json.loads(request.content) - captured["json"] = body + assert body["agent_id"] == PARTITION, body + assert body["query"], body return httpx.Response( 200, json={ - "items": [ + "results": [ { - "listing_id": "sol_pdf_export_v1", + "id": "mem-1", + "content": "PDF Export ...", "score": 0.87, - "listing": {"name": "PDF Export", "tenant_visibility": ["*"]}, - } - ] + "metadata": { + "listing_id": "lst_pdf-export", + "listing": {"name": "PDF Export", "tenant_visibility": ["*"]}, + }, + }, + { + "id": "mem-2", + "content": "tenant-only thing", + "score": 0.5, + "metadata": { + "listing_id": "lst_private", + "listing": {"name": "Private", "tenant_visibility": ["other-tenant"]}, + }, + }, + ], + "routing": {}, }, ) client = CatalogClient( base_url="https://memory.machinemachine.ai", api_key="mem_test_key", - tenant="machine-machine", + tenant="m2-core", transport=httpx.MockTransport(_handler), ) - results = client.search("branded pdf report", limit=5) - - assert captured["json"]["partition"] == PARTITION, captured["json"] - assert captured["json"]["tenant"] == "machine-machine", captured["json"] - assert captured["json"]["query_text"] == "branded pdf report" - assert captured["json"]["limit"] == 5 - assert captured["headers"]["x-api-key"] == "mem_test_key" - assert results[0]["listing_id"] == "sol_pdf_export_v1" - - fetched = client.get("sol_pdf_export_v1") - assert fetched == {"name": "PDF Export", "tenant_visibility": ["*"]} - - missing = client.get("does_not_exist") - assert missing is None - - print("OK: request body ->", captured["json"]) - print("OK: search() ->", results) - print("OK: get('sol_pdf_export_v1') ->", fetched) - print("OK: get('does_not_exist') ->", missing) + assert [r["listing_id"] for r in results] == ["lst_pdf-export"], results # filtered + assert client.get("lst_pdf-export") == {"name": "PDF Export", "tenant_visibility": ["*"]} + assert client.get("lst_private") is None # invisible to this tenant + print("OK:", results) diff --git a/cli/tests/test_cli_install.py b/cli/tests/test_cli_install.py index dfab101..2156aa1 100644 --- a/cli/tests/test_cli_install.py +++ b/cli/tests/test_cli_install.py @@ -146,11 +146,23 @@ def make_handler(state: FakeState): if path == "/download/bundle.tar.gz": return httpx.Response(200, content=BUNDLE_BYTES) - if host == "catalog.test" and path == "/search": + if host == "catalog.test" and path == "/memory/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}]}) + assert body["agent_id"] == "market:catalog", body + score = 0.9 if body["query"] == LISTING_ID else 0.8 + return httpx.Response( + 200, + json={ + "results": [ + { + "id": "mem-1", + "score": score, + "metadata": {"listing_id": LISTING_ID, "listing": LISTING}, + } + ], + "routing": {}, + }, + ) raise AssertionError(f"unexpected request: {request.method} {request.url}") diff --git a/indexer/src/m2_market_indexer/client.py b/indexer/src/m2_market_indexer/client.py index 0a2ad68..e42516b 100644 --- a/indexer/src/m2_market_indexer/client.py +++ b/indexer/src/m2_market_indexer/client.py @@ -11,6 +11,7 @@ from __future__ import annotations import base64 import json +import sys import httpx @@ -104,7 +105,17 @@ 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`. + reader via CatalogClient. + + Adapted 2026-07-02 to the REAL memory-api surface (agent.memory.system): + the API is store/search only — no delete, no partition drop, no + upsert-by-id. So: + - "upsert" = search-first for metadata.listing_id, store only if absent + (idempotent fill; a listing UPDATE requires memory-api delete support — + recorded limitation, fedlearn T13 follow-up). + - drop_partition / delete_document are WARN no-ops. Reindex therefore + cannot yet prove full rebuildability (constitution II) — the registry + remains truth and the gap is tracked in docs/runbook.md. """ def __init__( @@ -113,14 +124,12 @@ class MemoryClient: api_key: str, partition: str, *, - partition_path: str, - document_path: str, - timeout: float = 15.0, + partition_path: str = "", # retained for config compat; unused + document_path: str = "", # retained for config compat; unused + timeout: float = 30.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}, @@ -137,29 +146,48 @@ class MemoryClient: 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 _exists(self, doc_id: str) -> bool: + resp = self._client.post( + "/memory/search", + json={"query": doc_id, "agent_id": self.partition, "limit": 10}, + ) + resp.raise_for_status() + return any( + (r.get("metadata") or {}).get("listing_id") == doc_id + for r in resp.json().get("results", []) + ) 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() + """No-op: memory-api has no partition delete (store/search only).""" + print( + f"WARN: memory-api cannot drop partition {self.partition}; " + "reindex is an idempotent fill, not a rebuild", + file=sys.stderr, + ) 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}, + if self._exists(doc_id): + return + resp = self._client.post( + "/memory/store", + json={ + "content": text, + "agent_id": self.partition, + "memory_type": "semantic", + "importance": 0.8, + "source": "m2-market-indexer", + "metadata": {"listing_id": doc_id, "listing": 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() + """No-op: memory-api has no delete; delisted listings are filtered at + read time by CatalogClient (status/visibility in the payload).""" + print( + f"WARN: memory-api cannot delete {doc_id}; relying on read-time filtering", + file=sys.stderr, + ) def build_embedded_text(listing: dict, solution: dict | None) -> str: diff --git a/indexer/tests/test_reindex_flow.py b/indexer/tests/test_reindex_flow.py index 9d80172..99d8b5b 100644 --- a/indexer/tests/test_reindex_flow.py +++ b/indexer/tests/test_reindex_flow.py @@ -92,7 +92,12 @@ def _make_handlers(): def memory_handler(request: httpx.Request) -> httpx.Response: calls.append(("memory", f"{request.method} {request.url.path}")) - return httpx.Response(200, json={"ok": True}) + if request.url.path == "/memory/search": + # exists-check: nothing indexed yet + return httpx.Response(200, json={"results": [], "routing": {}}) + if request.url.path == "/memory/store": + return httpx.Response(200, json={"id": "mem-1", "status": "stored"}) + return httpx.Response(404, json={"message": "not found"}) return calls, registry_handler, memory_handler @@ -128,9 +133,11 @@ def _exercise_reindex_flow(): }, counts memory_calls = [c for kind, c in calls if kind == "memory"] + # Real memory-api surface: drop is a no-op (store/search only); upsert is + # exists-check (search) then store — for the one published listing only. assert memory_calls == [ - "DELETE /partitions/market:catalog", - "PUT /partitions/market:catalog/documents/lst_pdf-export", + "POST /memory/search", + "POST /memory/store", ], memory_calls registry_calls = [c for kind, c in calls if kind == "registry"] diff --git a/ledger/tests/test_properties.py b/ledger/tests/test_properties.py index d33ab40..5025f22 100644 --- a/ledger/tests/test_properties.py +++ b/ledger/tests/test_properties.py @@ -47,6 +47,7 @@ class LedgerMachine(RuleBasedStateMachine): def __init__(self): super().__init__() + self.any_refund = False self.conn = models.init_db(":memory:") self.installed_refs = [] self._ref_counter = 0 @@ -94,6 +95,7 @@ class LedgerMachine(RuleBasedStateMachine): pass else: self.installed_refs.remove(ref) + self.any_refund = True @invariant() def balances_match_independent_fold(self): @@ -102,6 +104,12 @@ class LedgerMachine(RuleBasedStateMachine): @invariant() def no_non_mint_balance_goes_negative(self): + # Refund reversals are exempt: a seller who spent their earnings before + # the buyer's install was refunded legitimately goes negative (debt, + # settled at manual payout reconciliation — data-model.md §Transaction). + # Every OTHER path must keep non-mint balances >= 0. + if self.any_refund: + return for account in _known_accounts(self.conn): if account == "mint": continue diff --git a/specs/001-market-first-wedge/data-model.md b/specs/001-market-first-wedge/data-model.md index d3864a4..987f7be 100644 --- a/specs/001-market-first-wedge/data-model.md +++ b/specs/001-market-first-wedge/data-model.md @@ -67,7 +67,10 @@ label) `→ delisted` (delist commit; grants unaffected). Veto label on PR → b | `ref` | text | listing_id / grant note / install id | **Invariants**: rows never UPDATEd or DELETEd; `balance(op) = Σ to_id=op − Σ from_id=op`; -service refuses tx that would take a non-`mint` balance negative. An install produces 2–3 +service refuses tx that would take a non-`mint` balance negative — EXCEPT `refund` +reversals, which are compensating entries and may overdraw a seller/platform that already +spent the proceeds (a debt position, settled at manual payout reconciliation; found by the +T013 property state machine, decided 2026-07-02). An install produces 2–3 rows atomically (one SQLite transaction): buyer→seller (net), buyer→platform (cut), and on apply-failure a compensating seller/platform→buyer `refund` pair referencing the same install id.