⚡ Bolt: Cache compiled regex patterns in APIRouter

.jules/bolt.md

@@ -0,0 +1,4 @@
+# 2023-11-20 - [Compile Regex Patterns Once in API Routing]
+
+**Learning:** Recompiling regular expressions on every incoming request for path parameter matching causes an unnecessary performance bottleneck. The `_path_to_regex` method in `APIRouter` was rebuilding and recompiling its parameter extraction regex for each endpoint evaluation loop.
+**Action:** Always cache or memoize compiled regex patterns in high-traffic routing logic or request parsers, either via `functools.lru_cache` or by storing them in a dictionary keyed by the endpoint path.

scripts/agents/hermes/evaluate_orchestrators.py

@@ -281,7 +281,7 @@ def extract_json_from_response(content: str) -> dict:
 def assess_script(client, script_info: dict, source_code: str) -> dict:
     """Use Hermes to assess a script based on stdout, stderr, and source code."""
     script_name = script_info["path"].name
-    
+
     prompt = f"""You are a senior principal engineer performing a code review and architectural assessment.
 
 Evaluate the following Python orchestration script based on the Codomyrmex "Thin Orchestrator" pattern.
@@ -333,7 +333,7 @@ def assess_script(client, script_info: dict, source_code: str) -> dict:
         response = client.chat_session(prompt=prompt)
         if response.is_success():
             eval_data = extract_json_from_response(response.content)
-            
+
             print_success(f"=== Hermes Assessment for {script_name} ===")
             print_info(f"  Adheres to pattern: {eval_data.get('adherence_assessment', {}).get('adheres', False)}")
             print_success("=" * 60)
@@ -387,7 +387,7 @@ def main() -> int:
     trace_path: Optional[Path] = (_REPO_ROOT / trace_file_rel) if trace_file_rel else None
     if trace_path:
         trace_path.parent.mkdir(parents=True, exist_ok=True)
-    
+
     # 1. Boot up Hermes Client
     if not args.dry_run:
         try:
@@ -403,7 +403,7 @@ def main() -> int:
     else:
         client = None
         print_info("  [DRY RUN] Hermes assessment will be skipped.")
-    
+
     print_info("═" * 60)
     print_info(f"  Hermes Script Evaluator — Target: scripts/{args.target}")
     print_info(f"  Saving outputs to: {output_path}")
@@ -480,15 +480,15 @@ def main() -> int:
                 script_path,
                 timeout=evaluator_cfg.get("script_timeout", 30),
             )
-        
+
         # Read source code
         try:
             with open(script_path, "r", encoding="utf-8") as f:
                 source_code = f.read()
         except Exception as e:
             print_error(f"  Could not read source code for {script_path.name}: {e}")
             continue
-        
+
         # Assess (skip in dry-run mode)
         if args.dry_run:
             print_info(f"  [DRY RUN] Would assess {script_path.name} with Hermes.")
@@ -530,25 +530,25 @@ def main() -> int:
             f.write(f"# Evaluator Orchestrations Report\n")
             f.write(f"Target: `scripts/{args.target}`\n")
             f.write(f"Generated: {datetime.now().isoformat()}\n\n")
-            
+
             for script_name, data in all_evaluations.items():
                 f.write(f"## Script: `{script_name}`\n")
-                
+
                 adherence = data.get("adherence_assessment", {})
                 pass_fail = "✅ STRICT ADHERENCE" if adherence.get("adheres", False) else "❌ NON-COMPLIANT"
                 f.write(f"**Pattern Adherence**: {pass_fail}\n\n")
                 f.write(f"> {adherence.get('reasoning', 'No reasoning provided.')}\n\n")
-                
+
                 f.write("### Technical Debt Identified:\n")
                 for debt in data.get("technical_debt", []):
                     f.write(f"- {debt}\n")
-                
+
                 f.write("\n### Underlying Method Improvements Required:\n")
                 for imp in data.get("underlying_improvements", []):
                     f.write(f"- {imp}\n")
-                
+
                 f.write("\n---\n\n")
-                
+
         print_success(f"Successfully compiled overall markdown report to: {report_file_path}")
     except Exception as e:
         print_error(f"Failed to save overall evaluation report: {e}")

src/codomyrmex/agents/hermes/hermes_client.py

@@ -286,7 +286,7 @@ def chat_session(self, prompt: str, session_id: str | None = None) -> AgentRespo
             history_text = ""
             for msg in session.messages[:-1]:  # exclude the current user prompt
                 history_text += f"[{msg['role'].upper()}]\n{msg['content']}\n\n"
-            
+
             if history_text:
                 full_prompt = (
                     f"Previous Conversation:\n{history_text}"
@@ -302,7 +302,7 @@ def chat_session(self, prompt: str, session_id: str | None = None) -> AgentRespo
             if response.is_success():
                 session.add_message("assistant", response.content)
                 store.save(session)
-                
+
             response.metadata["session_id"] = session.session_id
             return response
 

src/codomyrmex/api/standardization/rest_api.py

@@ -154,6 +154,7 @@ def __init__(self, prefix: str = ""):
         self.endpoints: dict[str, APIEndpoint] = {}
         self.sub_routers: list[APIRouter] = []
         self.middleware: list[Callable[[APIRequest], APIResponse | None]] = []
+        self._compiled_regexes: dict[str, tuple] = {}
 
     def add_endpoint(self, endpoint: APIEndpoint) -> None:
         """
