Nav System (Ros and Non-ros)

dimos/agents_deprecated/agent.py

@@ -897,5 +897,3 @@ def stream_query(self, query_text: str) -> Observable:  # type: ignore[type-arg]
         return create(
             lambda observer, _: self._observable_query(observer, incoming_query=query_text)  # type: ignore[arg-type]
         )
-
-

dimos/agents_deprecated/memory/spatial_vector_db.py

@@ -227,8 +227,8 @@ def _process_query_results(self, results) -> list[dict]:  # type: ignore[no-unty
                 )
 
             # Get the image from visual memory
-            #image = self.visual_memory.get(lookup_id)
-            #result["image"] = image
+            # image = self.visual_memory.get(lookup_id)
+            # result["image"] = image
 
             processed_results.append(result)
 

dimos/core/native_module.py

@@ -56,6 +56,7 @@ class MyCppModule(NativeModule):
 
 from dimos.core.core import rpc
 from dimos.core.module import Module, ModuleConfig
+from dimos.utils.change_detect import PathEntry, did_change
 from dimos.utils.logging_config import setup_logger
 
 if sys.version_info < (3, 13):
@@ -81,9 +82,10 @@ class NativeModuleConfig(ModuleConfig):
     extra_env: dict[str, str] = Field(default_factory=dict)
     shutdown_timeout: float = 10.0
     log_format: LogFormat = LogFormat.TEXT
+    rebuild_on_change: list[PathEntry] | None = None
 
     # Override in subclasses to exclude fields from CLI arg generation
