up

reme4/__init__.py

@@ -1,5 +1,7 @@
 """ReMe CLI package."""
 
+__version__ = "0.4.0.0"
+
 from . import config
 from . import constants
 from . import enumeration
@@ -22,5 +24,3 @@ __all__ = [
     "steps",
     "utils",
 ]
-
-__version__ = "0.4.0.0"

reme4/components/client/http_client.py

@@ -54,9 +54,7 @@ class HttpClient(BaseClient):
         response = await self.client.post(f"/{self.action}", json=self.kwargs)
         response.raise_for_status()
         result = response.json()
-        if "answer" in result:
-            return str(result["answer"])
-        return str(result)
+        return json.dumps(result, indent=2, ensure_ascii=False)
 
     async def _close(self) -> None:
         """Close the HTTP client."""

reme4/components/service/http_service.py

@@ -82,7 +82,6 @@ class HttpService(BaseService):
             allow_methods=["*"],
             allow_headers=["*"],
         )
-        self.service.post("/health")(lambda: {"status": "healthy"})
 
     def start_service(self, app: "Application") -> None:
         # uvicorn 0.41 still imports websockets.legacy / WebSocketServerProtocol

reme4/config/default.yaml

@@ -4,7 +4,7 @@ service:
 
 jobs:
   - backend: base
-    name: demo_job
+    name: demo
     description: "demo job description"
     parameters:
       type: object
@@ -22,6 +22,24 @@ jobs:
       - backend: demo_echo_step1
       - backend: demo_echo_step2
 
+  - backend: base
+    name: version
+    description: "return reme4 package version"
+    parameters:
+      type: object
+      properties: {}
+    steps:
+      - backend: version_step
+
+  - backend: base
+    name: health_check
+    description: "return a concise health-check snapshot of reme4 components"
+    parameters:
+      type: object
+      properties: {}
+    steps:
+      - backend: health_check_step
+
 components:
   # 1. tokenizer — no dependencies
   tokenizer:

reme4/reme.py

@@ -7,7 +7,7 @@ from .application import Application
 from .components import R
 from .config import parse_args, resolve_app_config
 from .enumeration import ComponentEnum
-from .utils import load_env
+from .utils import cli_find_reme, load_env, precheck_start
 
 
 class ReMe(Application):
@@ -19,7 +19,7 @@ async def call_server(action: str, **kwargs):
     backend: str = kwargs.pop("backend", "http")
     client_cls = R.get(ComponentEnum.CLIENT, backend)
     async with client_cls(action=action, **kwargs) as client:
-        await client()
+        print(await client())
 
 
 def main():
@@ -28,7 +28,11 @@ def main():
     if action == "start":
         load_env()
         kwargs = resolve_app_config(**kwargs)
+        if not precheck_start(kwargs.get("service")):
+            return
         ReMe(**kwargs).run_app()
+    elif action == "find_reme":
+        cli_find_reme()
     else:
         asyncio.run(call_server(action, **kwargs))
 

reme4/steps/__init__.py

@@ -1,10 +1,9 @@
 """steps"""
 
+from . import common
 from .base_step import BaseStep
-from .demo import DemoEchoStep1, DemoEchoStep2
 
 __all__ = [
+    "common",
     "BaseStep",
-    "DemoEchoStep1",
-    "DemoEchoStep2",
 ]

reme4/steps/common/__init__.py

@@ -0,0 +1,12 @@
+"""Common steps."""
+
+from .demo import DemoEchoStep1, DemoEchoStep2
+from .health_check import HealthCheckStep
+from .version import VersionStep
+
+__all__ = [
+    "DemoEchoStep1",
+    "DemoEchoStep2",
+    "HealthCheckStep",
+    "VersionStep",
+]

reme4/steps/common/demo.py

@@ -1,7 +1,7 @@
 """Demo steps for smoke-testing the application stack."""
 
-from .base_step import BaseStep
-from ..components import R
+from ..base_step import BaseStep
+from ...components import R
 
 
 @R.register("demo_echo_step1")

reme4/steps/common/health_check.py