@@ -291,6 +292,10 @@ def _path_to_regex(self, path: str) -> tuple:
         Returns:
             Tuple of (compiled_regex, parameter_names)
         """
+        # Return cached regex if available
+        if path in self._compiled_regexes:
+            return self._compiled_regexes[path]
+
         # Replace {param} with named capture groups
         param_pattern = re.compile(r"\{([^}]+)\}")
         param_names = param_pattern.findall(path)
@@ -299,7 +304,10 @@ def _path_to_regex(self, path: str) -> tuple:
         regex_pattern = param_pattern.sub(r"(?P<\1>[^/]+)", path)
         regex_pattern = f"^{regex_pattern}$"
 
-        return re.compile(regex_pattern), param_names
+        compiled = re.compile(regex_pattern)
+        self._compiled_regexes[path] = (compiled, param_names)
+
+        return compiled, param_names
 
     def get_all_endpoints(self) -> list[APIEndpoint]:
         """
@@ -457,7 +465,11 @@ def handle_request(
 
             processing_time = time.time() - start_time
             logger.info(
-                "Request completed: %s %s -> %s (%.3fs)", method, path, response.status_code.value, processing_time
+                "Request completed: %s %s -> %s (%.3fs)",
+                method,
+                path,
+                response.status_code.value,
+                processing_time,
             )
 
             return response

src/codomyrmex/tests/integration/calendar_integration/test_mcp_tools.py

@@ -90,10 +90,10 @@ def test_calendar_attendee_injection(monkeypatch):
     """Verify that _DEFAULT_ATTENDEE is always injected as an attendee."""
     test_attendee = "test@example.com"
     monkeypatch.setenv("CODOMYRMEX_CALENDAR_ATTENDEE", test_attendee)
-    
-    # Needs to match the env var or use monkeypatch's effect on the module. 
+
+    # Needs to match the env var or use monkeypatch's effect on the module.
     # But wait, mcp_tools reads os.environ at IMPORT time!
-    # So monkeypatching os.environ during the test execution 
+    # So monkeypatching os.environ during the test execution
     # won't change mcp_tools._DEFAULT_ATTENDEE because it was evaluated at import.
     # We must patch mcp_tools._DEFAULT_ATTENDEE directly!
     import codomyrmex.calendar_integration.mcp_tools as mcp_tools_mod

src/codomyrmex/tests/unit/agents/test_agents_hermes_client.py

@@ -132,7 +132,7 @@ class TestHermesClientSessionIntegration:
     def test_chat_session_new_and_continue(self, tmp_path) -> None:
         """Test creating a new session and appending to it."""
         db_path = tmp_path / "test_sessions.db"
-        
+
         # Use echo to simulate a fast, mock-free successful execution
         client = HermesClient(config={
             "hermes_command": "echo",
@@ -170,7 +170,7 @@ def test_mcp_session_lifecycle(self, monkeypatch, tmp_path) -> None:
         # Patch _get_client to inject our test config
         from codomyrmex.agents.hermes import mcp_tools
         from codomyrmex.agents.hermes.hermes_client import HermesClient
-        
+
         def mock_get_client(**kwargs):
             return HermesClient(config={
                 "hermes_command": "echo",
@@ -179,9 +179,9 @@ def mock_get_client(**kwargs):
                 "hermes_model": kwargs.get("model", "hermes3"),
                 "hermes_timeout": kwargs.get("timeout", 120),
             })
-            
+
         monkeypatch.setattr(mcp_tools, "_get_client", mock_get_client)
-        
+
         from codomyrmex.agents.hermes.mcp_tools import (
             hermes_chat_session,
             hermes_session_clear,

src/codomyrmex/tests/unit/audio/test_mcp_tools.py

@@ -102,7 +102,7 @@ def test_audio_batch_transcribe_invalid_files(
     result = audio_batch_transcribe(
         ["/invalid/1.wav", "/invalid/2.wav"], str(temp_output_dir), model_size="tiny"
     )
-    
+
     if WHISPER_AVAILABLE:
         assert result["success"] is True
         assert result["result"]["processed"] == 0
@@ -118,7 +118,7 @@ def test_audio_transcribe_invalid_format(dummy_wav_file: Path) -> None:
     """Test transcription with invalid format (e.g. by changing suffix)."""
     bad_format_path = dummy_wav_file.with_suffix(".txt")
     dummy_wav_file.rename(bad_format_path)
-    
+
     result = audio_transcribe(str(bad_format_path))
     assert result["success"] is False
     assert "AudioFormatError" in result["error"]["type"]

src/codomyrmex/tests/unit/auth/test_google_auth.py

@@ -20,13 +20,13 @@ def test_initialization(self):
         with tempfile.NamedTemporaryFile() as tmp:
             client_secrets = tmp.name
             token_cache = tempfile.mktemp()
-            
+
             auth = GoogleAuthenticator(
                 client_secrets_file=client_secrets,
                 token_cache_file=token_cache,
                 scopes=["https://www.googleapis.com/auth/calendar"]
             )
-            
+
             assert auth.client_secrets_file == client_secrets
             assert auth.token_file == token_cache
             assert auth.scopes == ["https://www.googleapis.com/auth/calendar"]

src/codomyrmex/tests/unit/bio_simulation/test_bio_simulation.py

@@ -149,11 +149,11 @@ def test_colony_idle_to_foraging():
     colony = Colony(population=100)
     for ant in colony.ants:
         ant.state = AntState.IDLE
-    
+
     # Run a few ticks, some should switch
     for _ in range(10):
         colony._step_tick()
-    
+
     states = [ant.state for ant in colony.ants]
     assert AntState.FORAGING in states