-    cli_exclude: frozenset[str] = frozenset()
+    cli_exclude: frozenset[str] = frozenset({"rebuild_on_change"})
 
     def to_cli_args(self) -> list[str]:
         """Auto-convert subclass config fields to CLI args.
@@ -132,17 +134,31 @@ class NativeModule(Module[_NativeConfig]):
     _process: subprocess.Popen[bytes] | None = None
     _watchdog: threading.Thread | None = None
     _stopping: bool = False
-    _last_stderr_lines: collections.deque[str]
+    _stderr_tail: list[str]
+    _stdout_tail: list[str]
+    _tail_lock: threading.Lock
+
+    @property
+    def _mod_label(self) -> str:
+        """Short human-readable label: ClassName(executable_basename)."""
+        exe = Path(self.config.executable).name if self.config.executable else "?"
+        return f"{type(self).__name__}({exe})"
 
     def __init__(self, **kwargs: Any) -> None:
         super().__init__(**kwargs)
-        self._last_stderr_lines = collections.deque(maxlen=50)
+        self._stderr_tail: collections.deque[str] = collections.deque(maxlen=50)
+        self._stdout_tail: collections.deque[str] = collections.deque(maxlen=50)
+        self._tail_lock = threading.Lock()
         self._resolve_paths()
 
     @rpc
     def start(self) -> None:
         if self._process is not None and self._process.poll() is None:
-            logger.warning("Native process already running", pid=self._process.pid)
+            logger.warning(
+                "Native process already running",
+                module=self._mod_label,
+                pid=self._process.pid,
+            )
             return
 
         self._maybe_build()
@@ -158,10 +174,14 @@ def start(self) -> None:
         env = {**os.environ, **self.config.extra_env}
         cwd = self.config.cwd or str(Path(self.config.executable).resolve().parent)
 
-        module_name = type(self).__name__
+        # Reset tail buffers for this run.
+        with self._tail_lock:
+            self._stderr_tail.clear()
+            self._stdout_tail.clear()
+
         logger.info(
-            f"Starting native process: {module_name}",
-            module=module_name,
+            "Starting native process",
+            module=self._mod_label,
             cmd=" ".join(cmd),
             cwd=cwd,
         )
@@ -173,26 +193,36 @@ def start(self) -> None:
             stderr=subprocess.PIPE,
         )
         logger.info(
-            f"Native process started: {module_name}",
-            module=module_name,
+            "Native process started",
+            module=self._mod_label,
             pid=self._process.pid,
         )
 
         self._stopping = False
-        self._watchdog = threading.Thread(target=self._watch_process, daemon=True)
+        self._watchdog = threading.Thread(
+            target=self._watch_process,
+            daemon=True,
+            name=f"native-watchdog-{self._mod_label}",
+        )
         self._watchdog.start()
 
     @rpc
     def stop(self) -> None:
         self._stopping = True
         if self._process is not None and self._process.poll() is None:
-            logger.info("Stopping native process", pid=self._process.pid)
+            logger.info(
+                "Stopping native process",
+                module=self._mod_label,
+                pid=self._process.pid,
+            )
             self._process.send_signal(signal.SIGTERM)
             try:
                 self._process.wait(timeout=self.config.shutdown_timeout)
             except subprocess.TimeoutExpired:
                 logger.warning(
-                    "Native process did not exit, sending SIGKILL", pid=self._process.pid
+                    "Native process did not exit, sending SIGKILL",
+                    module=self._mod_label,
+                    pid=self._process.pid,
                 )
                 self._process.kill()
                 self._process.wait(timeout=5)
@@ -207,57 +237,110 @@ def _watch_process(self) -> None:
         if self._process is None:
             return
 
-        stdout_t = self._start_reader(self._process.stdout, "info")
-        stderr_t = self._start_reader(self._process.stderr, "warning")
+        stdout_t = self._start_reader(self._process.stdout, "info", self._stdout_tail)
+        stderr_t = self._start_reader(self._process.stderr, "warning", self._stderr_tail)
         rc = self._process.wait()
         stdout_t.join(timeout=2)
         stderr_t.join(timeout=2)
 
         if self._stopping:
+            logger.info(
+                "Native process exited (expected)",
+                module=self._mod_label,
+                pid=self._process.pid,
+                returncode=rc,
+            )
             return
 
-        module_name = type(self).__name__
-        exe_name = Path(self.config.executable).name if self.config.executable else "unknown"
-
-        # Use buffered stderr lines from the reader thread for the crash report.
-        last_stderr = "\n".join(self._last_stderr_lines)
+        # Grab the tail for diagnostics.
+        with self._tail_lock:
+            stderr_snapshot = list(self._stderr_tail)
+            stdout_snapshot = list(self._stdout_tail)
 
         logger.error(
-            f"Native process crashed: {module_name} ({exe_name})",
-            module=module_name,
-            executable=exe_name,
+            "Native process died unexpectedly",
+            module=self._mod_label,
             pid=self._process.pid,
             returncode=rc,
-            last_stderr=last_stderr[:500] if last_stderr else None,
         )
+
+        # Log the last stderr/stdout lines so the cause is visible.
+        if stderr_snapshot:
+            logger.error(
+                f"Last {len(stderr_snapshot)} stderr lines from {self._mod_label}:",
+                module=self._mod_label,
+                pid=self._process.pid,
+            )
+            for line in stderr_snapshot:
+                logger.error(f"  stderr| {line}", module=self._mod_label)
+
+        if stdout_snapshot and not stderr_snapshot:
+            # Only dump stdout if stderr was empty (avoid double-noise).
+            logger.error(
+                f"Last {len(stdout_snapshot)} stdout lines from {self._mod_label}:",
+                module=self._mod_label,
+                pid=self._process.pid,
+            )
+            for line in stdout_snapshot:
+                logger.error(f"  stdout| {line}", module=self._mod_label)
+
+        if not stderr_snapshot and not stdout_snapshot:
+            logger.error(
+                "No output captured from native process — "
+                "binary may have crashed before producing any output",
+                module=self._mod_label,
+                pid=self._process.pid,
+            )
+
         self.stop()
 
-    def _start_reader(self, stream: IO[bytes] | None, level: str) -> threading.Thread:
+    def _start_reader(
+        self,
+        stream: IO[bytes] | None,
+        level: str,
+        tail_buf: list[str],
+    ) -> threading.Thread:
         """Spawn a daemon thread that pipes a subprocess stream through the logger."""
-        t = threading.Thread(target=self._read_log_stream, args=(stream, level), daemon=True)
+        t = threading.Thread(
+            target=self._read_log_stream,
+            args=(stream, level, tail_buf),
+            daemon=True,
+            name=f"native-reader-{level}-{self._mod_label}",
+        )
         t.start()
         return t
 
-    def _read_log_stream(self, stream: IO[bytes] | None, level: str) -> None:
+    def _read_log_stream(
+        self,
+        stream: IO[bytes] | None,
+        level: str,
+        tail_buf: list[str],
+    ) -> None:
         if stream is None:
             return
         log_fn = getattr(logger, level)
-        is_stderr = level == "warning"
         for raw in stream:
             line = raw.decode("utf-8", errors="replace").rstrip()
             if not line:
                 continue
-            if is_stderr:
-                self._last_stderr_lines.append(line)
+
+            # Keep a rolling tail buffer for crash diagnostics.
+            with self._tail_lock:
+                tail_buf.append(line)
+
             if self.config.log_format == LogFormat.JSON:
                 try:
                     data = json.loads(line)
                     event = data.pop("event", line)
-                    log_fn(event, **data)
+                    log_fn(event, module=self._mod_label, **data)
                     continue
                 except (json.JSONDecodeError, TypeError):
-                    logger.warning("malformed JSON from native module", raw=line)
-            log_fn(line, pid=self._process.pid if self._process else None)
+                    logger.warning(
+                        "malformed JSON from native module",
+                        module=self._mod_label,
+                        raw=line,
+                    )
+            log_fn(line, module=self._mod_label, pid=self._process.pid if self._process else None)
         stream.close()
 
     def _resolve_paths(self) -> None:
@@ -269,18 +352,39 @@ def _resolve_paths(self) -> None:
         if not Path(self.config.executable).is_absolute() and self.config.cwd is not None:
             self.config.executable = str(Path(self.config.cwd) / self.config.executable)
 
+    def _build_cache_name(self) -> str:
+        """Return a stable, unique cache name for this module's build state."""
+        source_file = Path(inspect.getfile(type(self))).resolve()
+        return f"native_{source_file}"
+
     def _maybe_build(self) -> None:
