harness: split scale test into same-session and multi-user concurrency

Author: Muhra95

harness/config.py

@@ -100,3 +100,26 @@ def timeout_state(self) -> dict:
             "latency_ewma_ms": round(self._latency_ewma_ms, 1),
             "latency_samples": self._latency_samples,
         }
+
+    def clone_for_user(self, user_id: str) -> "BenchConfig":
+        """Create a per-user clone preserving harness behavior settings."""
+        return BenchConfig(
+            base_url=self.base_url,
+            token=self.token,
+            user_id=user_id,
+            timeout=self.timeout,
+            timeout_dynamic=self.timeout_dynamic,
+            timeout_floor_secs=self.timeout_floor_secs,
+            timeout_ceiling_secs=self.timeout_ceiling_secs,
+            timeout_multiplier=self.timeout_multiplier,
+            timeout_grace_secs=self.timeout_grace_secs,
+            chat_endpoint=self.chat_endpoint,
+            health_endpoint=self.health_endpoint,
+            ready_endpoint=self.ready_endpoint,
+            diagnostics_endpoint=self.diagnostics_endpoint,
+            metrics_endpoint=self.metrics_endpoint,
+            memory_sample_size=self.memory_sample_size,
+            scale_concurrency=self.scale_concurrency,
+            latency_requests=self.latency_requests,
+            schedule_wait_secs=self.schedule_wait_secs,
+        )

harness/dim7_scale.py

@@ -21,8 +21,56 @@ def _timed_chat(config: BenchConfig, message: str) -> dict:
     }
 
 
+def _run_concurrency_scenario(configs: list[BenchConfig], message_prefix: str) -> tuple[list[float], int, float]:
+    """Run one concurrency scenario and return (latencies_ms, errors, wall_time_ms)."""
+    latencies: list[float] = []
+    errors = 0
+    start_all = time.monotonic()
+
+    with concurrent.futures.ThreadPoolExecutor(max_workers=len(configs)) as pool:
+        futures = []
+        for i, cfg in enumerate(configs):
+            futures.append(
+                pool.submit(
+                    _timed_chat,
+                    cfg,
+                    f"{message_prefix} {i}: what is {i}+{i}?",
+                )
+            )
+        for f in concurrent.futures.as_completed(futures):
+            r = f.result()
+            if r["error"]:
+                errors += 1
+            else:
+                latencies.append(r["latency_ms"])
+
+    wall_time_ms = (time.monotonic() - start_all) * 1000
+    return latencies, errors, wall_time_ms
+
+
+def _latency_stats(latencies: list[float]) -> dict:
+    if not latencies:
+        return {}
+    ordered = sorted(latencies)
+    return {
+        "p50_ms": round(statistics.median(ordered), 1),
+        "p95_ms": round(
+            ordered[int(len(ordered) * 0.95)] if len(ordered) >= 5 else max(ordered),
+            1,
+        ),
+        "p99_ms": round(
+            ordered[int(len(ordered) * 0.99)] if len(ordered) >= 10 else max(ordered),
+            1,
+        ),
+    }
+
+
 def run(config: BenchConfig) -> dict:
-    """Test scale under concurrent load."""
+    """Test scale in three modes:
+    1) same-session contention (single user)
+    2) multi-user parallel (one request per user)
+    3) transport/pool metric snapshot
+    """
     results = {}
     concurrency = config.scale_concurrency
 
@@ -40,47 +88,50 @@ def run(config: BenchConfig) -> dict:
         results["baseline_p50_ms"] = None
         results["baseline_error"] = "All baseline requests failed"
 
-    # Phase 2: Concurrent load
-    latencies = []
-    errors = 0
-    start_all = time.monotonic()
-
-    with concurrent.futures.ThreadPoolExecutor(max_workers=concurrency) as pool:
-        futures = []
-        for i in range(concurrency):
-            futures.append(
-                pool.submit(
-                    _timed_chat,
-                    config,
-                    f"Concurrent test request {i}: what is {i}+{i}?",
-                )
-            )
-        for f in concurrent.futures.as_completed(futures):
-            r = f.result()
-            if r["error"]:
-                errors += 1
-            else:
-                latencies.append(r["latency_ms"])
+    # Phase 2a: Same-user same-session contention (diagnostic only; not primary scale metric)
+    same_session_cfgs = [config for _ in range(concurrency)]
+    same_latencies, same_errors, same_wall_ms = _run_concurrency_scenario(
+        same_session_cfgs,
+        "Same-session contention request",
+    )
+    same_stats = _latency_stats(same_latencies)
+    results["same_session"] = {
+        "requests": concurrency,
+        "errors": same_errors,
+        "success": len(same_latencies),
+        "wall_time_ms": round(same_wall_ms, 1),
+        "p50_ms": same_stats.get("p50_ms"),
+        "p95_ms": same_stats.get("p95_ms"),
+        "p99_ms": same_stats.get("p99_ms"),
+    }
 
