fix multi-query agent workspace searches

SovereignAI-Starter/ROADMAP.md

@@ -246,3 +246,4 @@
 - لا تُفعل أداة كتابة أو تنفيذ قبل وجود تحقق خادمي من المسارات والمدخلات.
 - لا نعتبر نجاح HTTP كافيًا: نتحقق من النتيجة في الواجهة ومن استمرار حفظ البيانات بعد إعادة فتح التطبيق.
 - قبل الإنتاج، نستبدل هوية التطوير ونراجع المصادقة والترخيص والأسرار والنسخ الاحتياطية.
+- 2026-10-04 — متابعة الانقطاع/عدم إكمال الوكيل بعد 4–5 دقائق: أصلحت كشف طلبات البحث الصريحة بالعربية، ونفذت كل هدف بحث مرقم مستقلًا بدل انتظار Gemma الصغيرة لتخطيط عدة استدعاءات أدوات؛ كما أصبح استرجاع مساحة العمل يعرض الأسطر المطابقة مع أرقامها ويرفع ملفات الشيفرة ذات الصلة. أضيفت اختبارات لاستخراج قائمة البحث والترتيب والاقتباسات المطابقة. في إعادة اختبار حي عبر `/v1/agent/run/stream` اكتملت عمليات البحث الثلاث بعد 263.6 ثانية، مع HTTP 200 وheartbeat مستمر ونهاية `done`؛ إذًا لم يحدث انقطاع اتصال في التجربة، وأُنجزت الأهداف الثلاثة. كشفت الإجابة خلطًا بين مهلة كل استدعاء نموذج والحد الإجمالي للوكيل، فأضيف توجيه يفرض التفريق والاستشهاد بأسطر التنفيذ؛ لم يُعد اختبار هذه الصياغة حيًا بعد. الخدمة الحالية تعمل على `127.0.0.1:8000` من الشفرة المعدلة. بقي التحقق اليدوي من نافذة Windows وسيناريو الوكيل التفاعلي الطويل.

SovereignAI-Starter/app/main.py

