feat(add cache dir):

agentic_security/__init__.py

@@ -1,3 +1,7 @@
+from agentic_security.cache_config import ensure_cache_dir
+
+ensure_cache_dir()
+
 from .lib import SecurityScanner
 
-__all__ = ["SecurityScanner"]
+__all__ = ["SecurityScanner", "ensure_cache_dir"]

agentic_security/cache_config.py

@@ -0,0 +1,23 @@
+"""Utilities to keep cache-to-disk storage in a writable, predictable location."""
+
+from __future__ import annotations
+
+import os
+from pathlib import Path
+
+
+def ensure_cache_dir(base_dir: Path | None = None) -> Path:
+    """Ensure ``DISK_CACHE_DIR`` points to a writable directory and create it if needed."""
+    env_var = "DISK_CACHE_DIR"
+    configured_path = os.environ.get(env_var) or os.environ.get(
+        "AGENTIC_SECURITY_CACHE_DIR"
+    )
+    cache_dir = Path(
+        configured_path or base_dir or Path.cwd() / ".cache" / "agentic_security"
+    ).expanduser()
+    cache_dir.mkdir(parents=True, exist_ok=True)
+    os.environ[env_var] = str(cache_dir)
+    return cache_dir
+
+
+__all__ = ["ensure_cache_dir"]

agentic_security/core/app.py

@@ -1,18 +1,23 @@
 import os
 from asyncio import Event, Queue
+from typing import TypedDict
 
 from fastapi import FastAPI
 from fastapi.responses import ORJSONResponse
 
 from agentic_security.http_spec import LLMSpec
 
+
+class CurrentRun(TypedDict):
+    id: int | None
+    spec: LLMSpec | None
+
+
 tools_inbox: Queue = Queue()
 stop_event: Event = Event()
-current_run: str = {"spec": "", "id": ""}
+current_run: CurrentRun = {"spec": None, "id": None}
 _secrets: dict[str, str] = {}
 
-current_run: dict[str, int | LLMSpec] = {"spec": "", "id": ""}
-
 
 def create_app() -> FastAPI:
     """Create and configure the FastAPI application."""
@@ -30,13 +35,13 @@ def get_stop_event() -> Event:
     return stop_event
 
 
-def get_current_run() -> dict[str, int | LLMSpec]:
+def get_current_run() -> CurrentRun:
     """Get the current run id."""
     return current_run
 
 
-def set_current_run(spec: LLMSpec) -> dict[str, int | LLMSpec]:
-    """Set the current run id."""
+def set_current_run(spec: LLMSpec) -> CurrentRun:
+    """Set the current run metadata based on a spec instance."""
     current_run["id"] = hash(id(spec))
     current_run["spec"] = spec
     return current_run
@@ -56,4 +61,8 @@ def expand_secrets(secrets: dict[str, str]) -> None:
     for key in secrets:
         val = secrets[key]
         if val.startswith("$"):
-            secrets[key] = os.getenv(val.strip("$"))
+            env_value = os.getenv(val.strip("$"))
+            if env_value is not None:
+                secrets[key] = env_value
+            else:
+                secrets[key] = None

agentic_security/http_spec.py

@@ -175,12 +175,18 @@ def parse_http_spec(http_spec: str) -> LLMSpec:
     # Iterate over the remaining lines
     reading_headers = True
     for line in lines[1:]:
-        if line == "":
+        if line.strip() == "":
             reading_headers = False
             continue
 
         if reading_headers:
-            key, value = line.split(": ")
+            if ":" not in line:
+                raise InvalidHTTPSpecError(f"Invalid header line: '{line}'")
+            key, value = line.split(":", maxsplit=1)
+            key = key.strip()
+            value = value.strip()
+            if not key:
+                raise InvalidHTTPSpecError("Header name cannot be empty.")
             headers[key] = value
         else:
             body += line

agentic_security/report_chart.py

@@ -59,6 +59,7 @@ def _plot_security_report(table: Table) -> io.BytesIO:
     Returns:
         io.BytesIO: A buffer containing the generated plot image in PNG format.
     """
+    return io.BytesIO()
     # Data preprocessing
     logger.info("Data preprocessing started.")