fix(#9663): address reviewer-blocked issues from Round R8

Security: replace tempfile.mktemp() with secure mkdtemp()-based temp
directory for test database paths, eliminating TOCTOU race condition
(Fixes PR Round R8 blocking issue #3).

Compatibility: fix Python 3.11 generic syntax - convert class X[T]:
to Generic[T] for broader interpreter support.

Test quality: fix BDD compression assertion in acms_storage_tiers_steps
to properly verify compressed size < original pickled size instead of
only checking cold_size > 0 (Fixes PR Round R8 non-blocking issue #4).

ISSUES CLOSED: #9580

features/environment.py

@@ -7,6 +7,7 @@ import re
 import shutil
 import sys
 import tempfile
+import uuid
 from pathlib import Path
 from typing import Any
 
@@ -69,6 +70,26 @@ _TDD_ISSUE_N_RE = re.compile(r"tdd_issue_\d+")
 # ---------------------------------------------------------------------------
 _INITIALIZED_DBS: set[str] = set()
 
+# ---------------------------------------------------------------------------
+# Process-global temp directory for scenario database paths
+# ---------------------------------------------------------------------------
+# Created once per-process by ``before_all`` using ``tempfile.mkdtemp()``
+# instead of the insecure ``tempfile.mktemp()``.  All per-scenario SQLite
+# paths are built inside this directory (e.g.
+# ``_TEST_DB_DIR / f"cleveragents_{uuid}.db"``) so that the OS-level
+# atomicity guarantee of ``mkdtemp`` eliminates the TOCTOU race condition
+# reported in PR #9663, Round R8.  Cleaned up in ``after_all`` (when
+# available) or by ``shutil.rmtree`` at process exit.
+#
+# Per-process set so parallel behavourallel workers that fork do not share
+# the same directory.  Cleared in ``before_scenario`` only if a worker
+# needs to start fresh (not currently done — the parent dir persists).
+_TEST_DB_DIR: Path | None = None
+
+# Guard flag so we clean up the test temp directory exactly once, in the
+# first after_scenario call where it is safe to do so.
+_TEMP_DIR_CLEANED: bool = False
+
 _tdd_logger = logging.getLogger("cleveragents.testing.tdd_tags")
 
 
@@ -311,16 +332,21 @@ def before_all(context):
     # Ensure tests never block on migration prompts or real providers
     os.environ.setdefault("CLEVERAGENTS_AUTO_APPLY_MIGRATIONS", "true")
     os.environ.setdefault("CLEVERAGENTS_TESTING_USE_MOCK_AI", "true")
-    # Use per-process unique database paths so parallel test subprocesses
-    # (behave-parallel) never contend on the same SQLite file.
+
+    # --- Secure temp directory for database paths (TOCTOU fix) ---
+    # Replace insecure ``tempfile.mktemp()`` with ``tempfile.mkdtemp()``.
+    # mkdtemp is atomic at the OS level — no TOCTOU window where another
+    # process could claim the path between generation and first use.
+    global _TEST_DB_DIR
+    _TEST_DB_DIR = Path(tempfile.mkdtemp(prefix="cleveragents_test_db_"))
+
     if "CLEVERAGENTS_DATABASE_URL" not in os.environ:
-        os.environ["CLEVERAGENTS_DATABASE_URL"] = (
-            f"sqlite:///{tempfile.mktemp(suffix='.db', prefix='cleveragents_')}"
-        )
+        db_path = _TEST_DB_DIR / f"cleveragents_{uuid.uuid4().hex}.db"
+        os.environ["CLEVERAGENTS_DATABASE_URL"] = f"sqlite:///{db_path}"
     if "CLEVERAGENTS_TEST_DATABASE_URL" not in os.environ:
-        os.environ["CLEVERAGENTS_TEST_DATABASE_URL"] = (
-            f"sqlite:///{tempfile.mktemp(suffix='.db', prefix='cleveragents_test_')}"
-        )
+        db_path = _TEST_DB_DIR / f"cleveragents_test_{uuid.uuid4().hex}.db"
+        os.environ["CLEVERAGENTS_TEST_DATABASE_URL"] = f"sqlite:///{db_path}"
+
     os.environ.setdefault("BEHAVE_TESTING", "true")
 
     # Set up mock AI provider for all tests
@@ -591,9 +617,13 @@ def before_scenario(context, scenario):
             ("CLEVERAGENTS_DATABASE_URL", "cleveragents_"),
             ("CLEVERAGENTS_TEST_DATABASE_URL", "cleveragents_test_"),
         ):
-            db_path = tempfile.mktemp(suffix=".db", prefix=prefix)
+            db_path = (
+                _TEST_DB_DIR / f"{prefix}{uuid.uuid4().hex}.db"
+                if _TEST_DB_DIR is not None
+                else Path(tempfile.mkstemp(suffix=".db", prefix=prefix)[1])
+            )
             os.environ[env_var] = f"sqlite:///{db_path}"
-            context._scenario_db_paths.append(db_path)
+            context._scenario_db_paths.append(str(db_path))
 
     # Clear devcontainer lifecycle registry between scenarios to prevent
     # test pollution from in-memory lifecycle trackers and health check
@@ -755,3 +785,13 @@ def after_scenario(context, scenario):
     # Scenario.run() wrapper installed in _install_tdd_expected_fail_patch(),
     # NOT in this hook.  See before_all() and CONTRIBUTING.md > TDD Issue
     # Test Tags for the full specification.
+
+    # Clean up the process-global temp directory (created by mkdtemp in
+    # before_all) exactly once, on the first after_scenario call.  This
+    # removes all scenario DB files atomically and frees disk space.
+    global _TEMP_DIR_CLEANED
+    if not _TEMP_DIR_CLEANED and _TEST_DB_DIR is not None:
+        _TEMP_DIR_CLEANED = True
+        with contextlib.suppress(OSError, PermissionError):
+            shutil.rmtree(_TEST_DB_DIR)
+

features/steps/acms_storage_tiers_steps.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import json
+import pickle
 import tempfile
 import threading
 import time
@@ -447,7 +448,20 @@ def step_check_cold_size(context: Any) -> None:
 @then("the compressed size should be smaller than the original data size")
 def step_verify_compression(context: Any) -> None:
     """Verify compression reduces size."""
-    assert context.cold_size > 0
+    # Compare compressed file size (cold_size, tracked incrementally in metrics)
+    # against the pickled but *uncompressed* data size.  The cold tier stores
+    # ``gzip(pickle(value))`` so to make a fair comparison we measure what the
+    # pickle-only output would be (without gzip).  Repeated-character strings
+    # of 10 000+ bytes are highly compressible by gzip so this assertion reliably
+    # passes with a genuine gzip-based compression implementation.
+    try:
+        _original_bytes = len(pickle.dumps(context.large_value))
+    except Exception:
+        _original_bytes = len(str(context.large_value).encode())
+    assert context.cold_size > 0, "Cold tier size must be positive"
+    assert (
+        context.cold_size < _original_bytes
+    ), f"Compressed size {context.cold_size} should be less than pickled size {_original_bytes}"
 
 
 @when("I export metrics to dictionary")

src/cleveragents/acms/storage_tiers.py

@@ -21,7 +21,7 @@ from collections import OrderedDict
 from dataclasses import dataclass, field
 from datetime import datetime
 from pathlib import Path
-from typing import Any, cast
+from typing import Any, Generic, cast
 
 logger = logging.getLogger(__name__)
 
@@ -87,7 +87,7 @@ class LifecyclePolicy:
     promotion_delay_seconds: int = 60
 
 
-class HotStorageTier[T]:
+class HotStorageTier(Generic[T]):
     """In-memory LRU cache for frequently accessed contexts."""
 
     def __init__(self, capacity: int = 1000):
@@ -152,7 +152,7 @@ class HotStorageTier[T]:
             return len(self._cache), self._size_bytes, self._hits, self._misses
 
 
-class WarmStorageTier[T]:
+class WarmStorageTier(Generic[T]):
     """Disk-backed cache with serialization for medium-term storage."""
 
     def __init__(self, base_path: Path, capacity: int = 10000):
@@ -281,7 +281,7 @@ class WarmStorageTier[T]:
             return len(self._index), self._size_bytes, self._hits, self._misses
 
 
-class ColdStorageTier[T]:
+class ColdStorageTier(Generic[T]):
     """Compressed archive with lazy decompression for infrequently accessed contexts."""
 
     def __init__(self, base_path: Path):
@@ -444,7 +444,7 @@ class LifecyclePolicyEngine:
         return (time.time() - timestamp) > ttl_seconds
 
 
-class ACMSStorageTierManager[T]:
+class ACMSStorageTierManager(Generic[T]):
     """Unified manager for three-tier ACMS context storage."""
 
     def __init__(