feat: bound repository context size in load_repo_context (#24)

repo_router.py

@@ -19,6 +19,58 @@
 # rather than trying to block private/internal IP ranges for an open-ended fetcher.
 ALLOWED_EXTERNAL_DOMAINS = {"stackoverflow.com"}
 
+# Context size bounds for load_repo_context
+DEFAULT_MAX_FILE_CHARS = 2000
+DEFAULT_MAX_TOTAL_CONTEXT_CHARS = 12000
+TRUNCATION_MARKER = "\n... (truncated)"
+
+
+def _read_capped_file(file_path: Path, max_chars: int = DEFAULT_MAX_FILE_CHARS) -> str:
+    """Read a Markdown context file up to max_chars characters.
+    If the file content exceeds max_chars, return truncated content with a truncation marker.
+    """
+    with open(file_path, "r", encoding="utf-8") as f:
+        content = f.read()
+    if len(content) > max_chars:
+        return content[:max_chars] + TRUNCATION_MARKER
+    return content
+
+
+def _append_bounded_section(
+    context_parts: list[str],
+    header: str,
+    content: str,
+    max_total_chars: int = DEFAULT_MAX_TOTAL_CONTEXT_CHARS,
+) -> bool:
+    """Append a section header and content to context_parts while enforcing max_total_chars.
+    Returns True if section was added in full, False if section was truncated or skipped due to budget.
+    """
+    # Current assembled length if joined with "\n\n"
+    current_len = sum(len(p) for p in context_parts) + (len("\n\n") * max(0, len(context_parts) - 1))
+    separator_len = len("\n\n") if context_parts else 0
+
+    remaining_budget = max_total_chars - current_len - separator_len
+    if remaining_budget <= 0:
+        return False
+
+    prefix = f"{header}\n"
+    if len(prefix) >= remaining_budget:
+        return False
+
+    content_budget = remaining_budget - len(prefix)
+    if len(content) > content_budget:
+        marker = TRUNCATION_MARKER
+        if content_budget > len(marker):
+            capped_content = content[: content_budget - len(marker)] + marker
+        else:
+            capped_content = marker[:content_budget]
+        context_parts.append(f"{prefix}{capped_content}")
+        return False
+    else:
+        context_parts.append(f"{prefix}{content}")
+        return True
+
+
 
 def get_repo_details(repo_name: str) -> dict:
     """Get metadata for a repository directly from REPO_METADATA."""
@@ -142,15 +194,25 @@ def _keyword_hit(text: str, keyword: str) -> bool:
     return re.search(rf"\b{re.escape(keyword)}\b", text) is not None
 
 
-def load_repo_context(repo_name: str, query: str = "") -> str:
+def load_repo_context(
+    repo_name: str,
+    query: str = "",
+    max_file_chars: int = DEFAULT_MAX_FILE_CHARS,
+    max_total_chars: int = DEFAULT_MAX_TOTAL_CONTEXT_CHARS,
+) -> str:
     """Dynamically load context: ALWAYS load operational-data.md and architecture.md,
-    and dynamically load specific instruction/core files based on user query intent."""
+    and dynamically load specific instruction/core files based on user query intent.
+    Enforces per-file size limits and an overall total context budget."""
     context_parts = []
     repo_path = get_repo_path(repo_name)
     if not repo_path:
         return ""
 
-    context_parts.append(f"=== REPOSITORY: {repo_name} ===")
+    repo_header = f"=== REPOSITORY: {repo_name} ==="
+    if len(repo_header) > max_total_chars:
+        return repo_header[:max_total_chars]
+    context_parts.append(repo_header)
+
     q = query.lower()
 
     agent_dir = repo_path / ".agent"
@@ -159,17 +221,27 @@ def load_repo_context(repo_name: str, query: str = "") -> str:
         ops_md = agent_dir / "info" / "operational-data.md"
         if ops_md.exists():
             try:
-                with open(ops_md, "r", encoding="utf-8") as f:
-                    context_parts.append(f"--- Operational Data (.agent/info/operational-data.md) ---\n{f.read()}")
+                content = _read_capped_file(ops_md, max_chars=max_file_chars)
+                _append_bounded_section(
+                    context_parts,
+                    "--- Operational Data (.agent/info/operational-data.md) ---",
+                    content,
+                    max_total_chars=max_total_chars,
+                )
             except Exception as e:
                 logger.error(f"Error reading operational data file {ops_md}: {e}")
 
         # 2. ALWAYS LOAD: Core Architecture (.agent/core/architecture.md)
         arch_md = agent_dir / "core" / "architecture.md"
         if arch_md.exists():
             try:
-                with open(arch_md, "r", encoding="utf-8") as f:
-                    context_parts.append(f"--- Core Architecture (.agent/core/architecture.md) ---\n{f.read()}")
+                content = _read_capped_file(arch_md, max_chars=max_file_chars)
+                _append_bounded_section(
+                    context_parts,
+                    "--- Core Architecture (.agent/core/architecture.md) ---",
+                    content,
+                    max_total_chars=max_total_chars,
+                )
             except Exception as e:
                 logger.error(f"Error reading architecture file {arch_md}: {e}")
 
@@ -198,8 +270,13 @@ def load_repo_context(repo_name: str, query: str = "") -> str:
                 if should_load:
                     rel_path = core_file.relative_to(repo_path)
                     try:
-                        with open(core_file, "r", encoding="utf-8") as f:
-                            context_parts.append(f"--- Core Context ({rel_path}) ---\n{f.read()}")
+                        content = _read_capped_file(core_file, max_chars=max_file_chars)
+                        _append_bounded_section(
+                            context_parts,
+                            f"--- Core Context ({rel_path}) ---",
+                            content,
+                            max_total_chars=max_total_chars,
+                        )
                     except Exception as e:
                         logger.error(f"Error reading core file {core_file}: {e}")
 
@@ -229,8 +306,13 @@ def load_repo_context(repo_name: str, query: str = "") -> str:
                 if should_load:
                     rel_path = md_file.relative_to(repo_path)
                     try:
-                        with open(md_file, "r", encoding="utf-8") as f:
-                            context_parts.append(f"--- Instruction ({rel_path}) ---\n{f.read()}")
+                        content = _read_capped_file(md_file, max_chars=max_file_chars)
+                        _append_bounded_section(
+                            context_parts,
+                            f"--- Instruction ({rel_path}) ---",
+                            content,
+                            max_total_chars=max_total_chars,
+                        )
                     except Exception as e:
                         logger.error(f"Error reading instruction file {md_file}: {e}")
 
@@ -241,10 +323,13 @@ def load_repo_context(repo_name: str, query: str = "") -> str:
             skill_name = skill_file.parent.name.lower()
             if not q or _keyword_hit(q, skill_name):
                 try:
-                    with open(skill_file, "r", encoding="utf-8") as f:
-                        context_parts.append(
-                            f"--- Local Skill ({skill_file.parent.name}) ---\n{f.read()}"
-                        )
+                    content = _read_capped_file(skill_file, max_chars=max_file_chars)
+                    _append_bounded_section(
+                        context_parts,
+                        f"--- Local Skill ({skill_file.parent.name}) ---",
+                        content,
+                        max_total_chars=max_total_chars,
+                    )
                 except Exception as e:
                     logger.error(f"Error reading {skill_file}: {e}")
 
@@ -254,10 +339,13 @@ def load_repo_context(repo_name: str, query: str = "") -> str:
         readme_md = repo_path / "README.md"
         if readme_md.exists():
             try:
-                with open(readme_md, "r", encoding="utf-8") as f:
-                    content = f.read()
-                    excerpt = content[:2000] + ("\n... (truncated)" if len(content) > 2000 else "")
-                    context_parts.append(f"--- README.md ---\n{excerpt}")
+                content = _read_capped_file(readme_md, max_chars=max_file_chars)
+                _append_bounded_section(
+                    context_parts,
+                    "--- README.md ---",
+                    content,
+                    max_total_chars=max_total_chars,
+                )
             except Exception as e:
                 logger.error(f"Error reading {readme_md}: {e}")
 
@@ -281,8 +369,13 @@ def load_repo_context(repo_name: str, query: str = "") -> str:
 
                     if should_load_org_skill:
                         try:
-                            with open(skill_file, "r", encoding="utf-8") as f:
-                                context_parts.append(f"--- Org-Wide Skill ({skill_dir.name}) ---\n{f.read()}")
+                            content = _read_capped_file(skill_file, max_chars=max_file_chars)
+                            _append_bounded_section(
+                                context_parts,
+                                f"--- Org-Wide Skill ({skill_dir.name}) ---",
+                                content,
+                                max_total_chars=max_total_chars,
+                            )
                         except Exception as e:
                             logger.error(f"Error reading org skill {skill_file}: {e}")
 

scripts/test_bounded_context.py

@@ -0,0 +1,183 @@
+"""
+Unit and smoke tests for bounded Markdown context loading in load_repo_context
+(addresses AOSSIE-Org/SkillBot#24).
+
+Follows the same convention as scripts/test_link_and_issue_features.py:
+can be run directly with python:
+    python scripts/test_bounded_context.py
+or with pytest:
+    pytest scripts/test_bounded_context.py
+"""
+
+import io
+import sys
+import tempfile
+from pathlib import Path
+
+if sys.platform == "win32":
+    sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8", errors="replace")
+
+bot_root = Path(__file__).resolve().parent.parent
+if str(bot_root) not in sys.path:
+    sys.path.insert(0, str(bot_root))
+
+import repo_router as rr
+
+PASSED = 0
+FAILED = 0
+
+
+def check(label: str, condition: bool, detail: str = ""):
+    global PASSED, FAILED
+    if condition:
+        PASSED += 1
+        print(f"  PASS  {label}")
+    else:
+        FAILED += 1
+        print(f"  FAIL  {label}  {('- ' + detail) if detail else ''}")
+
+
+def section_1_read_capped_file():
+    print("\n--- 1. Capped file read helper (_read_capped_file) ---")
+    with tempfile.TemporaryDirectory() as tmp_dir:
+        tmp_path = Path(tmp_dir)
+        short_file = tmp_path / "short.md"
+        short_file.write_text("Small content within limit", encoding="utf-8")
+
+        res_short = rr._read_capped_file(short_file, max_chars=100)
+        check("short file content intact", res_short == "Small content within limit")
+        check("no truncation marker on short file", rr.TRUNCATION_MARKER not in res_short)
+
+        large_file = tmp_path / "large.md"
+        large_file.write_text("X" * 500, encoding="utf-8")
+
+        res_large = rr._read_capped_file(large_file, max_chars=100)
+        check("large file truncated to max_chars prefix", res_large.startswith("X" * 100))
+        check("truncation marker appended", res_large.endswith(rr.TRUNCATION_MARKER))
+        check("length equals cap + marker length", len(res_large) == 100 + len(rr.TRUNCATION_MARKER))
+
+
+def section_2_append_bounded_section():
+    print("\n--- 2. Section budget enforcement (_append_bounded_section) ---")
+    parts = ["=== REPOSITORY: TestRepo ==="]
+    success = rr._append_bounded_section(
+        parts,
+        "--- Header ---",
+        "Short content",
+        max_total_chars=1000,
+    )
+    check("section fits returns True", success is True)
+    check("section added to parts", len(parts) == 2)
+    check("joined text includes header and content", "--- Header ---\nShort content" in "\n\n".join(parts))
+
+    # Partial fit where content is truncated to respect budget
+    initial_len = len(parts[0]) + len("\n\n") + len(parts[1])
+    budget = initial_len + 40
+    partial_success = rr._append_bounded_section(
+        parts,
+        "--- Second ---",
+        "Z" * 200,
+        max_total_chars=budget,
+    )
+    check("overflowing section returns False", partial_success is False)
+    joined = "\n\n".join(parts)
+    check("assembled length stays within budget", len(joined) <= budget)
+    check("truncation marker in assembled text", rr.TRUNCATION_MARKER in joined)
+
+    # Exhausted budget rejects further additions
+    exhausted_success = rr._append_bounded_section(
+        parts,
+        "--- Third ---",
+        "Will not fit",
+        max_total_chars=budget,
+    )
+    check("exhausted budget returns False", exhausted_success is False)
+    check("no parts added when budget exhausted", len(parts) == 3)
+
+
+def section_3_load_repo_context_mock():
+    print("\n--- 3. load_repo_context bounds on mock repository ---")
+    with tempfile.TemporaryDirectory() as tmp_dir:
+        repo_dir = Path(tmp_dir) / "MockRepo"
+        agent_dir = repo_dir / ".agent"
+        (agent_dir / "info").mkdir(parents=True)
+        (agent_dir / "core").mkdir(parents=True)
+        (agent_dir / "instructions").mkdir(parents=True)
+
+        (agent_dir / "info" / "operational-data.md").write_text("OPS_" + "O" * 4000, encoding="utf-8")
+        (agent_dir / "core" / "architecture.md").write_text("ARCH_" + "A" * 4000, encoding="utf-8")
+        (agent_dir / "instructions" / "setup.md").write_text("SETUP_" + "S" * 4000, encoding="utf-8")
+        (repo_dir / "README.md").write_text("README_" + "R" * 4000, encoding="utf-8")
+
+        orig_get_repo_path = rr.get_repo_path
+        rr.get_repo_path = lambda name: repo_dir if name == "MockRepo" else orig_get_repo_path(name)
+        try:
+            ctx = rr.load_repo_context("MockRepo", "setup", max_file_chars=150, max_total_chars=1200)
+            check("context total length strictly <= 1200", len(ctx) <= 1200)
+            check("repository header present", "=== REPOSITORY: MockRepo ===" in ctx)
+            check("operational data section present", "--- Operational Data" in ctx)
+            check("architecture section present", "--- Core Architecture" in ctx)
+            check("truncation marker present", rr.TRUNCATION_MARKER in ctx)
+
+            # Check individual section lengths
+            sections = ctx.split("\n\n")
+            for section in sections:
+                lines = section.split("\n", 1)
+                if len(lines) == 2 and lines[0].startswith("---"):
+                    body = lines[1]
+                    if rr.TRUNCATION_MARKER in body:
+                        before_marker = body.split(rr.TRUNCATION_MARKER)[0]
+                        check("file content capped at max_file_chars", len(before_marker) <= 150)
+                        break
+
+            # Very small total budget
+            ctx_tiny = rr.load_repo_context("MockRepo", "setup", max_file_chars=500, max_total_chars=250)
+            check("tiny budget context <= 250", len(ctx_tiny) <= 250)
+        finally:
+            rr.get_repo_path = orig_get_repo_path
+
+
+def section_4_load_repo_context_real_repo():
+    print("\n--- 4. Real repository context loading with defaults ---")
+    ctx = rr.load_repo_context("SocialShareButton", "How do I setup social share button?")
+    check("real repo context <= DEFAULT_MAX_TOTAL_CONTEXT_CHARS", len(ctx) <= rr.DEFAULT_MAX_TOTAL_CONTEXT_CHARS)
+    check("SocialShareButton header present", "=== REPOSITORY: SocialShareButton ===" in ctx)
+    check("operational data or architecture loaded", ("--- Operational Data" in ctx) or ("--- Core Architecture" in ctx))
+
+
+# Pytest adapters
+def test_read_capped_file():
+    section_1_read_capped_file()
+    assert FAILED == 0
+
+
+def test_append_bounded_section():
+    section_2_append_bounded_section()
+    assert FAILED == 0
+
+
+def test_load_repo_context_mock():
+    section_3_load_repo_context_mock()
+    assert FAILED == 0
+
+
+def test_load_repo_context_real_repo():
+    section_4_load_repo_context_real_repo()
+    assert FAILED == 0
+
+
+def run_all():
+    global PASSED, FAILED
+    PASSED = 0
+    FAILED = 0
+    section_1_read_capped_file()
+    section_2_append_bounded_section()
+    section_3_load_repo_context_mock()
+    section_4_load_repo_context_real_repo()
+    print(f"\n=== {PASSED} passed, {FAILED} failed ===")
+    return FAILED == 0
+
+
+if __name__ == "__main__":
+    success = run_all()
+    sys.exit(0 if success else 1)