get entire test suite working with offload

.dockerignore

@@ -0,0 +1,14 @@
+# Exclude directories that are modified by concurrent processes (agents, tests)
+# during Modal's image upload. The Docker image only needs current.tar.gz;
+# all other repo files are delivered via the tarball.
+.git/
+**/__pycache__/
+**/*.pyc
+**/.pytest_cache/
+**/.test_output/
+.venv/
+**/.ruff_cache/
+.mngr/dev/
+test-results/
+tmr_*/
+node_modules/

.gitignore

@@ -248,6 +248,8 @@ tmr_*/
 
 # Offload caches and local files
 .offload/**
+.offload-image-cache
+.offload-cache-key
 test-results
 current.tar.gz
 

.mngr/settings.toml

@@ -2,6 +2,9 @@ is_allowed_in_pytest = false
 #default_destroyed_host_persisted_seconds = 36000
 default_destroyed_host_persisted_seconds = 10
 
+[work_dir_extra_paths]
+test-results = "SHARE"
+
 [pre_command_scripts]
 create = ["""bash -c './scripts/make_tar_of_repo.sh `cat .mngr/image_commit_hash` .mngr/dev/build'"""]
 

.offload-base-commit

@@ -1 +1 @@
-73ce8f3cc81924aba41ecf000524975b18979f6a
+36f9e3b6d88585b335655d893890adeb123d4a97

.offload-image-cache

@@ -1 +1 @@
-im-Hg9iOacASyJ2Im5Gt1qJ91
+im-xNeY6u8qRzETya9SVpJWdi

apps/minds/imbue/minds/forwarding_server/agent_creator.py

@@ -288,6 +288,7 @@ class AgentCreator(MutableModel):
     _redirect_urls: dict[str, str] = PrivateAttr(default_factory=dict)
     _errors: dict[str, str] = PrivateAttr(default_factory=dict)
     _log_queues: dict[str, queue.Queue[str]] = PrivateAttr(default_factory=dict)
+    _threads: dict[str, threading.Thread] = PrivateAttr(default_factory=dict)
     _lock: threading.Lock = PrivateAttr(default_factory=threading.Lock)
 
     def start_creation(
@@ -317,9 +318,25 @@ def start_creation(
             daemon=True,
             name="agent-creator-{}".format(agent_id),
         )
+        with self._lock:
+            self._threads[str(agent_id)] = thread
         thread.start()
         return agent_id
 
+    def wait_for_completion(self, agent_id: AgentId, timeout: float) -> None:
+        """Wait for the background creation thread to finish."""
+        with self._lock:
+            thread = self._threads.get(str(agent_id))
+        if thread is not None:
+            thread.join(timeout=timeout)
+
+    def close(self, timeout: float = 10.0) -> None:
+        """Wait for all background creation threads to finish."""
+        with self._lock:
+            threads = list(self._threads.values())
+        for thread in threads:
+            thread.join(timeout=timeout)
+
     def get_creation_info(self, agent_id: AgentId) -> AgentCreationInfo | None:
         """Get the current creation status for an agent, or None if not tracked."""
         with self._lock:

apps/minds/imbue/minds/forwarding_server/agent_creator_test.py

@@ -179,6 +179,9 @@ def test_agent_creator_start_creation_returns_agent_id_and_tracks_status(tmp_pat
     assert info.agent_id == agent_id
     assert info.status == AgentCreationStatus.CLONING
 
+    # Wait for the background thread to finish so git clone doesn't leak.
+    creator.wait_for_completion(agent_id, timeout=10.0)
+
 
 def test_agent_creator_start_creation_with_custom_name(tmp_path: Path) -> None:
     """Verify start_creation accepts a custom agent name."""
@@ -189,6 +192,8 @@ def test_agent_creator_start_creation_with_custom_name(tmp_path: Path) -> None:
     info = creator.get_creation_info(agent_id)
     assert info is not None
 
+    creator.wait_for_completion(agent_id, timeout=10.0)
+
 
 def test_agent_creator_get_log_queue_returns_none_for_unknown() -> None:
     creator = AgentCreator(
@@ -205,6 +210,8 @@ def test_agent_creator_get_log_queue_returns_queue_for_tracked() -> None:
     q = creator.get_log_queue(agent_id)
     assert q is not None
 
+    creator.close()
+
 
 def test_make_log_callback_puts_lines_into_queue() -> None:
     log_queue: queue_mod.Queue[str] = queue_mod.Queue()

apps/minds/imbue/minds/forwarding_server/backend_resolver_test.py

@@ -571,6 +571,8 @@ def test_stream_manager_full_snapshot_updates_agent_ids() -> None:
     )
     with manager._cg:
         manager._handle_discovery_line(line)
+        for process in manager._events_processes.values():
+            process.terminate()
 
     ids = manager.resolver.list_known_agent_ids()
     assert _AGENT_A in ids
@@ -601,6 +603,9 @@ def test_stream_manager_host_ssh_info_populates_resolver() -> None:
         ssh_line = _make_host_ssh_info_line(host_id, ssh_data)
         manager._handle_discovery_line(ssh_line)
 
+        for process in manager._events_processes.values():
+            process.terminate()
+
     ssh_info = manager.resolver.get_ssh_info(_AGENT_A)
     assert ssh_info is not None
     assert ssh_info.host == "remote.example.com"
@@ -620,6 +625,9 @@ def test_stream_manager_no_ssh_for_local_hosts() -> None:
         )
         manager._handle_discovery_line(line)
 
+        for process in manager._events_processes.values():
+            process.terminate()
+
     assert manager.resolver.list_known_agent_ids() == (_AGENT_A,)
     assert manager.resolver.get_ssh_info(_AGENT_A) is None
 
@@ -647,6 +655,11 @@ def test_stream_manager_mixed_local_and_remote() -> None:
         ssh_line = _make_host_ssh_info_line(remote_host_id, ssh_data)
         manager._handle_discovery_line(ssh_line)
 
+        # Terminate background mngr events processes before the CG exits,
+        # otherwise they time out on slow systems (e.g. Modal containers).
+        for process in manager._events_processes.values():
+            process.terminate()
+
     assert manager.resolver.get_ssh_info(_AGENT_A) is None
     ssh_info = manager.resolver.get_ssh_info(_AGENT_B)
     assert ssh_info is not None
@@ -677,6 +690,9 @@ def test_stream_manager_ssh_info_before_full_snapshot() -> None:
         )
         manager._handle_discovery_line(full_line)
 
+        for process in manager._events_processes.values():
+            process.terminate()
+
     ssh_info = manager.resolver.get_ssh_info(_AGENT_A)
     assert ssh_info is not None
     assert ssh_info.host == "remote.example.com"

apps/minds/imbue/minds/forwarding_server/ssh_tunnel.py

@@ -19,7 +19,7 @@
 
 _SELECT_TIMEOUT_SECONDS: Final[float] = 1.0
 
-_ACCEPT_TIMEOUT_SECONDS: Final[float] = 1.0
+_SHUTDOWN_POLL_SECONDS: Final[float] = 0.2
 
 _SOCKET_POLL_SECONDS: Final[float] = 0.01
 
@@ -234,7 +234,7 @@ def _tunnel_accept_loop(
         server.bind(str(sock_path))
         os.chmod(str(sock_path), 0o600)
         server.listen(8)
-        server.settimeout(_ACCEPT_TIMEOUT_SECONDS)
+        server.settimeout(_SHUTDOWN_POLL_SECONDS)
 
         while not shutdown_event.is_set():
             try:

apps/minds/imbue/minds/forwarding_server/ssh_tunnel_test.py

@@ -329,5 +329,5 @@ def test_tunnel_accept_loop_shutdown_event_stops_loop(short_tmp_path: Path) -> N
     _wait_for_socket(sock_path, timeout=10.0)
 
     shutdown_event.set()
-    accept_thread.join(timeout=3.0)
+    accept_thread.join(timeout=10.0)
     assert not accept_thread.is_alive()