@@ -0,0 +1,150 @@
+"""Return a concise health check snapshot of ReMe runtime components."""
+
+import sys
+from collections.abc import Mapping
+
+import numpy as np
+
+from ..base_step import BaseStep
+from ... import __version__
+from ...components import R
+from ...enumeration import ComponentEnum
+
+
+def _deep_size(obj, _seen: set | None = None) -> int:
+    """Recursive sizeof; uses ndarray.nbytes for numpy and walks containers / __dict__."""
+    if _seen is None:
+        _seen = set()
+    obj_id = id(obj)
+    if obj_id in _seen:
+        return 0
+    _seen.add(obj_id)
+
+    if isinstance(obj, np.ndarray):
+        return int(obj.nbytes) + sys.getsizeof(obj)
+
+    size = sys.getsizeof(obj)
+    if isinstance(obj, (str, bytes, bytearray, int, float, bool, type(None))):
+        return size
+    if isinstance(obj, Mapping):
+        size += sum(_deep_size(k, _seen) + _deep_size(v, _seen) for k, v in obj.items())
+    elif isinstance(obj, (list, tuple, set, frozenset)):
+        size += sum(_deep_size(item, _seen) for item in obj)
+    elif hasattr(obj, "__dict__"):
+        size += _deep_size(vars(obj), _seen)
+    elif hasattr(obj, "__slots__"):
+        for slot in obj.__slots__:
+            if hasattr(obj, slot):
+                size += _deep_size(getattr(obj, slot), _seen)
+    return size
+
+
+def _mb_str(*objs) -> str:
+    """Return summed deep size of objs formatted as 'X.XX MB'."""
+    seen: set = set()
+    total = sum(_deep_size(o, seen) for o in objs)
+    return f"{total / (1024 * 1024):.2f} MB"
+
+
+def _embedding_status(comp) -> dict:
+    return {
+        "is_started": comp.is_started,
+        "is_healthy": getattr(comp, "is_healthy", None),
+        "model_name": getattr(comp, "model_name", None),
+        "dimensions": getattr(comp, "dimensions", None),
+        "cache_size": len(getattr(comp, "_embedding_cache", {}) or {}),
+        "memory": _mb_str(getattr(comp, "_embedding_cache", {}) or {}),
+    }
+
+
+def _file_graph_status(comp) -> dict:
+    nodes = getattr(comp, "_nodes", {}) or {}
+    inverse = getattr(comp, "_inverse", {}) or {}
+    pending = getattr(comp, "_pending", {}) or {}
+    return {
+        "is_started": comp.is_started,
+        "n_nodes": len(nodes),
+        "n_edges": sum(len(s) for s in inverse.values()),
+        "memory": _mb_str(nodes, inverse, pending),
+    }
+
+
+def _file_store_status(comp) -> dict:
+    chunks = getattr(comp, "file_chunks", {}) or {}
+    return {
+        "is_started": comp.is_started,
+        "n_chunks": len(chunks),
+        "n_chunks_with_embedding": sum(1 for c in chunks.values() if getattr(c, "embedding", None) is not None),
+        "memory": _mb_str(chunks),
+    }
+
+
+def _file_watcher_status(comp) -> dict:
+    bg = getattr(comp, "_background_task", None)
+    return {
+        "is_started": comp.is_started,
+        "background_running": bool(bg and not bg.done()),
+        "watch_paths": [str(p) for p in (getattr(comp, "watch_paths", []) or [])],
+    }
+
+
+def _keyword_index_status(comp) -> dict:
+    return {
+        "is_started": comp.is_started,
+        "n_docs": getattr(comp, "n_docs", None),
+        "vocab_size": len(getattr(comp, "vocab", {}) or {}),
+        "memory": _mb_str(
+            getattr(comp, "vocab", {}) or {},
+            getattr(comp, "inverted_index", {}) or {},
+            getattr(comp, "doc_meta", {}) or {},
+            getattr(comp, "_idf_cache", {}) or {},
+        ),
+    }
+
+
+_HANDLERS = {
+    ComponentEnum.EMBEDDING_MODEL: _embedding_status,
+    ComponentEnum.FILE_GRAPH: _file_graph_status,
+    ComponentEnum.FILE_STORE: _file_store_status,
+    ComponentEnum.FILE_WATCHER: _file_watcher_status,
+    ComponentEnum.KEYWORD_INDEX: _keyword_index_status,
+}
+
+
+def _is_status_healthy(ctype: ComponentEnum, status: dict) -> bool:
+    """Per-component health rule. Unstarted = unhealthy; type-specific extras checked."""
+    if not status.get("is_started"):
+        return False
+    if ctype is ComponentEnum.EMBEDDING_MODEL and status.get("is_healthy") is False:
+        return False
+    if ctype is ComponentEnum.FILE_WATCHER and not status.get("background_running"):
+        return False
+    return True
+
+
+@R.register("health_check_step")
+class HealthCheckStep(BaseStep):
+    """Collect a concise health-check snapshot of the relevant components."""
+
+    async def execute(self):
+        assert self.context is not None
+
+        components: dict = {}
+        healthy = True
+        if self.app_context is not None:
+            for ctype, handler in _HANDLERS.items():
+                comp_map = self.app_context.components.get(ctype, {})
+                bucket = {}
+                for name, comp in comp_map.items():
+                    s = handler(comp)
+                    bucket[name] = s
+                    if not _is_status_healthy(ctype, s):
+                        healthy = False
+                components[ctype.value] = bucket
+
+        health = {"version": __version__, "healthy": healthy, "components": components}
+        self.logger.info(f"[{self.name}] health collected: {health}")
+
+        self.context.response.answer = f"ReMe v{__version__} - {'healthy' if healthy else 'unhealthy'}"
+        self.context.response.metadata["health"] = health
+        return self.context.response

reme4/steps/common/version.py