-        """Run ``build_command`` if the executable does not exist."""
+        """Run ``build_command`` if the executable does not exist or sources changed."""
         exe = Path(self.config.executable)
-        if exe.exists():
+
+        # Check if rebuild needed due to source changes
+        needs_rebuild = False
+        if self.config.rebuild_on_change and exe.exists():
+            if did_change(
+                self._build_cache_name(), self.config.rebuild_on_change, cwd=self.config.cwd
+            ):
+                logger.info("Source files changed, triggering rebuild", executable=str(exe))
+                needs_rebuild = True
+
+        if exe.exists() and not needs_rebuild:
             return
+
         if self.config.build_command is None:
             raise FileNotFoundError(
-                f"Executable not found: {exe}. "
+                f"[{self._mod_label}] Executable not found: {exe}. "
                 "Set build_command in config to auto-build, or build it manually."
             )
+
+        # Don't unlink the exe before rebuilding — the build command is
+        # responsible for replacing it.  For nix builds the exe lives inside
+        # a read-only store; `nix build -o` atomically swaps the output
+        # symlink without touching store contents.
         logger.info(
-            "Executable not found, running build",
+            "Rebuilding" if needs_rebuild else "Executable not found, building",
             executable=str(exe),
             build_command=self.config.build_command,
         )
@@ -293,25 +397,36 @@ def _maybe_build(self) -> None:
             stderr=subprocess.PIPE,
         )
         stdout, stderr = proc.communicate()
-        for line in stdout.decode("utf-8", errors="replace").splitlines():
+
+        stdout_lines = stdout.decode("utf-8", errors="replace").splitlines()
+        stderr_lines = stderr.decode("utf-8", errors="replace").splitlines()
+
+        for line in stdout_lines:
             if line.strip():
-                logger.info(line)
-        for line in stderr.decode("utf-8", errors="replace").splitlines():
+                logger.info(line, module=self._mod_label)
+        for line in stderr_lines:
             if line.strip():
-                logger.warning(line)
+                logger.warning(line, module=self._mod_label)
+
         if proc.returncode != 0:
-            stderr_tail = stderr.decode("utf-8", errors="replace").strip()[-1000:]
+            # Include the last stderr lines in the exception for RPC callers.
+            tail = [l for l in stderr_lines if l.strip()][-20:]
+            tail_str = "\n".join(tail) if tail else "(no stderr output)"
             raise RuntimeError(
-                f"Build command failed (exit {proc.returncode}): {self.config.build_command}\n"
-                f"stderr: {stderr_tail}"
+                f"[{self._mod_label}] Build command failed "
+                f"(exit {proc.returncode}): {self.config.build_command}\n"
+                f"--- last stderr ---\n{tail_str}"
             )
         if not exe.exists():
             raise FileNotFoundError(
-                f"Build command succeeded but executable still not found: {exe}\n"
-                f"Build output may have been written to a different path. "
-                f"Check that build_command produces the executable at the expected location."
+                f"[{self._mod_label}] Build command succeeded but executable still not found: {exe}"
             )
 
+        # Seed the cache after a successful build so the next check has a baseline
+        # (needed for the initial build when the pre-build change check was skipped)
+        if self.config.rebuild_on_change:
+            did_change(self._build_cache_name(), self.config.rebuild_on_change, cwd=self.config.cwd)
+
     def _collect_topics(self) -> dict[str, str]:
         """Extract LCM topic strings from blueprint-assigned stream transports."""
         topics: dict[str, str] = {}