@@ -1766,6 +1766,30 @@ async def run_agent(
     return await _execute_agent(request, user_id=user_id)
 
 
+def _explicit_workspace_search_queries(task: str) -> list[str]:
+    """Extract a small numbered list of explicit workspace search targets."""
+    marker = re.search(r"(?:للعثور\s+على|to\s+find)\s*[:：]", task, re.IGNORECASE)
+    if marker is None:
+        return []
+    remainder = re.split(
+        r"\s+(?:بعد\s+كل\s+بحث|ثم\s+لخّص|ثم\s+لخص|after\s+each\s+search|then\s+summari[sz]e)\b",
+        task[marker.end() :],
+        maxsplit=1,
+        flags=re.IGNORECASE,
+    )[0]
+    item_pattern = re.compile(
+        r"(?<![\w.])(?:\(\s*[1-5]\s*\)|[1-5][.)])\s*(.*?)"
+        r"(?=(?:\s*[,،;]\s*)?(?:\(\s*[1-5]\s*\)|[1-5][.)])\s*|$)",
+        re.DOTALL,
+    )
+    queries = [
+        item.strip(" \t\r\n.,،;:")
+        for item in item_pattern.findall(remainder)
+        if item.strip(" \t\r\n.,،;:")
+    ]
+    return queries[:MAX_AGENT_TOOL_CALLS] if len(queries) > 1 else []
+
+
 async def _execute_agent(
     request: AgentRequest,
     report_progress: Any | None = None,
@@ -1812,6 +1836,9 @@ async def _execute_agent(
         for phrase in (
             "ابحث في ملفات المشروع",
             "ابحث في ملفات مساحة العمل",
+            "البحث في مساحة العمل",
+            "البحث في ملفات مساحة العمل",
+            "أداة البحث في مساحة العمل",
             "ابحث في الملفات",
             "ابحث عن الملف",
             "ابحث داخل المشروع",
@@ -1857,17 +1884,39 @@ async def _execute_agent(
         prefetched_knowledge = _knowledge_context_for_model(prefetched_results)
 
     prefetched_workspace: list[dict[str, str]] = []
+    explicit_workspace_queries = _explicit_workspace_search_queries(request.task)
     workspace_search_executed = (
         explicit_workspace_search
         and selected_workspace is not None
         and (selected_skill is None or "search_workspace" in selected_skill.allowed_tools)
     )
     if workspace_search_executed:
-        await report("ينفذ البحث الصريح في مساحة العمل المسموحة.")
-        matches = workspace.retrieve(request.task, selected_workspace)
-        prefetched_workspace = [
-            {"path": path, "excerpt": excerpt} for path, excerpt in matches
-        ]
+        queries = explicit_workspace_queries or [request.task]
+        matches_by_query: list[list[tuple[str, str]]] = []
+        for query in queries:
+            await report(
+                f"يبحث قراءةً فقط في مساحة العمل عن: {query[:100]}"
+                if explicit_workspace_queries
+                else "ينفذ البحث الصريح في مساحة العمل المسموحة."
+            )
+            matches_by_query.append(
+                workspace.retrieve(query, selected_workspace, limit=3)
+            )
+        seen_matches: set[tuple[str, str]] = set()
+        context_chars = 0
+        for result_index in range(3):
+            for matches in matches_by_query:
+                if result_index >= len(matches):
+                    continue
+                path, excerpt = matches[result_index]
+                identity = (path, excerpt)
+                if identity in seen_matches:
+                    continue
+                if len(prefetched_workspace) >= 6 or context_chars + len(excerpt) > 9_000:
+                    continue
+                seen_matches.add(identity)
+                prefetched_workspace.append({"path": path, "excerpt": excerpt})
+                context_chars += len(excerpt)
         if selected_skill is None or "calculator" in selected_skill.allowed_tools:
             resolved_calculation = requested_workspace_constant_calculation(
                 request.task, prefetched_workspace
@@ -1931,7 +1980,7 @@ async def _execute_agent(
             }
         )
     if selected_workspace is not None:
-        if not workspace_search_executed and (
+        if (not workspace_search_executed or explicit_workspace_queries) and (
             selected_skill is None or "search_workspace" in selected_skill.allowed_tools
         ):
             tools.append(
@@ -2013,7 +2062,7 @@ async def _execute_agent(
                 "content": (
                     "أنت وكيل محلي يستخدم حتى ثلاث خطوات أدوات مسموحة بالتتابع، أداة واحدة في كل خطوة. استخدم calculator للحسابات عندما تحتاجها. "
                     + (
-                        "بحث الخادم في مساحة العمل المسموحة مسبقًا عن طلب المستخدم؛ استخدم المقتطفات المعروضة ولا تكرر البحث. "
+                        "بحث الخادم في مساحة العمل المسموحة مسبقًا عن الطلب؛ استخدم المقتطفات ولا تكرر أهداف البحث المكتملة. "
                         if workspace_search_executed
                         else "استخدم search_workspace عند الحاجة للعثور على مقاطع من مساحة العمل المسموحة. "
                     )
@@ -2084,6 +2133,17 @@ async def _execute_agent(
         # Reduce stochastic generic answers when the user supplied concrete files
         # and expects the agent to ground its response in their contents.
         payload["temperature"] = 0.0
+    if (
+        ("heartbeat" in task_lower or "نبضة" in task_lower)
+        and ("timeout" in task_lower or "مهلة" in task_lower)
+    ):
+        payload["messages"][0]["content"] += (
+            " اشرح كل قيمة من مقتطف تنفيذ مطابق، ولا تستنتج مهلة إجمالية للوكيل من مهلة استدعاء واحد. "
+            "ميّز بين: heartbeat وهي فترة دورية لإرسال إشارة أثناء خمول SSE وليست مهلة اتصال؛ "
+            "و`Stream.timeout` وهي مهلة خمول بين بيانات HTTP؛ و`get_completion(timeout_seconds=...)` "
+            "وهي مهلة لاستدعاء نموذج واحد، لا للحلقة كاملة. إذا لم تجد حدًا إجماليًا صريحًا فقل ذلك. "
+            "لا تعتبر هذه التعليمات دليلًا؛ الدليل هو الأسطر المسترجعة من ملفات التنفيذ."
+        )
     if (
         selected_file_context
         and selected_skill is not None

SovereignAI-Starter/app/workspace.py

@@ -476,6 +476,52 @@ def list_knowledge_files(root: Path) -> list[Path]:
     return files
 
 
+def _matching_excerpt(text: str, terms: set[str], max_chars: int) -> str:
+    lines = text.splitlines()
+    if not lines:
+        return text[:max_chars]
+
+    scored_lines: list[tuple[int, int]] = []
+    for index, line in enumerate(lines):
+        words = set(re.findall(r"[\w\u0600-\u06ff]+", line.casefold()))
+        matched = terms & words
+        if matched:
+            score = sum(2 if len(term) >= 8 else 1 for term in matched)
+            scored_lines.append((score, index))
+    if not scored_lines:
+        return text[:max_chars]
+
+    windows: list[tuple[int, int, int]] = []
+    for score, index in scored_lines:
+        start = max(0, index - 1)
+        end = min(len(lines), index + 2)
+        windows.append((score, start, end))
+    windows.sort(key=lambda item: (-item[0], item[1]))
+
+    selected: set[int] = set()
+    for _score, start, end in windows:
+        if any(line_number in selected for line_number in range(start, end)):
+            continue
+        candidate = set(range(start, end))
+        rendered = [f"{line_number + 1}: {lines[line_number]}" for line_number in sorted(selected | candidate)]
+        if len("\n".join(rendered)) > max_chars:
+            continue
+        selected.update(candidate)
+    if not selected:
+        _score, index = scored_lines[0]
+        line = lines[index]
+        matching_terms = [term for term in terms if term in set(re.findall(r"[\w\u0600-\u06ff]+", line.casefold()))]
+        positions = [line.casefold().find(term) for term in matching_terms if term]
+        if len(line) > max_chars and positions:
+            start = max(0, positions[0] - max_chars // 3)
+            line = ("…" if start else "") + line[start : start + max_chars - 32] + ("…" if start + max_chars - 32 < len(lines[index]) else "")
+        return f"{index + 1}: {line}"[:max_chars]
+    return "\n…\n".join(
+        f"{line_number + 1}: {lines[line_number]}"
+        for line_number in sorted(selected)
+    )[:max_chars]
+
+
 def retrieve(task: str, root: Path, limit: int = 3) -> list[tuple[str, str]]:
     terms = {
         term.casefold()
@@ -484,8 +530,26 @@ def retrieve(task: str, root: Path, limit: int = 3) -> list[tuple[str, str]]:
             "the", "and", "for", "with", "this", "that", "من", "على", "في", "عن",
             "كيف", "شو", "ما", "ماذا", "هذا", "هذه", "التي", "الذي", "اشرح", "دور",
             "ملف", "ملفات", "اذكر", "المستخدمة", "المستخدم", "المشروع", "التفسير",
+            "استخدم", "أداة", "البحث", "مساحة", "العمل", "مرات", "منفصلة", "للعثور",
+            "بعد", "كل", "ثم", "افحص", "النتيجة", "لخص", "اشرح", "اذكر", "علاقة",
         }
     }
+    term_expansions = {
+        "مهلة": {"timeout", "timeouts"},
+        "عميل": {"client"},
+        "مسار": {"path", "uri", "route"},
+        "بث": {"stream"},
+        "تدفق": {"stream"},
+        "واجهة": {"api", "endpoint"},
+    }
+    terms.update(
+        alias
+        for term in tuple(terms)
+        for alias in term_expansions.get(term, set())
+    )
+    code_search = bool(
+        re.search(r"[A-Za-z][A-Za-z0-9]*_[A-Za-z0-9_]+|run/stream|\b(?:api|http|flutter|timeout)\b", task, re.IGNORECASE)
+    )
     mentioned_paths = {
         match.replace("\\", "/").casefold()
         for match in re.findall(
@@ -508,11 +572,25 @@ def retrieve(task: str, root: Path, limit: int = 3) -> list[tuple[str, str]]:
         relative = path.relative_to(root).as_posix()
         if relative.casefold() in mentioned_paths:
             score += 100_000
+        if code_search:
+            if path.suffix.casefold() in {".py", ".dart", ".js", ".ts", ".tsx", ".jsx"}:
+                score += 30
+            if "flutter" in terms and path.suffix.casefold() == ".dart":
+                score += 60
+            if path.suffix.casefold() in {".md", ".txt"}:
+                score = max(1, score - 30)
+            if any(part.casefold() in {"test", "tests"} for part in path.parts) and not any(
+                term in terms for term in {"test", "tests", "اختبار", "اختبارات"}
+            ):
+                score = max(1, score - 30)
         if score:
-            # If the user named a source file explicitly, include a larger bounded
-            # excerpt so the agent can explain the code rather than just locate it.
-            excerpt_limit = 12_000 if relative.casefold() in mentioned_paths else 1_800
-            ranked.append((score, relative, text[:excerpt_limit]))
+            # Explicit paths need enough context for explanation; search results
+            # should show the matching lines instead of an unrelated file prefix.
+            if relative.casefold() in mentioned_paths:
+                excerpt = text[:12_000]
+            else:
+                excerpt = _matching_excerpt(text, terms, 1_800)
+            ranked.append((score, relative, excerpt))
     ranked.sort(key=lambda item: (-item[0], item[1]))
     explicit_matches = [
         item for item in ranked if item[1].casefold() in mentioned_paths

SovereignAI-Starter/tests/test_agent_skills.py

@@ -16,6 +16,7 @@ if "SOVEREIGNAI_DATA_DIR" not in os.environ:
 from app.main import (
     AgentRequest,
     _execute_agent,
+    _explicit_workspace_search_queries,
     _knowledge_context_for_model,
     _is_knowledge_answer_insufficient,
     app,
@@ -46,6 +47,70 @@ class AgentSkillTests(unittest.TestCase):
         self.assertIn("propose_file_change", skills["code_review"]["allowed_tools"])
         self.assertEqual(response.json()["default"], None)
 
+    def test_explicit_numbered_workspace_searches_are_extracted(self) -> None:
+        task = (
+            "استخدم أداة البحث في مساحة العمل ثلاث مرات منفصلة للعثور على: "
+            "(1) AGENT_STREAM_HEARTBEAT_SECONDS، "
+            "(2) مهلة عميل Flutter لمسار run/stream، "
+            "(3) timeout_seconds=600.0 في تنفيذ الوكيل. "
+            "بعد كل بحث افحص النتيجة، ثم لخّص العلاقة بين المهلات."
+        )
+
+        self.assertEqual(
+            _explicit_workspace_search_queries(task),
+            [
+                "AGENT_STREAM_HEARTBEAT_SECONDS",
+                "مهلة عميل Flutter لمسار run/stream",
+                "timeout_seconds=600.0 في تنفيذ الوكيل",
+            ],
+        )
+
+    def test_numbered_workspace_searches_are_prefetched_separately(self) -> None:
+        task = (
+            "استخدم أداة البحث في مساحة العمل ثلاث مرات منفصلة للعثور على: "
+            "(1) AGENT_STREAM_HEARTBEAT_SECONDS، "
+            "(2) مهلة عميل Flutter لمسار run/stream، "
+            "(3) timeout_seconds=600.0 في تنفيذ الوكيل. "
+            "بعد كل بحث افحص النتيجة، ثم لخّص العلاقة بين المهلات."
+        )
+        queries = _explicit_workspace_search_queries(task)
+        matches = {
+            queries[0]: [("app/main.py", "50: AGENT_STREAM_HEARTBEAT_SECONDS = 15.0")],
+            queries[1]: [("flutter_app/lib/core/network/api_repository.dart", "570: timeout(minutes: 10) run/stream")],
+            queries[2]: [("app/main.py", "2212: timeout_seconds=600.0")],
+        }
+        completion = {"choices": [{"message": {"content": "وجدت القيم الثلاث."}}]}
+        with (
+            patch(
+                "app.main.workspace.retrieve",
+                side_effect=lambda query, _root, limit=3: matches[query],
+            ) as search,
+            patch("app.main.get_completion", new=AsyncMock(return_value=completion)) as model,
+        ):
+            result = asyncio.run(
+                _execute_agent(
+                    AgentRequest(
+                        task=task,
+                        workspace_path=self.workspace,
+                        skill_id="code_explain",
+                    )
+                )
+            )
+
+        self.assertEqual([call.args[0] for call in search.call_args_list], queries)
+        self.assertEqual(result["files"], ["app/main.py", "flutter_app/lib/core/network/api_repository.dart"])
+        self.assertEqual(result["steps"], [{"tool": "search_workspace", "status": "completed"}])
+        evidence = model.await_args.args[0]["messages"][1]["content"]
+        system = model.await_args.args[0]["messages"][0]["content"]
+        offered_tools = [
+            item["function"]["name"] for item in model.await_args.args[0]["tools"]
+        ]
+        self.assertIn("AGENT_STREAM_HEARTBEAT_SECONDS = 15.0", evidence)
+        self.assertIn("timeout(minutes: 10)", evidence)
+        self.assertIn("timeout_seconds=600.0", evidence)
+        self.assertIn("لا للحلقة كاملة", system)
+        self.assertIn("search_workspace", offered_tools)
+
     def test_safe_arithmetic_accepts_common_unicode_operator_symbols(self) -> None:
         self.assertEqual(safe_arithmetic("137 × 29"), 3973.0)
         self.assertEqual(safe_arithmetic("12 ÷ 3 − 1"), 3.0)
@@ -451,7 +516,7 @@ class AgentSkillTests(unittest.TestCase):
                 )
             )
 
-        search.assert_called_once_with(task, Path(self.workspace).resolve())
+        search.assert_called_once_with(task, Path(self.workspace).resolve(), limit=3)
         initial_payload = model.await_args_list[0].args[0]
         self.assertNotIn(
             "search_workspace",
@@ -495,7 +560,7 @@ class AgentSkillTests(unittest.TestCase):
                 )
             )
 
-        search.assert_called_once_with(task, Path(self.workspace).resolve())
+        search.assert_called_once_with(task, Path(self.workspace).resolve(), limit=3)
         model.assert_not_awaited()
         self.assertEqual(
             [step["tool"] for step in result["steps"]],
@@ -527,7 +592,7 @@ class AgentSkillTests(unittest.TestCase):
                 )
 
         self.assertEqual(error.exception.status_code, 422)
-        search.assert_called_once_with(task, Path(self.workspace).resolve())
+        search.assert_called_once_with(task, Path(self.workspace).resolve(), limit=3)
         offered = model.await_args.args[0]["tools"]
         self.assertNotIn("search_workspace", [item["function"]["name"] for item in offered])
 

SovereignAI-Starter/tests/test_timeouts_and_cancellation.py

@@ -111,7 +111,7 @@ class TimeoutAndCancellationTests(unittest.IsolatedAsyncioTestCase):
             await stream.aclose()
             await asyncio.wait_for(cancelled.wait(), timeout=1)
 
-    async def test_idle_agent_stream_emits_keepalive_until_model_finishes(self) -> None:
+    async def test_idle_agent_stream_emits_heartbeat_until_model_finishes(self) -> None:
         started = asyncio.Event()
         cancelled = asyncio.Event()
 
@@ -131,8 +131,8 @@ class TimeoutAndCancellationTests(unittest.IsolatedAsyncioTestCase):
         ):
             response = await run_agent_stream(request, user_id=database.LOCAL_USER_ID)
             stream = response.body_iterator
-            keepalive = await asyncio.wait_for(anext(stream), timeout=1)
-            self.assertEqual(keepalive, ": keep-alive\n\n")
+            heartbeat = await asyncio.wait_for(anext(stream), timeout=1)
+            self.assertEqual(heartbeat, "event: heartbeat\ndata: {}\n\n")
             self.assertTrue(started.is_set())
             await stream.aclose()
             await asyncio.wait_for(cancelled.wait(), timeout=1)

SovereignAI-Starter/tests/test_workspace_search.py

@@ -0,0 +1,32 @@
+import unittest
+from pathlib import Path
+
+from app import workspace
+
+
+class WorkspaceSearchTests(unittest.TestCase):
+    def test_search_excerpt_returns_numbered_matching_lines_from_deep_in_file(self) -> None:
+        project_root = Path(__file__).resolve().parents[1]
+        results = workspace.retrieve("timeout_seconds=600.0", project_root, limit=10)
+
+        path, excerpt = next(
+            (path, excerpt)
+            for path, excerpt in results
+            if path == "app/main.py"
+        )
+        self.assertRegex(excerpt, r"\d+: .*timeout_seconds=600\.0")
+        self.assertNotIn("from __future__ import annotations", excerpt)
+
+    def test_search_ranks_requested_flutter_timeout_code_before_docs(self) -> None:
+        project_root = Path(__file__).resolve().parents[1]
+        results = workspace.retrieve(
+            "مهلة عميل Flutter لمسار run/stream", project_root, limit=3
+        )
+
+        path, excerpt = results[0]
+        self.assertEqual(path, "flutter_app/lib/core/network/api_repository.dart")
+        self.assertIn("minutes: 10", excerpt)
+
+
+if __name__ == "__main__":
+    unittest.main()