-    wall_time_ms = (time.monotonic() - start_all) * 1000
+    # Phase 2b: Multi-user parallel (primary scale metric)
+    try:
+        base_user_num = int(config.user_id)
+        user_ids = [str(base_user_num + i) for i in range(concurrency)]
+    except ValueError:
+        user_ids = [f"{config.user_id}-bench-{i}" for i in range(concurrency)]
+    multi_cfgs = [config.clone_for_user(uid) for uid in user_ids]
+    multi_latencies, multi_errors, multi_wall_ms = _run_concurrency_scenario(
+        multi_cfgs,
+        "Multi-user concurrent request",
+    )
+    multi_stats = _latency_stats(multi_latencies)
+    results["multi_user"] = {
+        "requests": concurrency,
+        "errors": multi_errors,
+        "success": len(multi_latencies),
+        "wall_time_ms": round(multi_wall_ms, 1),
+        "p50_ms": multi_stats.get("p50_ms"),
+        "p95_ms": multi_stats.get("p95_ms"),
+        "p99_ms": multi_stats.get("p99_ms"),
+    }
 
-    results["concurrent_requests"] = concurrency
-    results["concurrent_errors"] = errors
-    results["concurrent_success"] = len(latencies)
-    results["wall_time_ms"] = round(wall_time_ms, 1)
-
-    if latencies:
-        latencies.sort()
-        results["concurrent_p50_ms"] = round(statistics.median(latencies), 1)
-        results["concurrent_p95_ms"] = (
-            round(latencies[int(len(latencies) * 0.95)], 1)
-            if len(latencies) >= 5
-            else round(max(latencies), 1)
-        )
-        results["concurrent_p99_ms"] = (
-            round(latencies[int(len(latencies) * 0.99)], 1)
-            if len(latencies) >= 10
-            else round(max(latencies), 1)
+    same_p95 = results["same_session"].get("p95_ms")
+    multi_p95 = results["multi_user"].get("p95_ms")
+    if same_p95 and multi_p95 and multi_p95 > 0:
+        results["contention_ratio_same_session_over_multi_user"] = round(
+            same_p95 / multi_p95, 2
         )
 
     # Phase 3: Parse metrics for pool/transport stats
@@ -96,23 +147,22 @@ def run(config: BenchConfig) -> dict:
                     results["metrics_snapshot"][parts[0]] = parts[1]
 
     # Score calculation
-    success_rate = len(latencies) / concurrency if concurrency > 0 else 0
-    p95 = results.get("concurrent_p95_ms", 60000)
+    multi_success_rate = len(multi_latencies) / concurrency if concurrency > 0 else 0
+    multi_p95 = results["multi_user"].get("p95_ms") or 60000
 
-    # inverse_rss (projected — can't measure from outside), inverse_p95, max_users, horizontal, inverse_cost
-    p95_score = max(0, 100 - (p95 / 1000))  # 0ms = 100, 100s = 0
-    success_score = success_rate * 100
+    # Primary measured scale signal should use multi-user throughput, not same-session contention.
+    multi_p95_score = max(0, 100 - (multi_p95 / 1000))  # 0ms = 100, 100s = 0
+    verified_score = multi_p95_score * multi_success_rate
+
+    # Projection still includes unmeasured dimensions.
     projected_score = (
-        0.70 * 0.25  # inverse_rss: projected (2.6MB binary, ~43MB test RSS)
-        + p95_score / 100 * 0.20
-        + 0.70 * 0.25  # max_users: projected (~1000/instance)
-        + 0.80 * 0.20  # horizontal: projected (tenant lock + Postgres)
+        0.70 * 0.25  # inverse_rss: projected
+        + (multi_p95_score / 100) * 0.20
+        + 0.70 * 0.25  # max_users: projected
+        + 0.80 * 0.20  # horizontal: projected
         + 0.60 * 0.10  # inverse_cost: projected
     ) * 100
-
-    # Adjust based on actual success rate
-    projected_score = projected_score * success_rate
-    verified_score = p95_score * success_rate
+    projected_score = projected_score * multi_success_rate
 
     results["score"] = round(min(100, projected_score), 1)
     results["verified_score"] = round(min(100, verified_score), 1)

harness/runner.py

@@ -114,6 +114,12 @@ def main():
         action="store_true",
         help="Skip the 3-minute schedule execution wait",
     )
+    parser.add_argument(
+        "--scale-concurrency",
+        type=int,
+        default=20,
+        help="Concurrent requests for scale dimension (default: 20)",
+    )
     args = parser.parse_args()
 
     config = BenchConfig(
@@ -126,6 +132,7 @@ def main():
         timeout_ceiling_secs=args.timeout_ceiling,
         timeout_multiplier=args.timeout_multiplier,
         timeout_grace_secs=args.timeout_grace,
+        scale_concurrency=args.scale_concurrency,
         schedule_wait_secs=0 if args.skip_schedule_wait else 180,
     )