Merge remote-tracking branch 'origin/main' into test/compute-release-completion

test_unset_triggers_when_budget_approached assumed a flat 4 chars/token
ratio via "A" * 4 * n. Under tiktoken (available when its vocab is cached,
e.g. on CI) a degenerate all-A string collapses to ~8 chars/token, falling
under the resolved budget so needs_compaction returned False and the
assertion failed. Size payloads with the compactor's own estimate_tokens
so both the trigger and no-trigger assertions hold under the accurate and
heuristic token-counting paths. Production code unchanged.

Co-authored-by: Mervin Praison <[email protected]>

src/praisonai-agents/tests/unit/hooks/test_dead_hook_events.py

@@ -256,32 +256,63 @@ KNOWN_DEAD_OUT_OF_SCOPE = {
 }
 
 
+def _product_event_references(root):
+    """Index references once across the same non-test Python source scope."""
+    import os
+    from pathlib import Path
+    import re
+
+    references = set()
+    pattern = re.compile(r"(?:HookEvent|PluginHook)\.([A-Za-z_][A-Za-z_0-9]*)")
+    for directory, folders, files in os.walk(root):
+        folders[:] = [name for name in folders if name != "tests"]
+        for name in files:
+            path = Path(directory) / name
+            if path.suffix != ".py" or "hooks/types.py" in path.as_posix():
+                continue
+            references.update(pattern.findall(path.read_text(encoding="utf-8", errors="replace")))
+    return references
+
+
+def test_reference_scan_reads_product_files_once(tmp_path, monkeypatch):
+    from pathlib import Path
+
+    product = tmp_path / "src" / "module.py"
+    product.parent.mkdir()
+    product.write_text("HookEvent.BEFORE_TOOL\nPluginHook.ON_INIT\n", encoding="utf-8")
+    excluded = tmp_path / "tests" / "test_fake.py"
+    excluded.parent.mkdir()
+    excluded.write_text("HookEvent.TEST_ONLY\n", encoding="utf-8")
+    definitions = tmp_path / "src" / "hooks" / "types.py"
+    definitions.parent.mkdir()
+    definitions.write_text("HookEvent.DEFINITION_ONLY\n", encoding="utf-8")
+    reads = []
+    original = Path.read_text
+
+    def read(path, *args, **kwargs):
+        reads.append(path)
+        return original(path, *args, **kwargs)
+
+    monkeypatch.setattr(Path, "read_text", read)
+    assert _product_event_references(tmp_path) == {"BEFORE_TOOL", "ON_INIT"}
+    assert reads == [product]
+
+
 def test_no_new_hook_event_is_declared_without_an_emission_site():
     """Every canonical HookEvent must be referenced by non-test product code."""
-    import subprocess
     from pathlib import Path
 
     # tests/unit/hooks/<file> -> repo root
     root = Path(__file__).resolve().parents[5]
 
-    def _refs(name):
-        hits = []
-        for prefix in ("HookEvent.", "PluginHook."):
-            hits += subprocess.run(
-                ["grep", "-rn", "--include=*.py", f"{prefix}{name}", str(root)],
-                capture_output=True, text=True,
-            ).stdout.splitlines()
-        return [
-            h for h in hits
-            if "/tests/" not in h and "hooks/types.py" not in h
-        ]
+    references = _product_event_references(root)
 
     # Control probe (must be non-zero): a known-live event.
-    assert _refs("BEFORE_TOOL"), "control probe failed - the grep found nothing for a live event"
+    assert "BEFORE_TOOL" in references, "control probe failed - no reference to a live event"
     # Control probe (must be zero): a name that does not exist.
-    assert not _refs("DEFINITELY_NOT_AN_EVENT"), "control probe failed - grep matched a bogus name"
+    assert "DEFINITELY_NOT_AN_EVENT" not in references, "control probe matched a bogus name"
 
