From 0a74052f72db5c371a9105c8ded46cdc5e55eafe Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Tue, 21 Jul 2026 01:01:41 +0000 Subject: [PATCH 1/7] Initial plan From 619e122962d9329b90a9959dff4fed969d8fb60a Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Tue, 21 Jul 2026 01:04:03 +0000 Subject: [PATCH 2/7] Track per-future observed CPU/GPU metrics --- tests/test_runtime_sqlalchemy.py | 30 +++++++++++++++ ventis/controller/local_controller.py | 55 ++++++++++++++++++++++++++- ventis/controller/utils/sqlalchemy.py | 26 ++++++++++--- 3 files changed, 104 insertions(+), 7 deletions(-) diff --git a/tests/test_runtime_sqlalchemy.py b/tests/test_runtime_sqlalchemy.py index c0a80f1..cffb11a 100644 --- a/tests/test_runtime_sqlalchemy.py +++ b/tests/test_runtime_sqlalchemy.py @@ -94,6 +94,36 @@ def test_pull_and_upsert(self): for value in row: self.assertNotIn(value, (None, "")) + def test_observed_metrics_override_config_allocations(self): + redis = _FakeRedis( + { + "future:xyz": { + "id": "xyz", + "request_id": "req2", + "agent": "AgentB", + "created_at": "10.0", + "finished_at": "12.0", + "execution_time": "1.25", + "cpu_resource": "37.5", + "gpu_resource": "256", + }, + "request:req2:workflow": "wf2", + } + ) + + rows = sqlmod.pull_data(redis) + sqlmod.send_data(rows, {"AgentB": {"cpu": 8, "gpu": 2}}, redis) + with sqlmod._get_engine("").connect() as conn: + row = conn.execute( + text( + "SELECT execution_time, cpu_resource, gpu_resource " + "FROM runtime_information WHERE future_id='xyz'" + ) + ).fetchone() + self.assertEqual(row[0], 1.25) + self.assertEqual(row[1], 37.5) + self.assertEqual(row[2], 256.0) + if __name__ == "__main__": unittest.main() diff --git a/ventis/controller/local_controller.py b/ventis/controller/local_controller.py index 1dd6895..adc97ef 100644 --- a/ventis/controller/local_controller.py +++ b/ventis/controller/local_controller.py @@ -6,6 +6,7 @@ import logging import os import random +import subprocess import sys import time import importlib.util @@ -40,6 +41,40 @@ POLICY_RULES_KEY = "policy:rules" +def _read_gpu_memory_for_pid(pid): + """Read total GPU memory (MiB) used by a process pid via nvidia-smi.""" + try: + result = subprocess.run( + [ + "nvidia-smi", + "--query-compute-apps=pid,used_memory", + "--format=csv,noheader,nounits", + ], + capture_output=True, + text=True, + timeout=2, + check=False, + ) + except (FileNotFoundError, subprocess.SubprocessError): + return 0.0 + + if result.returncode != 0: + return 0.0 + + total = 0.0 + for line in result.stdout.splitlines(): + line = line.strip() + if not line: + continue + try: + pid_value, mem_value = [part.strip() for part in line.split(",", 1)] + if int(pid_value) == int(pid): + total += float(mem_value) + except (TypeError, ValueError): + continue + return total + + class LocalController(object): """Manages the gRPC frontend and processes incoming requests from the queue.""" @@ -412,6 +447,9 @@ def _execute_locally( self, service, function, args, future_id, origin=None, request_id=None ): """Execute a request on the local agent and write the result to Redis.""" + wall_start = time.time() + thread_cpu_start = time.thread_time() + # Propagate the request_id context into this worker thread if request_id: ventis_context.set_request_id(request_id) @@ -461,7 +499,22 @@ def _execute_locally( if origin and origin != self._my_endpoint: self._send_result_callback(origin, future_id, f"Execution failed: {e}") - self.redis.hset(f"future:{future_id}", "finished_at", time.time()) + wall_end = time.time() + self.redis.hset(f"future:{future_id}", "finished_at", wall_end) + + execution_time = max(wall_end - wall_start, 0.0) + cpu_seconds = max(time.thread_time() - thread_cpu_start, 0.0) + cpu_percent = (cpu_seconds / execution_time * 100.0) if execution_time else 0.0 + gpu_resource = _read_gpu_memory_for_pid(os.getpid()) + + self.redis.hset_multiple( + f"future:{future_id}", + { + "execution_time": execution_time, + "cpu_resource": cpu_percent, + "gpu_resource": gpu_resource, + }, + ) # ------------------------------------------------------------------ # # Request forwarding # diff --git a/ventis/controller/utils/sqlalchemy.py b/ventis/controller/utils/sqlalchemy.py index 37ff883..d30056d 100644 --- a/ventis/controller/utils/sqlalchemy.py +++ b/ventis/controller/utils/sqlalchemy.py @@ -39,6 +39,13 @@ def _get_engine(database_url): return _engine +def _safe_float(value, fallback=0.0): + try: + return float(value) + except (TypeError, ValueError): + return float(fallback) + + def pull_data(redis_client): """Scan node Redis for future data""" rows = [] @@ -58,7 +65,7 @@ def send_data( redis_client: RedisClient | None = None, database_url="", ): - """UPSERT rows and attach allocated cpu/gpu from resources_by_agent.""" + """UPSERT rows and attach observed cpu/gpu (fallback to allocated resources).""" if not rows: return resources_by_agent = resources_by_agent or {} @@ -75,8 +82,15 @@ def send_data( if redis_client is not None else None ) - start = float(raw.get("created_at") or 0) - end = float(raw.get("finished_at") or time.time()) + start = _safe_float(raw.get("created_at"), 0) + end = _safe_float(raw.get("finished_at"), time.time()) + execution_time = _safe_float(raw.get("execution_time"), end - start) + cpu_resource = _safe_float( + raw.get("cpu_resource"), _safe_float(res.get("cpu"), 0) + ) + gpu_resource = _safe_float( + raw.get("gpu_resource"), _safe_float(res.get("gpu"), 0) + ) conn.execute( _UPSERT, @@ -85,9 +99,9 @@ def send_data( "session_id": session_id, "workflow": workflow, "agent": agent, - "execution_time": end - start, - "cpu_resource": float(res.get("cpu", 0)), - "gpu_resource": float(res.get("gpu", 0)), + "execution_time": max(execution_time, 0.0), + "cpu_resource": max(cpu_resource, 0.0), + "gpu_resource": max(gpu_resource, 0.0), "created_at": str(start), "updated_at": str(end), }, From 6546ad977261b74d4db0e07a39acc03cd619d7cd Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Tue, 21 Jul 2026 17:29:37 +0000 Subject: [PATCH 3/7] Remove _safe_float helper from runtime SQL metrics path --- ventis/controller/utils/sqlalchemy.py | 46 +++++++++++++++++---------- 1 file changed, 30 insertions(+), 16 deletions(-) diff --git a/ventis/controller/utils/sqlalchemy.py b/ventis/controller/utils/sqlalchemy.py index d30056d..f153887 100644 --- a/ventis/controller/utils/sqlalchemy.py +++ b/ventis/controller/utils/sqlalchemy.py @@ -39,13 +39,6 @@ def _get_engine(database_url): return _engine -def _safe_float(value, fallback=0.0): - try: - return float(value) - except (TypeError, ValueError): - return float(fallback) - - def pull_data(redis_client): """Scan node Redis for future data""" rows = [] @@ -82,15 +75,36 @@ def send_data( if redis_client is not None else None ) - start = _safe_float(raw.get("created_at"), 0) - end = _safe_float(raw.get("finished_at"), time.time()) - execution_time = _safe_float(raw.get("execution_time"), end - start) - cpu_resource = _safe_float( - raw.get("cpu_resource"), _safe_float(res.get("cpu"), 0) - ) - gpu_resource = _safe_float( - raw.get("gpu_resource"), _safe_float(res.get("gpu"), 0) - ) + try: + start = float(raw.get("created_at") or 0) + except (TypeError, ValueError): + start = 0.0 + + try: + end = float(raw.get("finished_at") or time.time()) + except (TypeError, ValueError): + end = float(time.time()) + + try: + execution_time = float(raw.get("execution_time")) + except (TypeError, ValueError): + execution_time = end - start + + try: + cpu_resource = float(raw.get("cpu_resource")) + except (TypeError, ValueError): + try: + cpu_resource = float(res.get("cpu", 0)) + except (TypeError, ValueError): + cpu_resource = 0.0 + + try: + gpu_resource = float(raw.get("gpu_resource")) + except (TypeError, ValueError): + try: + gpu_resource = float(res.get("gpu", 0)) + except (TypeError, ValueError): + gpu_resource = 0.0 conn.execute( _UPSERT, From cd159b8752c7c03bc1f45e295a807c6d6ab3ad99 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Tue, 21 Jul 2026 18:00:46 +0000 Subject: [PATCH 4/7] Remove try-based float safety in runtime SQL metrics parsing --- ventis/controller/utils/sqlalchemy.py | 41 ++++++--------------------- 1 file changed, 8 insertions(+), 33 deletions(-) diff --git a/ventis/controller/utils/sqlalchemy.py b/ventis/controller/utils/sqlalchemy.py index f153887..ac989e9 100644 --- a/ventis/controller/utils/sqlalchemy.py +++ b/ventis/controller/utils/sqlalchemy.py @@ -75,36 +75,11 @@ def send_data( if redis_client is not None else None ) - try: - start = float(raw.get("created_at") or 0) - except (TypeError, ValueError): - start = 0.0 - - try: - end = float(raw.get("finished_at") or time.time()) - except (TypeError, ValueError): - end = float(time.time()) - - try: - execution_time = float(raw.get("execution_time")) - except (TypeError, ValueError): - execution_time = end - start - - try: - cpu_resource = float(raw.get("cpu_resource")) - except (TypeError, ValueError): - try: - cpu_resource = float(res.get("cpu", 0)) - except (TypeError, ValueError): - cpu_resource = 0.0 - - try: - gpu_resource = float(raw.get("gpu_resource")) - except (TypeError, ValueError): - try: - gpu_resource = float(res.get("gpu", 0)) - except (TypeError, ValueError): - gpu_resource = 0.0 + start = float(raw.get("created_at") or 0) + end = float(raw.get("finished_at") or time.time()) + execution_time = float(raw.get("execution_time") or (end - start)) + cpu_resource = float(raw.get("cpu_resource") or res.get("cpu", 0)) + gpu_resource = float(raw.get("gpu_resource") or res.get("gpu", 0)) conn.execute( _UPSERT, @@ -113,9 +88,9 @@ def send_data( "session_id": session_id, "workflow": workflow, "agent": agent, - "execution_time": max(execution_time, 0.0), - "cpu_resource": max(cpu_resource, 0.0), - "gpu_resource": max(gpu_resource, 0.0), + "execution_time": execution_time, + "cpu_resource": cpu_resource, + "gpu_resource": gpu_resource, "created_at": str(start), "updated_at": str(end), }, From 1365697e053797317de8d37cf7685be6edd2b262 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Tue, 21 Jul 2026 18:03:41 +0000 Subject: [PATCH 5/7] Remove runtime GPU observation and simplify execution time persistence --- tests/test_runtime_sqlalchemy.py | 8 +++--- ventis/controller/local_controller.py | 37 --------------------------- ventis/controller/utils/sqlalchemy.py | 6 ++--- 3 files changed, 7 insertions(+), 44 deletions(-) diff --git a/tests/test_runtime_sqlalchemy.py b/tests/test_runtime_sqlalchemy.py index cffb11a..fc725ea 100644 --- a/tests/test_runtime_sqlalchemy.py +++ b/tests/test_runtime_sqlalchemy.py @@ -94,7 +94,7 @@ def test_pull_and_upsert(self): for value in row: self.assertNotIn(value, (None, "")) - def test_observed_metrics_override_config_allocations(self): + def test_observed_cpu_overrides_config_and_gpu_uses_config(self): redis = _FakeRedis( { "future:xyz": { @@ -112,7 +112,7 @@ def test_observed_metrics_override_config_allocations(self): ) rows = sqlmod.pull_data(redis) - sqlmod.send_data(rows, {"AgentB": {"cpu": 8, "gpu": 2}}, redis) + sqlmod.send_data(rows, {"AgentB": {"cpu": 8, "gpu": 0}}, redis) with sqlmod._get_engine("").connect() as conn: row = conn.execute( text( @@ -120,9 +120,9 @@ def test_observed_metrics_override_config_allocations(self): "FROM runtime_information WHERE future_id='xyz'" ) ).fetchone() - self.assertEqual(row[0], 1.25) + self.assertEqual(row[0], 2.0) self.assertEqual(row[1], 37.5) - self.assertEqual(row[2], 256.0) + self.assertEqual(row[2], 0.0) if __name__ == "__main__": diff --git a/ventis/controller/local_controller.py b/ventis/controller/local_controller.py index adc97ef..2d90b14 100644 --- a/ventis/controller/local_controller.py +++ b/ventis/controller/local_controller.py @@ -6,7 +6,6 @@ import logging import os import random -import subprocess import sys import time import importlib.util @@ -41,40 +40,6 @@ POLICY_RULES_KEY = "policy:rules" -def _read_gpu_memory_for_pid(pid): - """Read total GPU memory (MiB) used by a process pid via nvidia-smi.""" - try: - result = subprocess.run( - [ - "nvidia-smi", - "--query-compute-apps=pid,used_memory", - "--format=csv,noheader,nounits", - ], - capture_output=True, - text=True, - timeout=2, - check=False, - ) - except (FileNotFoundError, subprocess.SubprocessError): - return 0.0 - - if result.returncode != 0: - return 0.0 - - total = 0.0 - for line in result.stdout.splitlines(): - line = line.strip() - if not line: - continue - try: - pid_value, mem_value = [part.strip() for part in line.split(",", 1)] - if int(pid_value) == int(pid): - total += float(mem_value) - except (TypeError, ValueError): - continue - return total - - class LocalController(object): """Manages the gRPC frontend and processes incoming requests from the queue.""" @@ -505,14 +470,12 @@ def _execute_locally( execution_time = max(wall_end - wall_start, 0.0) cpu_seconds = max(time.thread_time() - thread_cpu_start, 0.0) cpu_percent = (cpu_seconds / execution_time * 100.0) if execution_time else 0.0 - gpu_resource = _read_gpu_memory_for_pid(os.getpid()) self.redis.hset_multiple( f"future:{future_id}", { "execution_time": execution_time, "cpu_resource": cpu_percent, - "gpu_resource": gpu_resource, }, ) diff --git a/ventis/controller/utils/sqlalchemy.py b/ventis/controller/utils/sqlalchemy.py index ac989e9..e18fad5 100644 --- a/ventis/controller/utils/sqlalchemy.py +++ b/ventis/controller/utils/sqlalchemy.py @@ -58,7 +58,7 @@ def send_data( redis_client: RedisClient | None = None, database_url="", ): - """UPSERT rows and attach observed cpu/gpu (fallback to allocated resources).""" + """UPSERT rows with observed CPU and configured GPU resource values.""" if not rows: return resources_by_agent = resources_by_agent or {} @@ -77,9 +77,9 @@ def send_data( ) start = float(raw.get("created_at") or 0) end = float(raw.get("finished_at") or time.time()) - execution_time = float(raw.get("execution_time") or (end - start)) + execution_time = end - start cpu_resource = float(raw.get("cpu_resource") or res.get("cpu", 0)) - gpu_resource = float(raw.get("gpu_resource") or res.get("gpu", 0)) + gpu_resource = float(res.get("gpu", 0)) conn.execute( _UPSERT, From fffdd3626fc8395d3af61773c272212978320797 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Tue, 21 Jul 2026 18:08:44 +0000 Subject: [PATCH 6/7] Remove execution_time runtime hash write in LocalController --- ventis/controller/local_controller.py | 14 ++++---------- 1 file changed, 4 insertions(+), 10 deletions(-) diff --git a/ventis/controller/local_controller.py b/ventis/controller/local_controller.py index 2d90b14..e67bb5c 100644 --- a/ventis/controller/local_controller.py +++ b/ventis/controller/local_controller.py @@ -467,17 +467,11 @@ def _execute_locally( wall_end = time.time() self.redis.hset(f"future:{future_id}", "finished_at", wall_end) - execution_time = max(wall_end - wall_start, 0.0) + wall_duration = max(wall_end - wall_start, 0.0) cpu_seconds = max(time.thread_time() - thread_cpu_start, 0.0) - cpu_percent = (cpu_seconds / execution_time * 100.0) if execution_time else 0.0 - - self.redis.hset_multiple( - f"future:{future_id}", - { - "execution_time": execution_time, - "cpu_resource": cpu_percent, - }, - ) + cpu_percent = (cpu_seconds / wall_duration * 100.0) if wall_duration else 0.0 + + self.redis.hset(f"future:{future_id}", "cpu_resource", cpu_percent) # ------------------------------------------------------------------ # # Request forwarding # From 2c292050bcf6101183a6516ecc4c32dafdcb4893 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Tue, 21 Jul 2026 18:11:47 +0000 Subject: [PATCH 7/7] Inline execution_time expression in SQL upsert payload --- ventis/controller/utils/sqlalchemy.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/ventis/controller/utils/sqlalchemy.py b/ventis/controller/utils/sqlalchemy.py index e18fad5..922a178 100644 --- a/ventis/controller/utils/sqlalchemy.py +++ b/ventis/controller/utils/sqlalchemy.py @@ -77,7 +77,6 @@ def send_data( ) start = float(raw.get("created_at") or 0) end = float(raw.get("finished_at") or time.time()) - execution_time = end - start cpu_resource = float(raw.get("cpu_resource") or res.get("cpu", 0)) gpu_resource = float(res.get("gpu", 0)) @@ -88,7 +87,7 @@ def send_data( "session_id": session_id, "workflow": workflow, "agent": agent, - "execution_time": execution_time, + "execution_time": end - start, "cpu_resource": cpu_resource, "gpu_resource": gpu_resource, "created_at": str(start),