@@ -0,0 +1,18 @@
+"""Return the package version."""
+
+from ..base_step import BaseStep
+from ... import __version__
+from ...components import R
+
+
+@R.register("version_step")
+class VersionStep(BaseStep):
+    """Emit reme4.__version__ as the response answer."""
+
+    async def execute(self):
+        assert self.context is not None
+        self.logger.info(f"[{self.name}] version={__version__}")
+
+        self.context.response.answer = __version__
+        self.context.response.metadata["version"] = __version__
+        return self.context.response

reme4/utils/__init__.py

@@ -4,6 +4,7 @@ from .common_utils import hash_text, execute_stream_task
 from .env_utils import load_env
 from .logger_utils import get_logger
 from .logo_utils import print_logo
+from .service_utils import find_reme, locate_reme, precheck_start, cli_find_reme
 from .similarity_utils import cosine_similarity, batch_cosine_similarity
 
 __all__ = [
@@ -12,6 +13,10 @@ __all__ = [
     "load_env",
     "get_logger",
     "print_logo",
+    "find_reme",
+    "locate_reme",
+    "precheck_start",
+    "cli_find_reme",
     "cosine_similarity",
     "batch_cosine_similarity",
 ]

reme4/utils/service_utils.py

@@ -0,0 +1,109 @@
+"""Service discovery utilities."""
+
+import asyncio
+import socket
+import subprocess
+import sys
+
+from ..components.client.http_client import HttpClient
+from ..constants import REME_DEFAULT_HOST, REME_DEFAULT_PORT
+
+
+async def find_reme(host: str, port: int) -> str:
+    """Probe host:port. Returns 'reme', 'occupied', or 'free'."""
+    try:
+        async with HttpClient(action="health_check", host=host, port=port, timeout=2.0) as c:
+            await c()
+        return "reme"
+    except Exception:
+        pass
+    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
+        s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+        try:
+            s.bind((host, port))
+            return "free"
+        except OSError:
+            return "occupied"
+
+
+def _run(cmd: list[str]) -> str:
+    """Run cmd; return stdout, or empty on failure."""
+    try:
+        return subprocess.check_output(cmd, stderr=subprocess.DEVNULL, text=True)
+    except (subprocess.CalledProcessError, FileNotFoundError):
+        return ""
+
+
+def _pid_on_port(port: int) -> int | None:
+    """Return PID listening on TCP port, or None."""
+    out = _run(["lsof", "-nP", f"-iTCP:{port}", "-sTCP:LISTEN", "-t"]).strip()
+    return int(out.splitlines()[0]) if out else None
+
+
+def _scan_reme_procs() -> list[tuple[int, str, int]]:
+    """Find 'reme ... start' processes. Returns [(pid, host, port), ...]."""
+    procs = []
+    for line in _run(["pgrep", "-af", "reme.* start"]).splitlines():
+        parts = line.split()
+        if not parts:
+            continue
+        try:
+            pid = int(parts[0])
+        except ValueError:
+            continue
+        host, port = REME_DEFAULT_HOST, REME_DEFAULT_PORT
+        for t in parts[1:]:
+            if t.startswith("service.host="):
+                host = t.split("=", 1)[1]
+            elif t.startswith("service.port="):
+                try:
+                    port = int(t.split("=", 1)[1])
+                except ValueError:
+                    pass
+        procs.append((pid, host, port))
+    return procs
+
+
+async def locate_reme() -> tuple[str, int, int | None] | None:
+    """Locate a running reme. Returns (host, port, pid) or None."""
+    # 1. Try default host:port
+    if await find_reme(REME_DEFAULT_HOST, REME_DEFAULT_PORT) == "reme":
+        return REME_DEFAULT_HOST, REME_DEFAULT_PORT, _pid_on_port(REME_DEFAULT_PORT)
+    # 2. Scan reme processes and probe each
+    for pid, host, port in _scan_reme_procs():
+        if await find_reme(host, port) == "reme":
+            return host, port, pid
+    return None
+
+
+def precheck_start(svc_config: dict | None) -> bool:
+    """Pre-flight check before `start`. Returns True if caller should proceed.
+
+    Prints a message and returns False if reme is already running.
+    Exits with code 1 if the port is occupied by a non-reme process.
+    """
+    host = (svc_config or {}).get("host") or REME_DEFAULT_HOST
+    port = (svc_config or {}).get("port") or REME_DEFAULT_PORT
+    status = asyncio.run(find_reme(host, port))
+    if status == "reme":
+        print(f"reme already running at {host}:{port}")
+        return False
+    if status == "occupied":
+        print(
+            f"port {port} is occupied by another process. "
+            f"Start with a different port: reme4 start service.port=<other_port>",
+            file=sys.stderr,
+        )
+        sys.exit(1)
+    return True
+
+
+def cli_find_reme() -> None:
+    """Handle the `find_reme` CLI action: print HOST/PORT/PID or a hint."""
+    found = asyncio.run(locate_reme())
+    if found:
+        host, port, pid = found
+        print(f"HOST={host} PORT={port} PID={pid if pid is not None else 'unknown'}")
+    else:
+        print("reme not started. Try: reme start", file=sys.stderr)
+        sys.exit(1)