-    dead = sorted(m.name for m in HookEvent if not _refs(m.name))
+    dead = sorted(m.name for m in HookEvent if m.name not in references)
     unexpected = sorted(set(dead) - KNOWN_DEAD_OUT_OF_SCOPE)
     assert unexpected == [], (
         "HookEvent members declared with no emission site (registering on one "

src/praisonai-agents/tests/unit/test_compaction_wiring.py

@@ -7,7 +7,7 @@ and ContextCompactor is called when enabled.
 """
 
 import pytest
-from unittest.mock import MagicMock, patch
+from unittest.mock import patch
 
 
 class TestExecutionConfigCompactionFields:
@@ -80,15 +80,34 @@ class TestCompactionMaxTokensResolution:
         # Must be far above the old flat 8000 for a large-context model.
         assert resolved > 8000
 
+    @staticmethod
+    def _content_for_tokens(compactor, target_tokens):
+        """Build text whose size is measured by the compactor's own tokeniser.
+
+        A fixed chars-per-token ratio is unsafe: a degenerate ``"A" * n`` string
+        collapses to far fewer tokens under tiktoken (available when its vocab is
+        cached, e.g. on CI) than the 4-chars/token heuristic assumes, so sizing by
+        a hard-coded ratio makes the assertion environment-dependent. Measuring
+        with ``compactor.estimate_tokens`` keeps the test deterministic under both
+        the accurate and heuristic token-counting paths.
+        """
+        word = "lorem ipsum dolor sit amet "
+        text = word * max(1, target_tokens // max(1, compactor.estimate_tokens(word)))
+        while compactor.estimate_tokens(text) < target_tokens:
+            text += word * 256
+        return text
+
     def test_unset_does_not_trigger_within_window(self):
         from praisonaiagents.config.feature_configs import ExecutionConfig
         from praisonaiagents.compaction import ContextCompactor
         agent = self._make_agent("gpt-4o")  # 128k window
         cfg = ExecutionConfig(context_compaction=True)
         resolved = agent._resolve_compaction_max_tokens(cfg)
-        # ~30k tokens of conversation should fit comfortably in a 128k model.
-        msgs = [{"role": "user", "content": "A" * 4 * 30000}]
         compactor = ContextCompactor(max_tokens=resolved)
+        # A conversation well under the resolved budget must not trigger.
+        content = self._content_for_tokens(compactor, resolved // 2)
+        assert compactor.estimate_tokens(content) < resolved
+        msgs = [{"role": "user", "content": content}]
         assert compactor.needs_compaction(msgs) is False
 
     def test_unset_triggers_when_budget_approached(self):
@@ -97,11 +116,11 @@ class TestCompactionMaxTokensResolution:
         agent = self._make_agent("gpt-4o")
         cfg = ExecutionConfig(context_compaction=True)
         resolved = agent._resolve_compaction_max_tokens(cfg)
-        # Well beyond the working budget must trigger compaction. Repeated
-        # characters compress heavily under tiktoken (~8 chars/token), so use
-        # 8× not the heuristic 4× when sizing synthetic overflow content.
-        msgs = [{"role": "user", "content": "A" * 8 * (resolved + 50000)}]
         compactor = ContextCompactor(max_tokens=resolved)
+        # Well beyond the working budget must trigger compaction.
+        content = self._content_for_tokens(compactor, resolved + 50000)
+        assert compactor.estimate_tokens(content) > resolved
+        msgs = [{"role": "user", "content": content}]
         assert compactor.needs_compaction(msgs) is True
 
 

src/praisonai-agents/tests/unit/test_run_on_docker.py

@@ -13,7 +13,6 @@ Everything that needs a live daemon is marked, so it is skipped loudly rather
 than passing quietly when Docker is not running.
 """
 
-import shutil
 import subprocess
 
 import pytest
@@ -308,11 +307,12 @@ def test_a_hosted_backend_reclaims_its_instance_when_collected():
     instance kept billing until the provider's own idle timer noticed, which
     docker and flyio do not have."""
     import gc
-    import time
+    import threading
 
     from praisonai.integrations.compute_managed_agent import ComputeManagedAgent
 
     released = []
+    release_done = threading.Event()
 
     class Fake:
         provider_name = "fake"
@@ -328,6 +328,7 @@ def test_a_hosted_backend_reclaims_its_instance_when_collected():
 
         async def shutdown(self, instance_id):
             released.append(instance_id)
+            release_done.set()
 
     import asyncio
 
@@ -339,12 +340,7 @@ def test_a_hosted_backend_reclaims_its_instance_when_collected():
 
     del backend
     gc.collect()
-    # During live GC the finalizer reclaims fire-and-forget on the shared
-    # AsyncBridge's background loop, so the shutdown lands a beat after
-    # gc.collect() returns. Wait (bounded) for it rather than racing it.
-    deadline = time.monotonic() + 5.0
-    while not released and time.monotonic() < deadline:
-        time.sleep(0.01)
+    assert release_done.wait(5), "the background instance release did not complete"
     assert released == ["inst-1"], "the instance outlived the backend that owned it"
 
 
@@ -368,25 +364,57 @@ def test_release_reclaims_even_on_an_event_loop_thread():
     ever runs, silently leaking the instance -- so _release must bridge the loop
     the way SharedCompute does, not call asyncio.run() directly."""
     import asyncio
-    import time
+    import threading
 
     from praisonai.integrations.compute_managed_agent import _release
 
     released = []
+    release_done = threading.Event()
 
     class Fake:
         async def shutdown(self, instance_id):
             released.append(instance_id)
+            release_done.set()
 
     async def collect_on_the_loop():
         _release(Fake(), "inst-loop", "docker")
+        assert await asyncio.to_thread(release_done.wait, 5), "background release did not complete"
 
     asyncio.run(collect_on_the_loop())
-    # _release routes through the shared AsyncBridge's background loop
-    # fire-and-forget, so the shutdown lands a beat after asyncio.run()
-    # returns. Wait (bounded) for it rather than racing it — the assertion
-    # still fails if _release ever leaks (never schedules the shutdown).
-    deadline = time.monotonic() + 5.0
-    while not released and time.monotonic() < deadline:
-        time.sleep(0.01)
     assert released == ["inst-loop"], "an instance leaked when GC ran on a live loop"
+
+
+def test_release_returns_before_a_slow_provider_finishes():
+    """A live event loop must remain runnable while the cloud teardown waits."""
+    import asyncio
+    import threading
+
+    from praisonai.integrations.compute_managed_agent import _release
+
+    started = threading.Event()
+    allow_completion = threading.Event()
+    completed = threading.Event()
+    released = []
+
+    class Fake:
+        async def shutdown(self, instance_id):
+            started.set()
+            try:
+                assert await asyncio.to_thread(allow_completion.wait, 5)
+                released.append(instance_id)
+            finally:
+                completed.set()
+
+    async def collect_on_the_loop():
+        try:
+            _release(Fake(), "slow-instance", "docker")
+            assert await asyncio.to_thread(started.wait, 5)
+            assert not completed.is_set(), "release blocked the caller until teardown ended"
+            # This caller's loop can still advance before permitting teardown.
+            await asyncio.sleep(0)
+        finally:
+            allow_completion.set()
+            assert await asyncio.to_thread(completed.wait, 5)
+
+    asyncio.run(collect_on_the_loop())
+    assert released == ["slow-instance"]

src/praisonai-agents/tests/unit/workspace/test_workspace_isolation.py

@@ -52,7 +52,12 @@ def test_no_isolation_shares_root(tmp_path):
     adapter.remove("a")
 
 
-def test_git_worktree_degrades_when_not_a_repo(tmp_path):
+def test_git_worktree_degrades_when_not_a_repo(tmp_path, monkeypatch):
+    # A temporary directory can have a repository ancestor (including /).
+    # Keep this negative fixture outside that ancestor's discovery scope.
+    monkeypatch.setenv("GIT_CEILING_DIRECTORIES", str(tmp_path.parent))
+    monkeypatch.delenv("GIT_DIR", raising=False)
+    monkeypatch.delenv("GIT_WORK_TREE", raising=False)
     adapter = GitWorktreeAdapter(root=tmp_path)
     assert adapter.available is False
     assert adapter.create("run") == str(tmp_path.resolve())

src/praisonai-desktop/engine/test_training.py

@@ -661,6 +661,20 @@ class SurvivingARestart(unittest.TestCase):
         self.first.command_builder = _script("print('done')")
         run = self.first.start(self.config, "run-done")
         self.assertTrue(_wait(lambda: run.state in training.TERMINAL), run.state)
+        import json
+
+        state_file = pathlib.Path(self.home, "runs", "run-done", "run.json")
+
+        def ending_is_persisted():
+            try:
+                saved = json.loads(state_file.read_text(encoding="utf-8"))
+            except (OSError, json.JSONDecodeError):
+                return False
+            return saved.get("state") == training.DONE
+
+        # finish() publishes the in-memory state before the supervisor writes
+        # run.json. Restart only once that asynchronous write has completed.
+        self.assertTrue(_wait(ending_is_persisted), "the completed run was not persisted")
         second = training.Trainer(self.home, sys.executable)
         self.assertEqual(second.get("run-done").state, training.DONE)
         self.assertIsNone(second.current, "a finished run was adopted as live")