From 63a00c28fb13c63c26a637095f8f61c6c4b22ada Mon Sep 17 00:00:00 2001 From: Saaketh Sodanapalli Date: Mon, 3 Aug 2026 15:43:27 -0700 Subject: [PATCH 1/8] added a failure field to the future that fails if it happens --- ventis/controller/local_controller.py | 8 ++++++++ ventis/future.py | 6 ++++++ ventis/llm/bedrock.py | 7 ++++++- 3 files changed, 20 insertions(+), 1 deletion(-) diff --git a/ventis/controller/local_controller.py b/ventis/controller/local_controller.py index efcc4d0..05800da 100644 --- a/ventis/controller/local_controller.py +++ b/ventis/controller/local_controller.py @@ -501,6 +501,8 @@ def _execute_locally( "method": function, "args": json.dumps(args), "created_at": wall_start, + "failed": 0, + "error_message": "", }, ) if request_id: @@ -528,6 +530,12 @@ def _execute_locally( ) result = method(**args) + if str(self.redis.hget(f"future:{future_id}:metrics", "failed")) == "1": + raise RuntimeError( + self.redis.hget(f"future:{future_id}:metrics", "error_message") + or "Unknown error" + ) + # Serialize the result if isinstance(result, (dict, list)): serialized = json.dumps(result) diff --git a/ventis/future.py b/ventis/future.py index c63c466..351c8e9 100644 --- a/ventis/future.py +++ b/ventis/future.py @@ -137,6 +137,12 @@ def _poll_redis(self): error = self.redis.hget(self._key(), "error") if error: raise RuntimeError(error) + failed = self.redis.hget(f"future:{self.id}:metrics", "failed") + if str(failed) == "1": + raise RuntimeError( + self.redis.hget(f"future:{self.id}:metrics", "error_message") + or "Unknown error" + ) result = self.redis.hget(self._key(), "result") if result is not None and result != "": self.result = result diff --git a/ventis/llm/bedrock.py b/ventis/llm/bedrock.py index 1a5fefb..b843234 100644 --- a/ventis/llm/bedrock.py +++ b/ventis/llm/bedrock.py @@ -27,7 +27,7 @@ def call_bedrock(model_id: str, messages: list, inference_config: dict, region: modelId=model_id, messages=messages, inferenceConfig=inference_config ) return response - except Exception: + except Exception as e: # Counts failed attempts for this call. With no retry logic today this is # always 0 or 1, matching `failed`; once retries are added, a call that # eventually succeeds can still report error_count > 0 while failed stays 0. @@ -35,6 +35,11 @@ def call_bedrock(model_id: str, messages: list, inference_config: dict, region: metrics_key = ventis_context.get_current_metrics_key() if metrics_key: _redis.hincrby(metrics_key, "error_count", 1) + if future_id: + _redis.hset_multiple(f"future:{future_id}:metrics", { + "failed": 1, + "error_message": str(e), + }) raise finally: if future_id: From c7d48684785354b5041f4ea89fc8db740567907f Mon Sep 17 00:00:00 2001 From: Saaketh Sodanapalli Date: Mon, 3 Aug 2026 16:21:04 -0700 Subject: [PATCH 2/8] fixed some issues --- tests/test_local_controller_metrics.py | 6 ++++++ ventis/controller/local_controller.py | 28 +++++++++++++++++++------- ventis/future.py | 13 ++++++------ 3 files changed, 34 insertions(+), 13 deletions(-) diff --git a/tests/test_local_controller_metrics.py b/tests/test_local_controller_metrics.py index aafa3e1..b995af0 100644 --- a/tests/test_local_controller_metrics.py +++ b/tests/test_local_controller_metrics.py @@ -175,6 +175,12 @@ def boom(name): ) self.assertIn("Execution failed", redis.hget("future:future-2", "result")) + self.assertEqual( + redis.hget("future:future-2:metrics", "failed"), 1 + ) + self.assertEqual( + redis.hget("future:future-2:metrics", "error_message"), "nope" + ) self.assertEqual( redis.hget("controller:localhost:50051:metrics", "requests_served"), 1 ) diff --git a/ventis/controller/local_controller.py b/ventis/controller/local_controller.py index 05800da..31931e1 100644 --- a/ventis/controller/local_controller.py +++ b/ventis/controller/local_controller.py @@ -293,6 +293,12 @@ def run(self): logger.error("Invalid JSON in request: %s", raw) except Exception as e: logger.error("Error processing request: %s", e) + future_id = data.get("future_id") + if future_id: + self.redis.hset_multiple( + f"future:{future_id}:metrics", + {"failed": 1, "error_message": str(e)}, + ) else: time.sleep(0.001) except KeyboardInterrupt: @@ -530,12 +536,6 @@ def _execute_locally( ) result = method(**args) - if str(self.redis.hget(f"future:{future_id}:metrics", "failed")) == "1": - raise RuntimeError( - self.redis.hget(f"future:{future_id}:metrics", "error_message") - or "Unknown error" - ) - # Serialize the result if isinstance(result, (dict, list)): serialized = json.dumps(result) @@ -561,7 +561,13 @@ def _execute_locally( logger.error("Failed to execute %s.%s: %s", service, function, e) # Treat script-level crash as a string result to avoid hanging - self.redis.hset(f"future:{future_id}:metrics", "failed", 1) + self.redis.hset_multiple( + f"future:{future_id}:metrics", + { + "failed": 1, + "error_message": str(e), + }, + ) self.redis.hset(f"future:{future_id}", "result", f"Execution failed: {e}") self.redis.hincrby(self._metrics_key, "full_failures", 1) if origin and origin != self._my_endpoint: @@ -618,6 +624,10 @@ def _forward_request(self, endpoint, data): future_id = data.get("future_id") if future_id: self.redis.hset(f"future:{future_id}", "error", str(e)) + self.redis.hset_multiple( + f"future:{future_id}:metrics", + {"failed": 1, "error_message": str(e)}, + ) def _send_result_callback(self, origin, future_id, result): """Send a result back to the originating controller via WriteResult RPC.""" @@ -645,6 +655,10 @@ def _send_result_callback(self, origin, future_id, result): except Exception as e: logger.error("Failed to send result callback to %s: %s", origin, e) + self.redis.hset_multiple( + f"future:{future_id}:metrics", + {"failed": 1, "error_message": str(e)}, + ) # ------------------------------------------------------------------ # # Shutdown # diff --git a/ventis/future.py b/ventis/future.py index 351c8e9..864ee4a 100644 --- a/ventis/future.py +++ b/ventis/future.py @@ -137,12 +137,6 @@ def _poll_redis(self): error = self.redis.hget(self._key(), "error") if error: raise RuntimeError(error) - failed = self.redis.hget(f"future:{self.id}:metrics", "failed") - if str(failed) == "1": - raise RuntimeError( - self.redis.hget(f"future:{self.id}:metrics", "error_message") - or "Unknown error" - ) result = self.redis.hget(self._key(), "result") if result is not None and result != "": self.result = result @@ -155,6 +149,13 @@ def value(self, timeout=None): Returns immediately if the result is already available locally. Polls Redis periodically to check for computed results. """ + failed = self.redis.hget(f"future:{self.id}:metrics", "failed") + if str(failed) == "1": + raise RuntimeError( + self.redis.hget(f"future:{self.id}:metrics", "error_message") + or "Unknown error" + ) + if self.result is not None: return self.result From 68ea2e668b46bf70ad3decfa9ee6d89f7cdd69af Mon Sep 17 00:00:00 2001 From: Saaketh Sodanapalli Date: Mon, 3 Aug 2026 18:15:14 -0700 Subject: [PATCH 3/8] simplified logic and sent it to origin host --- tests/test_error_propagation.py | 117 ++++++++++++- tests/test_future.py | 10 ++ tests/test_local_controller_metrics.py | 94 ++++++++++- ventis/controller/local_controller.py | 154 ++++++++++++------ .../controller/local_controller_frontend.py | 12 +- ventis/future.py | 6 +- 6 files changed, 325 insertions(+), 68 deletions(-) diff --git a/tests/test_error_propagation.py b/tests/test_error_propagation.py index ddd0a0b..a998cac 100644 --- a/tests/test_error_propagation.py +++ b/tests/test_error_propagation.py @@ -1,6 +1,7 @@ import os import sys import unittest +import json from types import SimpleNamespace from unittest.mock import MagicMock @@ -15,7 +16,9 @@ ) from ventis.controller.local_controller import LocalController +from ventis.controller.local_controller_frontend import LocalControllerServicer from ventis.future import Future +import local_controler_pb2 class _FakeRedis: @@ -25,19 +28,29 @@ def __init__(self): def hset(self, name, field, value): self.hashes.setdefault(name, {})[field] = value + def hset_multiple(self, name, mapping): + self.hashes.setdefault(name, {}).update(mapping) + def hget(self, name, field): return self.hashes.get(name, {}).get(field) +def _bind_failure_marker(controller): + controller._mark_future_failed = lambda future_id, error, origin=None: ( + LocalController._mark_future_failed(controller, future_id, error, origin) + ) + return controller + + class ErrorPropagationTests(unittest.TestCase): def test_forward_request_writes_future_error_on_grpc_failure(self): redis = _FakeRedis() stub = SimpleNamespace(Execute=MagicMock(side_effect=RuntimeError("boom"))) - controller = SimpleNamespace( + controller = _bind_failure_marker(SimpleNamespace( redis=redis, _my_endpoint="172.31.19.107:50051", _get_remote_stub=lambda endpoint: stub, - ) + )) data = { "future_id": "future-1", "service": "ExampleAgent", @@ -50,19 +63,28 @@ def test_forward_request_writes_future_error_on_grpc_failure(self): self.assertEqual(redis.hget("future:future-1", "error"), "boom") stub.Execute.assert_called_once() - def test_future_poll_redis_raises_runtime_error_when_error_is_present(self): + def test_future_value_raises_runtime_error_when_error_is_present(self): redis = _FakeRedis() redis.hset("future:future-1", "error", "boom") - future = SimpleNamespace(redis=redis, _key=lambda: "future:future-1") + future = SimpleNamespace( + redis=redis, + _key=lambda: "future:future-1", + _poll_redis=lambda: Future._poll_redis(future), + result=None, + ) with self.assertRaisesRegex(RuntimeError, "boom"): - Future._poll_redis(future) + Future.value(future) def test_future_poll_redis_returns_result_when_error_is_absent(self): redis = _FakeRedis() redis.hset("future:future-1", "result", "Hello, World!") future = SimpleNamespace( - redis=redis, _key=lambda: "future:future-1", result=None + redis=redis, + _key=lambda: "future:future-1", + _poll_redis=lambda: Future._poll_redis(future), + id="future-1", + result=None, ) result = Future._poll_redis(future) @@ -70,6 +92,89 @@ def test_future_poll_redis_returns_result_when_error_is_absent(self): self.assertEqual(result, "Hello, World!") self.assertEqual(future.result, "Hello, World!") + def test_future_value_raises_when_metrics_mark_it_failed(self): + redis = _FakeRedis() + redis.hset_multiple( + "future:future-1:metrics", + {"failed": 1, "error_message": "agent exploded"}, + ) + future = SimpleNamespace( + redis=redis, + _key=lambda: "future:future-1", + _poll_redis=lambda: Future._poll_redis(future), + id="future-1", + result=None, + ) + + with self.assertRaisesRegex(RuntimeError, "agent exploded"): + Future.value(future) + + def test_result_callback_sends_error_separately_from_result(self): + redis = _FakeRedis() + stub = SimpleNamespace(WriteResult=MagicMock()) + controller = SimpleNamespace( + redis=redis, + agent_name="ExampleAgent", + _get_remote_stub=lambda endpoint: stub, + ) + + LocalController._send_result_callback( + controller, + "origin:50051", + "future-1", + failed=1, + error_message="agent exploded", + ) + + payload = stub.WriteResult.call_args.args[0].resonse + self.assertEqual( + json.loads(payload), + { + "future_id": "future-1", + "result": None, + "failed": 1, + "error_message": "agent exploded", + }, + ) + + def test_write_result_persists_remote_error_as_terminal_failure(self): + redis = _FakeRedis() + servicer = SimpleNamespace(redis=redis) + request = local_controler_pb2.JsonResponse( + resonse=json.dumps( + { + "future_id": "future-1", + "failed": 1, + "error_message": "remote exploded", + } + ) + ) + context = SimpleNamespace(peer=lambda: "peer:50051") + + LocalControllerServicer.WriteResult(servicer, request, context) + + self.assertEqual( + redis.hget("future:future-1:metrics", "failed"), 1 + ) + self.assertEqual( + redis.hget("future:future-1:metrics", "error_message"), + "remote exploded", + ) + + def test_malformed_request_with_future_id_is_marked_failed(self): + redis = _FakeRedis() + controller = _bind_failure_marker( + SimpleNamespace(redis=redis, _my_endpoint="localhost:50051") + ) + + LocalController._process_request(controller, {"future_id": "future-1"}) + + self.assertEqual( + redis.hget("future:future-1", "error"), + "Malformed request: missing service, function, or future_id", + ) + self.assertEqual(redis.hget("future:future-1:metrics", "failed"), 1) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_future.py b/tests/test_future.py index 9e79fec..c277358 100644 --- a/tests/test_future.py +++ b/tests/test_future.py @@ -68,6 +68,16 @@ def test_parent_is_the_currently_executing_future_id(self): self.fake_redis.hashes[f"future:{f.id}"]["parent"], "caller-future-id" ) + def test_submission_failure_is_raised_by_value_not_constructor(self): + future_module.Future._stub.Execute.side_effect = RuntimeError("submit failed") + + future = future_module.Future( + parent="ignored/file.py", service="Svc", method="do_thing" + ) + + with self.assertRaisesRegex(RuntimeError, "submit failed"): + future.value() + if __name__ == "__main__": unittest.main() diff --git a/tests/test_local_controller_metrics.py b/tests/test_local_controller_metrics.py index b995af0..90491a7 100644 --- a/tests/test_local_controller_metrics.py +++ b/tests/test_local_controller_metrics.py @@ -2,6 +2,8 @@ import sys import threading import unittest +import json +from unittest.mock import MagicMock from concurrent.futures import ThreadPoolExecutor from types import SimpleNamespace from unittest.mock import patch @@ -19,6 +21,18 @@ from ventis.controller.local_controller import LocalController +def _bind_failure_marker(controller): + controller._mark_future_failed = lambda future_id, error, origin=None: ( + LocalController._mark_future_failed(controller, future_id, error, origin) + ) + controller._send_result_callback = ( + lambda origin, future_id, result=None, failed=0, error_message="": LocalController._send_result_callback( + controller, origin, future_id, result, failed, error_message + ) + ) + return controller + + class _FakeRedisClient: def __init__(self): self.counters = {} @@ -123,7 +137,7 @@ def stop_after_one_tick(timeout): def test_execute_locally_writes_gpu_resource_to_future_hash(self): redis = _FakeRedis() agent = SimpleNamespace(greet=lambda name: f"hello {name}") - controller = SimpleNamespace( + controller = _bind_failure_marker(SimpleNamespace( redis=redis, agent=agent, agent_name="Greeter", @@ -131,7 +145,7 @@ def test_execute_locally_writes_gpu_resource_to_future_hash(self): _my_endpoint="localhost:50051", _metrics_key="controller:localhost:50051:metrics", _resolve_future_args=lambda args: args, - ) + )) with patch( "ventis.controller.local_controller.read_gpu_percent", return_value=17.5 @@ -157,7 +171,7 @@ def boom(name): raise ValueError("nope") agent = SimpleNamespace(greet=boom) - controller = SimpleNamespace( + controller = _bind_failure_marker(SimpleNamespace( redis=redis, agent=agent, agent_name="Greeter", @@ -165,7 +179,7 @@ def boom(name): _my_endpoint="localhost:50051", _metrics_key="controller:localhost:50051:metrics", _resolve_future_args=lambda args: args, - ) + )) with patch( "ventis.controller.local_controller.read_gpu_percent", return_value=0.0 @@ -174,7 +188,8 @@ def boom(name): controller, "Greeter", "greet", {"name": "world"}, "future-2" ) - self.assertIn("Execution failed", redis.hget("future:future-2", "result")) + self.assertEqual(redis.hget("future:future-2", "error"), "nope") + self.assertIsNone(redis.hget("future:future-2", "result")) self.assertEqual( redis.hget("future:future-2:metrics", "failed"), 1 ) @@ -188,6 +203,75 @@ def boom(name): redis.hget("controller:localhost:50051:metrics", "full_failures"), 1 ) + def test_execute_locally_marks_missing_agent_as_failed(self): + redis = _FakeRedis() + controller = _bind_failure_marker(SimpleNamespace( + redis=redis, + agent=None, + agent_name="MissingAgent", + agent_id="agent-1", + _my_endpoint="localhost:50051", + _metrics_key="controller:localhost:50051:metrics", + )) + + with patch( + "ventis.controller.local_controller.read_gpu_percent", return_value=0.0 + ): + LocalController._execute_locally( + controller, "MissingAgent", "greet", {}, "future-3" + ) + + self.assertEqual( + redis.hget("future:future-3", "error"), "No agent loaded" + ) + self.assertEqual(redis.hget("future:future-3:metrics", "failed"), 1) + self.assertEqual( + redis.hget("future:future-3:metrics", "error_message"), + "No agent loaded", + ) + + def test_remote_execution_failure_sends_error_callback(self): + redis = _FakeRedis() + stub = SimpleNamespace(WriteResult=MagicMock()) + + def boom(): + raise ValueError("remote nope") + + controller = _bind_failure_marker(SimpleNamespace( + redis=redis, + agent=SimpleNamespace(greet=boom), + agent_name="Greeter", + agent_id="agent-1", + _my_endpoint="target:50051", + _metrics_key="controller:target:50051:metrics", + _resolve_future_args=lambda args: args, + _get_remote_stub=lambda endpoint: stub, + )) + + with patch( + "ventis.controller.local_controller.read_gpu_percent", return_value=0.0 + ): + LocalController._execute_locally( + controller, + "Greeter", + "greet", + {}, + "future-4", + origin="origin:50051", + ) + + self.assertEqual(redis.hget("future:future-4", "error"), "remote nope") + payload = stub.WriteResult.call_args.args[0].resonse + self.assertEqual( + json.loads(payload), + { + "future_id": "future-4", + "result": None, + "failed": 1, + "error_message": "remote nope", + }, + ) + if __name__ == "__main__": unittest.main() diff --git a/ventis/controller/local_controller.py b/ventis/controller/local_controller.py index 31931e1..3599704 100644 --- a/ventis/controller/local_controller.py +++ b/ventis/controller/local_controller.py @@ -286,6 +286,7 @@ def run(self): while True: if not self.request_queue.empty(): raw = self.request_queue.get() + data = None try: data = json.loads(raw) self._process_request(data) @@ -293,17 +294,35 @@ def run(self): logger.error("Invalid JSON in request: %s", raw) except Exception as e: logger.error("Error processing request: %s", e) - future_id = data.get("future_id") - if future_id: - self.redis.hset_multiple( - f"future:{future_id}:metrics", - {"failed": 1, "error_message": str(e)}, + if isinstance(data, dict): + self._mark_future_failed( + data.get("future_id"), e, data.get("origin") ) else: time.sleep(0.001) except KeyboardInterrupt: self.stop() + def _mark_future_failed(self, future_id, error, origin=None): + """Persist a terminal failure locally and, when needed, notify the origin.""" + if not future_id: + return + + error_message = str(error) or "Unknown error" + self.redis.hset(f"future:{future_id}", "error", error_message) + self.redis.hset_multiple( + f"future:{future_id}:metrics", + {"failed": 1, "error_message": error_message}, + ) + + if origin and origin != self._my_endpoint: + self._send_result_callback( + origin, + future_id, + failed=1, + error_message=error_message, + ) + def _process_request(self, data): """ Route a request to the correct controller. @@ -340,15 +359,18 @@ def _process_request(self, data): if not service or not function or not future_id: logger.error("Malformed request, missing required fields: %s", data) + self._mark_future_failed( + future_id, + "Malformed request: missing service, function, or future_id", + origin, + ) return # Check policy before routing if not self._check_policy(service, context): err_msg = f"Unauthorized: Policy denied access to service '{service}'" logger.warning(err_msg) - self.redis.hset(f"future:{future_id}", "result", err_msg) - if origin and origin != self._my_endpoint: - self._send_result_callback(origin, future_id, err_msg) + self._mark_future_failed(future_id, err_msg, origin) return # Resolve which endpoint to route to. @@ -366,6 +388,11 @@ def _process_request(self, data): logger.error( "No endpoint found for service '%s' in routing table.", service ) + self._mark_future_failed( + future_id, + f"No endpoint found for service '{service}'", + origin, + ) return if endpoint == self._my_endpoint: @@ -460,6 +487,17 @@ def _resolve_future_args(self, args, poll_interval=0.01, timeout=300): ) start = time.time() while True: + error = self.redis.hget(future_key, "error") + if error: + raise RuntimeError(error) + failed = self.redis.hget(f"future:{value}:metrics", "failed") + if str(failed) == "1": + raise RuntimeError( + self.redis.hget( + f"future:{value}:metrics", "error_message" + ) + or "Unknown error" + ) # print("Waiting for result for future next iteration %s", value) result = self.redis.hget(future_key, "result") if result is not None and result != "": @@ -518,11 +556,15 @@ def _execute_locally( ventis_context.set_current_metrics_key(self._metrics_key) if self.agent is None: logger.error("No agent loaded, cannot execute %s.%s", service, function) + self._mark_future_failed(future_id, "No agent loaded", origin) return method = getattr(self.agent, function, None) if method is None: logger.error("Agent %s has no method '%s'", self.agent_name, function) + self._mark_future_failed( + future_id, f"Agent {self.agent_name} has no method '{function}'", origin + ) return self.redis.hincrby(self._metrics_key, "requests_served", 1) @@ -547,7 +589,13 @@ def _execute_locally( # If the request came from another node, send result back to origin if origin and origin != self._my_endpoint: - self._send_result_callback(origin, future_id, serialized) + self._send_result_callback( + origin, + future_id, + result=serialized, + failed=0, + error_message="", + ) logger.info( "Completed %s.%s (future=%s) -> %s", @@ -560,37 +608,41 @@ def _execute_locally( except Exception as e: logger.error("Failed to execute %s.%s: %s", service, function, e) - # Treat script-level crash as a string result to avoid hanging - self.redis.hset_multiple( - f"future:{future_id}:metrics", - { - "failed": 1, - "error_message": str(e), - }, - ) - self.redis.hset(f"future:{future_id}", "result", f"Execution failed: {e}") + self._mark_future_failed(future_id, e, origin) self.redis.hincrby(self._metrics_key, "full_failures", 1) - if origin and origin != self._my_endpoint: - self._send_result_callback(origin, future_id, f"Execution failed: {e}") - - wall_end = time.time() - self.redis.hset(f"future:{future_id}:metrics", "finished_at", wall_end) - - wall_duration = max(wall_end - wall_start, 0.0) - cpu_seconds = max(time.thread_time() - thread_cpu_start, 0.0) - cpu_percent = (cpu_seconds / wall_duration * 100.0) if wall_duration else 0.0 - - self.redis.hset(f"future:{future_id}:metrics", "cpu_resource", cpu_percent) - self.redis.hset(f"future:{future_id}:metrics", "gpu_resource", read_gpu_percent()) - self.redis.hset(f"future:{future_id}:metrics", "agent", self.agent_id) - ventis_context.set_current_future_id(parent or "") - - if submitted_at is not None: - self.redis.hset( - f"future:{future_id}:metrics", - "queue_time", - max(wall_start - submitted_at, 0.0), + finally: + wall_end = time.time() + wall_duration = max(wall_end - wall_start, 0.0) + cpu_seconds = max(time.thread_time() - thread_cpu_start, 0.0) + cpu_percent = ( + (cpu_seconds / wall_duration * 100.0) if wall_duration else 0.0 ) + try: + gpu_percent = read_gpu_percent() + except Exception as metric_error: + logger.error("Failed to read GPU metrics: %s", metric_error) + gpu_percent = "" + + try: + self.redis.hset_multiple( + f"future:{future_id}:metrics", + { + "finished_at": wall_end, + "cpu_resource": cpu_percent, + "gpu_resource": gpu_percent, + "agent": self.agent_id, + **( + { + "queue_time": max(wall_start - submitted_at, 0.0) + } + if submitted_at is not None + else {} + ), + }, + ) + except Exception as metric_error: + logger.error("Failed to finalize metrics for future %s: %s", future_id, metric_error) + ventis_context.set_current_future_id(parent or "") # ------------------------------------------------------------------ # # Request forwarding # @@ -621,16 +673,12 @@ def _forward_request(self, endpoint, data): logger.debug("Forwarded request to %s", endpoint) except Exception as e: logger.error("Failed to forward request to %s: %s", endpoint, e) - future_id = data.get("future_id") - if future_id: - self.redis.hset(f"future:{future_id}", "error", str(e)) - self.redis.hset_multiple( - f"future:{future_id}:metrics", - {"failed": 1, "error_message": str(e)}, - ) + self._mark_future_failed(data.get("future_id"), e) - def _send_result_callback(self, origin, future_id, result): - """Send a result back to the originating controller via WriteResult RPC.""" + def _send_result_callback( + self, origin, future_id, result=None, failed=0, error_message="" + ): + """Send a result and its failure metadata to the originating controller.""" if not result: logger.warning( "Agent '%s' is sending an empty/None result for future %s to origin %s, result: %s", @@ -641,7 +689,12 @@ def _send_result_callback(self, origin, future_id, result): ) stub = self._get_remote_stub(origin) - payload = json.dumps({"future_id": future_id, "result": result}) + payload = json.dumps({ + "future_id": future_id, + "result": result, + "failed": int(bool(failed)), + "error_message": str(error_message or ""), + }) logger.info("Payload: Future %s,Sent %s ", future_id, payload) request = local_controler_pb2.JsonResponse(resonse=payload) try: @@ -655,10 +708,7 @@ def _send_result_callback(self, origin, future_id, result): except Exception as e: logger.error("Failed to send result callback to %s: %s", origin, e) - self.redis.hset_multiple( - f"future:{future_id}:metrics", - {"failed": 1, "error_message": str(e)}, - ) + self._mark_future_failed(future_id, f"Result callback failed: {e}") # ------------------------------------------------------------------ # # Shutdown # diff --git a/ventis/controller/local_controller_frontend.py b/ventis/controller/local_controller_frontend.py index c7ac28d..a6e478d 100644 --- a/ventis/controller/local_controller_frontend.py +++ b/ventis/controller/local_controller_frontend.py @@ -50,7 +50,8 @@ def WriteResult(self, request, context): data = json.loads(request.resonse) future_id = data.get("future_id") result = data.get("result") - error = data.get("error") + failed = int(bool(data.get("failed", 0))) + error_message = str(data.get("error_message") or "") logger.info( f"WriteResult: received result for future {future_id}: {result}" @@ -61,10 +62,13 @@ def WriteResult(self, request, context): ) if future_id: - if error is not None: - self.redis.hset(f"future:{future_id}", "error", error) + self.redis.hset_multiple( + f"future:{future_id}:metrics", + {"failed": failed, "error_message": error_message}, + ) + if failed: logger.info("WriteResult: wrote error for future %s", future_id) - if result is not None: + elif result is not None: self.redis.hset(f"future:{future_id}", "result", result) logger.info( "WriteResult: wrote result for future %s, result %s", diff --git a/ventis/future.py b/ventis/future.py index 864ee4a..598bc4f 100644 --- a/ventis/future.py +++ b/ventis/future.py @@ -118,7 +118,11 @@ def _submit_request(self): ) except Exception as e: logger.error("gRPC call failed for %s.%s: %s", self.service, self.method, e) - raise + self.redis.hset(f"future:{self.id}", "error", str(e)) + self.redis.hset_multiple(f"future:{self.id}:metrics", { + "failed": 1, + "error_message": str(e), + }) def _key(self): """Redis key for this future's hash.""" From ce69d82b75d3af78560e56031caa201d4466c0f5 Mon Sep 17 00:00:00 2001 From: Saaketh Sodanapalli Date: Mon, 3 Aug 2026 18:19:21 -0700 Subject: [PATCH 4/8] removed some useless code --- ventis/controller/local_controller.py | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/ventis/controller/local_controller.py b/ventis/controller/local_controller.py index 3599704..f1ac4d7 100644 --- a/ventis/controller/local_controller.py +++ b/ventis/controller/local_controller.py @@ -294,10 +294,9 @@ def run(self): logger.error("Invalid JSON in request: %s", raw) except Exception as e: logger.error("Error processing request: %s", e) - if isinstance(data, dict): - self._mark_future_failed( + self._mark_future_failed( data.get("future_id"), e, data.get("origin") - ) + ) else: time.sleep(0.001) except KeyboardInterrupt: From e40eba15b9e1cb4f22fdc9963e14268ae385bca0 Mon Sep 17 00:00:00 2001 From: Saaketh Sodanapalli Date: Mon, 3 Aug 2026 18:23:09 -0700 Subject: [PATCH 5/8] more slop removal --- ventis/controller/local_controller.py | 41 +++++++++++---------------- 1 file changed, 16 insertions(+), 25 deletions(-) diff --git a/ventis/controller/local_controller.py b/ventis/controller/local_controller.py index f1ac4d7..9f53a3e 100644 --- a/ventis/controller/local_controller.py +++ b/ventis/controller/local_controller.py @@ -616,31 +616,22 @@ def _execute_locally( cpu_percent = ( (cpu_seconds / wall_duration * 100.0) if wall_duration else 0.0 ) - try: - gpu_percent = read_gpu_percent() - except Exception as metric_error: - logger.error("Failed to read GPU metrics: %s", metric_error) - gpu_percent = "" - - try: - self.redis.hset_multiple( - f"future:{future_id}:metrics", - { - "finished_at": wall_end, - "cpu_resource": cpu_percent, - "gpu_resource": gpu_percent, - "agent": self.agent_id, - **( - { - "queue_time": max(wall_start - submitted_at, 0.0) - } - if submitted_at is not None - else {} - ), - }, - ) - except Exception as metric_error: - logger.error("Failed to finalize metrics for future %s: %s", future_id, metric_error) + gpu_percent = read_gpu_percent() + + self.redis.hset_multiple( + f"future:{future_id}:metrics", + { + "finished_at": wall_end, + "cpu_resource": cpu_percent, + "gpu_resource": gpu_percent, + "agent": self.agent_id, + **( + {"queue_time": max(wall_start - submitted_at, 0.0)} + if submitted_at is not None + else {} + ), + }, + ) ventis_context.set_current_future_id(parent or "") # ------------------------------------------------------------------ # From 06ad44c961c9d3368c4eca72d21763417eea4432 Mon Sep 17 00:00:00 2001 From: Saaketh Sodanapalli Date: Wed, 5 Aug 2026 10:11:27 -0700 Subject: [PATCH 6/8] Consolidate future:{future_id} and future:{future_id}:metrics into one Redis key Fold execution metrics (cpu/gpu, timings, LLM usage, failure details) into the future's own hash instead of a separate :metrics key, so a single snapshot travels between origin and executor in both request and completion callbacks. Also fixes the completion callback firing before final metrics (finished_at/cpu_resource/gpu_resource/agent/queue_time) were written. Co-Authored-By: Claude Sonnet 5 --- FUTURE_PLANS.md | 212 ++++++++++++++++++ examples/portfolio/agents/advisor_agent.py | 2 +- examples/portfolio/agents/intent_agent.py | 2 +- examples/text2sql/agents/vllm_agent.py | 2 +- tests/test_error_propagation.py | 92 +++++++- tests/test_local_controller_metrics.py | 74 ++++-- tests/test_runtime_sqlalchemy.py | 20 +- ventis/controller/local_controller.py | 77 ++++--- .../controller/local_controller_frontend.py | 13 +- ventis/controller/utils/future_schema.py | 60 +++++ ventis/controller/utils/sqlalchemy.py | 20 +- ventis/future.py | 8 +- ventis/llm/bedrock.py | 6 +- ventis/stub_generator.py | 8 + 14 files changed, 508 insertions(+), 88 deletions(-) create mode 100644 FUTURE_PLANS.md create mode 100644 ventis/controller/utils/future_schema.py diff --git a/FUTURE_PLANS.md b/FUTURE_PLANS.md new file mode 100644 index 0000000..746ba73 --- /dev/null +++ b/FUTURE_PLANS.md @@ -0,0 +1,212 @@ +# Future State Synchronization Plan + +## Purpose + +Ensure that the origin controller's `future:{future_id}` record contains the +complete future state, including execution metrics and failure details, while +preserving the executor-local metrics used by the local controller and runtime +reporting. + +## Current architecture + +The origin creates a future record in its Redis instance: + +```text +future:{future_id} +``` + +When a request is forwarded to another EC2 instance, the executor receives the +future ID and request data, not the origin's Redis hash or a live Redis object. +Because Redis keys are scoped to a Redis server, the executor's: + +```text +future:{future_id} +``` + +is a different Redis hash from the origin's identically named key. + +The executor also creates: + +```text +future:{future_id}:metrics +``` + +This hash contains execution-specific data, including: + +- `id` +- `request_id` +- `service` +- `method` +- `args` +- `created_at` +- `finished_at` +- `failed` +- `error_message` +- `cpu_resource` +- `gpu_resource` +- `agent` +- queue timing +- LLM model and token usage fields, when applicable + +The executor additionally updates its controller-level metrics hash: + +```text +controller:{host}:{port}:metrics +``` + +That hash contains aggregate controller data such as requests served and full +failures. It is separate from the per-future metrics hash and should remain so. + +## Current result handoff + +The executor currently sends this callback payload to the origin: + +```json +{ + "future_id": "", + "result": "" +} +``` + +The callback is created in `LocalController._send_result_callback()` and +received by `WriteResult()` in the local controller frontend. The origin writes +the received result into its local `future:{future_id}` hash. + +The callback currently does not send: + +- `failed` +- `error_message` +- the `future:{future_id}:metrics` hash +- CPU/GPU/timing fields +- LLM usage fields + +Therefore, the executor's metrics remain only in the executor's Redis instance. + +## Current failure behavior + +Execution starts by initializing the per-future metrics hash with `failed = 0` +and an empty `error_message`. + +Failure metadata is written for: + +- exceptions raised by the agent method +- Bedrock call failures +- request-processing failures with a known future ID +- forwarding failures +- result-callback failures + +The failure record is: + +```text +future:{future_id}:metrics.failed = 1 +future:{future_id}:metrics.error_message = str(exception) +``` + +`Future.value()` checks this metrics hash and raises the recorded error when +the metrics are available in the Redis instance being queried. + +For remote execution with separate Redis instances, the origin cannot see the +executor's metrics hash. It receives only the failure result string, so the +origin may return that string instead of raising the original `error_message`. + +Malformed JSON cannot be associated with a future because no reliable future ID +is available. Failures before a future ID exists have the same limitation. + +## Important callback ordering issue + +The executor currently sends the success or failure callback before it writes +all final metrics. In particular, fields such as these are written afterward: + +- `finished_at` +- `cpu_resource` +- `gpu_resource` +- `agent` +- `queue_time` + +Consequently, sending a metrics snapshot from the callback's current location +would produce an incomplete snapshot. The callback must be sent only after the +final metrics writes are complete. + +## Proposed future-state synchronization + +Treat the future ID as the logical identity, while explicitly synchronizing a +serialized snapshot between Redis instances. + +### Request path + +1. The origin creates `future:{future_id}`. +2. The origin sends the future state/request data and future ID to the executor. +3. The executor executes the request using that future ID. + +### Completion path + +1. The executor initializes and updates its local per-future metrics hash. +2. The executor writes the result or failure state. +3. The executor writes all final timing and resource metrics. +4. The executor reads the completed future state and metrics fields. +5. The executor sends the complete serialized snapshot to the origin. +6. The origin merges the returned fields into its own `future:{future_id}` hash. +7. `Future.value()` on the origin can read the synchronized failure fields and + raise the original `error_message`. + +A callback payload could contain: + +```json +{ + "future_id": "", + "future": { + "result": "", + "error": "" + }, + "metrics": { + "failed": 0, + "error_message": "", + "created_at": 0, + "finished_at": 0, + "cpu_resource": 0, + "gpu_resource": 0, + "agent": "..." + } +} +``` + +Alternatively, the origin can receive one flattened field map and merge all +fields into `future:{future_id}`. The important requirement is that the +metrics fields be explicitly transferred; matching key names across different +Redis instances does not synchronize them. + +## Data ownership + +After synchronization: + +- The executor keeps `future:{future_id}:metrics` locally for runtime polling + and local-controller reporting. +- The origin's `future:{future_id}` contains the result plus the synchronized + metrics needed by callers. +- The controller-level aggregate metrics hash remains local to each controller. +- Child and consumer bookkeeping should not be copied as part of the completed + future snapshot unless a separate requirement establishes ownership and merge + semantics for those sets. + +## Implementation considerations + +- Move success and failure callbacks until after final metrics writes. +- Extend the callback payload to include the completed metrics/state snapshot. +- Merge returned fields into the origin's `future:{future_id}` hash. +- Preserve the executor-local `future:{future_id}:metrics` hash. +- Define whether origin fields or executor fields win if the same field appears + in both snapshots. +- Ensure callback retries or duplicate callbacks are idempotent. +- Add tests for local execution and separate-origin/executor Redis instances. +- Verify that `Future.value()` raises the synchronized `error_message` for + remote failures. + +## Non-goals + +- Making identically named Redis keys automatically shared across instances. +- Moving aggregate controller metrics into the future record. +- Copying `children` or `consumers` bookkeeping without explicit merge rules. +- Raising failures from the local controller based on the metrics flag; failure + recording belongs in the controller, while failure surfacing belongs in + `Future.value()`. + diff --git a/examples/portfolio/agents/advisor_agent.py b/examples/portfolio/agents/advisor_agent.py index 035a87c..7ef8f82 100644 --- a/examples/portfolio/agents/advisor_agent.py +++ b/examples/portfolio/agents/advisor_agent.py @@ -3,7 +3,7 @@ # Final stage. Turns the computed portfolio metrics and risk figures into a # short, plain-English briefing using a small, cheap model on AWS Bedrock # (Converse API), called via ventis.llm.bedrock so token/cost telemetry gets -# recorded onto this execution's future::metrics hash. Configure +# recorded onto this execution's future: hash. Configure # with env vars: # BEDROCK_MODEL_ID (default: meta.llama3-8b-instruct-v1:0) # AWS_REGION (default: us-east-1) diff --git a/examples/portfolio/agents/intent_agent.py b/examples/portfolio/agents/intent_agent.py index 886ee14..e8f1dc9 100644 --- a/examples/portfolio/agents/intent_agent.py +++ b/examples/portfolio/agents/intent_agent.py @@ -9,7 +9,7 @@ # # Calls AWS Bedrock (Converse API) via ventis.llm.bedrock -- same pattern as # AdvisorAgent -- so token/cost telemetry gets recorded onto this execution's -# future::metrics hash. Configure with env vars: +# future: hash. Configure with env vars: # BEDROCK_MODEL_ID (default: meta.llama3-8b-instruct-v1:0) # AWS_REGION (default: us-east-1) # diff --git a/examples/text2sql/agents/vllm_agent.py b/examples/text2sql/agents/vllm_agent.py index cd48b7b..4a7a245 100644 --- a/examples/text2sql/agents/vllm_agent.py +++ b/examples/text2sql/agents/vllm_agent.py @@ -3,7 +3,7 @@ # LLM backend for SQL candidate generation, called remotely by # SQLGeneratorAgent. Calls AWS Bedrock (Converse API) via ventis.llm.bedrock # so token/cost telemetry gets recorded onto this execution's -# future::metrics hash — same pattern as +# future: hash — same pattern as # examples/portfolio/agents/advisor_agent.py. # Configure with env vars: # BEDROCK_MODEL_ID (default: meta.llama3-8b-instruct-v1:0) diff --git a/tests/test_error_propagation.py b/tests/test_error_propagation.py index a998cac..a29d2bf 100644 --- a/tests/test_error_propagation.py +++ b/tests/test_error_propagation.py @@ -34,6 +34,14 @@ def hset_multiple(self, name, mapping): def hget(self, name, field): return self.hashes.get(name, {}).get(field) + def hgetall(self, name): + return dict(self.hashes.get(name, {})) + + def hincrby(self, name, field, amount=1): + bucket = self.hashes.setdefault(name, {}) + bucket[field] = int(bucket.get(field, 0)) + amount + return bucket[field] + def _bind_failure_marker(controller): controller._mark_future_failed = lambda future_id, error, origin=None: ( @@ -95,7 +103,7 @@ def test_future_poll_redis_returns_result_when_error_is_absent(self): def test_future_value_raises_when_metrics_mark_it_failed(self): redis = _FakeRedis() redis.hset_multiple( - "future:future-1:metrics", + "future:future-1", {"failed": 1, "error_message": "agent exploded"}, ) future = SimpleNamespace( @@ -154,10 +162,10 @@ def test_write_result_persists_remote_error_as_terminal_failure(self): LocalControllerServicer.WriteResult(servicer, request, context) self.assertEqual( - redis.hget("future:future-1:metrics", "failed"), 1 + redis.hget("future:future-1", "failed"), 1 ) self.assertEqual( - redis.hget("future:future-1:metrics", "error_message"), + redis.hget("future:future-1", "error_message"), "remote exploded", ) @@ -173,7 +181,83 @@ def test_malformed_request_with_future_id_is_marked_failed(self): redis.hget("future:future-1", "error"), "Malformed request: missing service, function, or future_id", ) - self.assertEqual(redis.hget("future:future-1:metrics", "failed"), 1) + self.assertEqual(redis.hget("future:future-1", "failed"), 1) + + def test_cross_instance_failure_snapshot_merges_into_origin_and_raises(self): + """Simulate origin and executor on separate Redis instances: the + executor's completion callback must carry the full execution snapshot + so Future.value() on the origin raises the original error_message.""" + origin_redis = _FakeRedis() + executor_redis = _FakeRedis() + + origin_redis.hset_multiple( + "future:future-1", + {"id": "future-1", "service": "Greeter", "method": "greet", "result": ""}, + ) + + def boom(): + raise ValueError("executor exploded") + + stub = SimpleNamespace(WriteResult=MagicMock()) + callback_payloads = [] + + def capture_write_result(request): + callback_payloads.append(request.resonse) + + stub.WriteResult.side_effect = capture_write_result + + executor = SimpleNamespace( + redis=executor_redis, + agent=SimpleNamespace(greet=boom), + agent_name="Greeter", + agent_id="executor-agent", + _my_endpoint="executor:50051", + _metrics_key="controller:executor:50051:metrics", + _resolve_future_args=lambda args: args, + _get_remote_stub=lambda endpoint: stub, + ) + executor._mark_future_failed = lambda future_id, error, origin=None: ( + LocalController._mark_future_failed(executor, future_id, error, origin) + ) + executor._send_result_callback = lambda *a, **k: ( + LocalController._send_result_callback(executor, *a, **k) + ) + + LocalController._execute_locally( + executor, "Greeter", "greet", {}, "future-1", origin="origin:50051" + ) + + # Feed the captured callback into the origin's WriteResult receiver. + origin_servicer = SimpleNamespace(redis=origin_redis) + for payload in callback_payloads: + request = local_controler_pb2.JsonResponse(resonse=payload) + context = SimpleNamespace(peer=lambda: "executor:50051") + LocalControllerServicer.WriteResult(origin_servicer, request, context) + + self.assertEqual(origin_redis.hget("future:future-1", "failed"), 1) + self.assertEqual( + origin_redis.hget("future:future-1", "error_message"), + "executor exploded", + ) + self.assertIn("cpu_resource", origin_redis.hashes["future:future-1"]) + self.assertIn("finished_at", origin_redis.hashes["future:future-1"]) + self.assertEqual(origin_redis.hget("future:future-1", "agent"), "executor-agent") + + origin_future = SimpleNamespace( + redis=origin_redis, + _key=lambda: "future:future-1", + _poll_redis=lambda: Future._poll_redis(origin_future), + id="future-1", + result=None, + ) + with self.assertRaisesRegex(RuntimeError, "executor exploded"): + Future.value(origin_future) + + # Executor's own local copy is untouched by the origin-side merge. + self.assertEqual( + executor_redis.hget("future:future-1", "error_message"), + "executor exploded", + ) if __name__ == "__main__": diff --git a/tests/test_local_controller_metrics.py b/tests/test_local_controller_metrics.py index 90491a7..6e6ce59 100644 --- a/tests/test_local_controller_metrics.py +++ b/tests/test_local_controller_metrics.py @@ -154,12 +154,13 @@ def test_execute_locally_writes_gpu_resource_to_future_hash(self): controller, "Greeter", "greet", {"name": "world"}, "future-1" ) - self.assertEqual(redis.hget("future:future-1:metrics", "gpu_resource"), 17.5) + self.assertEqual(redis.hget("future:future-1", "gpu_resource"), 17.5) self.assertEqual(redis.hget("future:future-1", "result"), "hello world") self.assertEqual( - redis.hget("future:future-1:metrics", "agent"), + redis.hget("future:future-1", "agent"), "1f2e3d4c5b6a7988fedcba9876543210", ) + self.assertNotIn("future:future-1:metrics", redis.hashes) self.assertEqual( redis.hget("controller:localhost:50051:metrics", "requests_served"), 1 ) @@ -189,13 +190,14 @@ def boom(name): ) self.assertEqual(redis.hget("future:future-2", "error"), "nope") - self.assertIsNone(redis.hget("future:future-2", "result")) + self.assertEqual(redis.hget("future:future-2", "result"), "") self.assertEqual( - redis.hget("future:future-2:metrics", "failed"), 1 + redis.hget("future:future-2", "failed"), 1 ) self.assertEqual( - redis.hget("future:future-2:metrics", "error_message"), "nope" + redis.hget("future:future-2", "error_message"), "nope" ) + self.assertNotIn("future:future-2:metrics", redis.hashes) self.assertEqual( redis.hget("controller:localhost:50051:metrics", "requests_served"), 1 ) @@ -224,11 +226,12 @@ def test_execute_locally_marks_missing_agent_as_failed(self): self.assertEqual( redis.hget("future:future-3", "error"), "No agent loaded" ) - self.assertEqual(redis.hget("future:future-3:metrics", "failed"), 1) + self.assertEqual(redis.hget("future:future-3", "failed"), 1) self.assertEqual( - redis.hget("future:future-3:metrics", "error_message"), + redis.hget("future:future-3", "error_message"), "No agent loaded", ) + self.assertNotIn("future:future-3:metrics", redis.hashes) def test_remote_execution_failure_sends_error_callback(self): redis = _FakeRedis() @@ -261,17 +264,56 @@ def boom(): ) self.assertEqual(redis.hget("future:future-4", "error"), "remote nope") - payload = stub.WriteResult.call_args.args[0].resonse - self.assertEqual( - json.loads(payload), - { - "future_id": "future-4", - "result": None, - "failed": 1, - "error_message": "remote nope", - }, + payload = json.loads(stub.WriteResult.call_args.args[0].resonse) + self.assertEqual(payload["future_id"], "future-4") + self.assertIsNone(payload["result"]) + self.assertEqual(payload["failed"], 1) + self.assertEqual(payload["error_message"], "remote nope") + # The callback fires only after the finally block writes final metrics, + # so the snapshot sent to origin carries the full execution record. + self.assertEqual(payload["agent"], "agent-1") + self.assertIn("finished_at", payload) + self.assertIn("cpu_resource", payload) + self.assertEqual(payload["gpu_resource"], 0.0) + + def test_callback_fires_once_and_only_after_final_metrics_written(self): + redis = _FakeRedis() + seen_at_callback_time = {} + + def spy_send_result_callback(origin, future_id, result=None, failed=0, error_message=""): + seen_at_callback_time["snapshot"] = dict(redis.hashes.get(f"future:{future_id}", {})) + seen_at_callback_time["calls"] = seen_at_callback_time.get("calls", 0) + 1 + + controller = SimpleNamespace( + redis=redis, + agent=SimpleNamespace(greet=lambda name: f"hello {name}"), + agent_name="Greeter", + agent_id="agent-1", + _my_endpoint="target:50051", + _metrics_key="controller:target:50051:metrics", + _resolve_future_args=lambda args: args, + _mark_future_failed=lambda future_id, error, origin=None: None, + _send_result_callback=spy_send_result_callback, ) + with patch( + "ventis.controller.local_controller.read_gpu_percent", return_value=0.0 + ): + LocalController._execute_locally( + controller, + "Greeter", + "greet", + {"name": "world"}, + "future-5", + origin="origin:50051", + ) + + self.assertEqual(seen_at_callback_time["calls"], 1) + self.assertIn("finished_at", seen_at_callback_time["snapshot"]) + self.assertIn("cpu_resource", seen_at_callback_time["snapshot"]) + self.assertIn("gpu_resource", seen_at_callback_time["snapshot"]) + self.assertIn("agent", seen_at_callback_time["snapshot"]) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_runtime_sqlalchemy.py b/tests/test_runtime_sqlalchemy.py index c2bca59..8ac618e 100644 --- a/tests/test_runtime_sqlalchemy.py +++ b/tests/test_runtime_sqlalchemy.py @@ -59,7 +59,7 @@ def tearDown(self): def test_pull_and_upsert(self): redis = _FakeRedis( { - "future:abc:metrics": { + "future:abc": { "id": "abc", "request_id": "req1", "agent": "1f2e3d4c5b6a7988fedcba9876543210", @@ -111,7 +111,7 @@ def test_pull_and_upsert(self): def test_parent_id_defaults_to_none_when_absent(self): redis = _FakeRedis( { - "future:solo:metrics": { + "future:solo": { "id": "solo", "request_id": "req9", "agent": "1f2e3d4c5b6a7988fedcba9876543210", @@ -135,7 +135,7 @@ def test_parent_id_defaults_to_none_when_absent(self): def test_observed_cpu_and_gpu_are_recorded(self): redis = _FakeRedis( { - "future:xyz:metrics": { + "future:xyz": { "id": "xyz", "request_id": "req2", "agent": "aabbccddeeff00112233445566778899", @@ -165,7 +165,7 @@ def test_observed_cpu_and_gpu_are_recorded(self): def test_llm_token_and_error_fields_are_recorded(self): redis = _FakeRedis( { - "future:llm1:metrics": { + "future:llm1": { "id": "llm1", "request_id": "req4", "agent": "1f2e3d4c5b6a7988fedcba9876543210", @@ -201,7 +201,7 @@ def test_llm_token_and_error_fields_are_recorded(self): def test_llm_token_and_error_fields_default_when_absent(self): redis = _FakeRedis( { - "future:nollm:metrics": { + "future:nollm": { "id": "nollm", "request_id": "req5", "agent": "00112233445566778899aabbccddeeff", @@ -232,7 +232,7 @@ def test_llm_token_and_error_fields_default_when_absent(self): def test_cpu_and_gpu_default_to_zero_when_not_observed(self): redis = _FakeRedis( { - "future:noop:metrics": { + "future:noop": { "id": "noop", "request_id": "req3", "agent": "00112233445566778899aabbccddeeff", @@ -258,7 +258,7 @@ def test_cpu_and_gpu_default_to_zero_when_not_observed(self): def test_total_cost_computed_from_model_token_pricing(self): redis = _FakeRedis( { - "future:cost1:metrics": { + "future:cost1": { "id": "cost1", "request_id": "req6", "agent": "1f2e3d4c5b6a7988fedcba9876543210", @@ -289,7 +289,7 @@ def test_total_cost_computed_from_model_token_pricing(self): def test_total_cost_defaults_to_zero_for_unknown_model(self): redis = _FakeRedis( { - "future:cost2:metrics": { + "future:cost2": { "id": "cost2", "request_id": "req7", "agent": "1f2e3d4c5b6a7988fedcba9876543210", @@ -313,7 +313,7 @@ def test_total_cost_defaults_to_zero_for_unknown_model(self): def test_total_cost_includes_server_cost_from_agent_instance_type(self): redis = _FakeRedis( { - "future:cost3:metrics": { + "future:cost3": { "id": "cost3", "request_id": "req8", "agent": "ec2agent1", @@ -338,7 +338,7 @@ def test_total_cost_includes_server_cost_from_agent_instance_type(self): def test_demo_cost_multipliers_scale_costs_independently_and_warn(self): redis = _FakeRedis( { - "future:cost4:metrics": { + "future:cost4": { "id": "cost4", "request_id": "req9", "agent": "ec2agent2", diff --git a/ventis/controller/local_controller.py b/ventis/controller/local_controller.py index 9f53a3e..1b6eba5 100644 --- a/ventis/controller/local_controller.py +++ b/ventis/controller/local_controller.py @@ -18,10 +18,12 @@ try: from ventis.controller.local_controller_frontend import start_server from ventis.controller.utils.gpu_metrics import read_gpu_percent + from ventis.controller.utils.future_schema import snapshot_execution_fields from ventis.utils.redis_client import RedisClient except ImportError: from gpu_metrics import read_gpu_percent from local_controller_frontend import start_server + from future_schema import snapshot_execution_fields from redis_client import RedisClient # Add local generated grpc_stubs to path (Docker context copies them directly to /app) @@ -308,10 +310,9 @@ def _mark_future_failed(self, future_id, error, origin=None): return error_message = str(error) or "Unknown error" - self.redis.hset(f"future:{future_id}", "error", error_message) self.redis.hset_multiple( - f"future:{future_id}:metrics", - {"failed": 1, "error_message": error_message}, + f"future:{future_id}", + {"error": error_message, "failed": 1, "error_message": error_message}, ) if origin and origin != self._my_endpoint: @@ -489,12 +490,10 @@ def _resolve_future_args(self, args, poll_interval=0.01, timeout=300): error = self.redis.hget(future_key, "error") if error: raise RuntimeError(error) - failed = self.redis.hget(f"future:{value}:metrics", "failed") + failed = self.redis.hget(future_key, "failed") if str(failed) == "1": raise RuntimeError( - self.redis.hget( - f"future:{value}:metrics", "error_message" - ) + self.redis.hget(future_key, "error_message") or "Unknown error" ) # print("Waiting for result for future next iteration %s", value) @@ -529,12 +528,13 @@ def _execute_locally( wall_start = time.time() thread_cpu_start = time.thread_time() - # Write a complete, self-contained metrics record for this execution step - # entirely to this node's own Redis -- unlike future:{future_id} (created on - # the calling node), this is never split across two Redis instances, since - # both the "start" and "finish" writes below happen on the same node. + # Write a complete, self-contained execution record for this step entirely + # to this node's own Redis. When this node is the origin, this overlaps + # with the identity fields Future.__init__ already wrote; when it's a + # remote executor, these fields live only here until the completion + # callback merges them back into the origin's future:{future_id}. self.redis.hset_multiple( - f"future:{future_id}:metrics", + f"future:{future_id}", { "id": future_id, "request_id": request_id or "", @@ -568,6 +568,8 @@ def _execute_locally( self.redis.hincrby(self._metrics_key, "requests_served", 1) + succeeded = False + serialized = None try: # Resolve any Future IDs in the args before executing args = self._resolve_future_args(args) @@ -585,16 +587,8 @@ def _execute_locally( # Write result to local Redis self.redis.hset(f"future:{future_id}", "result", serialized) - - # If the request came from another node, send result back to origin - if origin and origin != self._my_endpoint: - self._send_result_callback( - origin, - future_id, - result=serialized, - failed=0, - error_message="", - ) + self.redis.hset(f"future:{future_id}", "failed", 0) + succeeded = True logger.info( "Completed %s.%s (future=%s) -> %s", @@ -603,11 +597,12 @@ def _execute_locally( future_id, serialized, ) - self.redis.hset(f"future:{future_id}:metrics", "failed", 0) except Exception as e: logger.error("Failed to execute %s.%s: %s", service, function, e) - self._mark_future_failed(future_id, e, origin) + # Persist the failure locally now; the callback to origin (if any) + # is sent below, after the finally block writes final metrics. + self._mark_future_failed(future_id, e) self.redis.hincrby(self._metrics_key, "full_failures", 1) finally: wall_end = time.time() @@ -619,7 +614,7 @@ def _execute_locally( gpu_percent = read_gpu_percent() self.redis.hset_multiple( - f"future:{future_id}:metrics", + f"future:{future_id}", { "finished_at": wall_end, "cpu_resource": cpu_percent, @@ -632,6 +627,20 @@ def _execute_locally( ), }, ) + + # Send the completion callback only now that every final metric + # has been written, so the snapshot sent to origin is complete. + if origin and origin != self._my_endpoint: + if succeeded: + self._send_result_callback( + origin, future_id, result=serialized, failed=0, error_message="" + ) + else: + error_message = self.redis.hget(f"future:{future_id}", "error_message") + self._send_result_callback( + origin, future_id, failed=1, error_message=error_message or "" + ) + ventis_context.set_current_future_id(parent or "") # ------------------------------------------------------------------ # @@ -668,7 +677,7 @@ def _forward_request(self, endpoint, data): def _send_result_callback( self, origin, future_id, result=None, failed=0, error_message="" ): - """Send a result and its failure metadata to the originating controller.""" + """Send the future's full execution snapshot to the originating controller.""" if not result: logger.warning( "Agent '%s' is sending an empty/None result for future %s to origin %s, result: %s", @@ -679,12 +688,16 @@ def _send_result_callback( ) stub = self._get_remote_stub(origin) - payload = json.dumps({ - "future_id": future_id, - "result": result, - "failed": int(bool(failed)), - "error_message": str(error_message or ""), - }) + snapshot = snapshot_execution_fields(self.redis, future_id) + snapshot.update( + { + "future_id": future_id, + "result": result, + "failed": int(bool(failed)), + "error_message": str(error_message or ""), + } + ) + payload = json.dumps(snapshot) logger.info("Payload: Future %s,Sent %s ", future_id, payload) request = local_controler_pb2.JsonResponse(resonse=payload) try: diff --git a/ventis/controller/local_controller_frontend.py b/ventis/controller/local_controller_frontend.py index a6e478d..d876503 100644 --- a/ventis/controller/local_controller_frontend.py +++ b/ventis/controller/local_controller_frontend.py @@ -18,6 +18,11 @@ import local_controler_pb2 import local_controler_pb2_grpc +try: + from ventis.controller.utils.future_schema import merge_execution_snapshot +except ImportError: + from future_schema import merge_execution_snapshot + logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) @@ -51,7 +56,6 @@ def WriteResult(self, request, context): future_id = data.get("future_id") result = data.get("result") failed = int(bool(data.get("failed", 0))) - error_message = str(data.get("error_message") or "") logger.info( f"WriteResult: received result for future {future_id}: {result}" @@ -62,14 +66,10 @@ def WriteResult(self, request, context): ) if future_id: - self.redis.hset_multiple( - f"future:{future_id}:metrics", - {"failed": failed, "error_message": error_message}, - ) + merge_execution_snapshot(self.redis, future_id, data) if failed: logger.info("WriteResult: wrote error for future %s", future_id) elif result is not None: - self.redis.hset(f"future:{future_id}", "result", result) logger.info( "WriteResult: wrote result for future %s, result %s", future_id, @@ -121,7 +121,6 @@ def _cleanup_request(self, request_id): f"future:{fid}", f"future:{fid}:children", f"future:{fid}:consumers", - f"future:{fid}:metrics", ] ) self.redis.delete(*keys_to_delete) diff --git a/ventis/controller/utils/future_schema.py b/ventis/controller/utils/future_schema.py new file mode 100644 index 0000000..c035380 --- /dev/null +++ b/ventis/controller/utils/future_schema.py @@ -0,0 +1,60 @@ +# Shared schema for the consolidated future:{future_id} Redis hash. +# +# IDENTITY_FIELDS are written once by the future's creator (the origin) and +# must never be overwritten by a remote execution snapshot. EXECUTION_FIELDS +# are produced during/after execution and are always last-write-wins when +# merged in from a remote node's callback -- the origin has no independent +# opinion about e.g. cpu_resource or finished_at. + +IDENTITY_FIELDS = [ + "id", + "request_id", + "parent", + "service", + "method", + "args", + "created_at", +] + +EXECUTION_FIELDS = [ + "result", + "error", + "failed", + "error_message", + "finished_at", + "cpu_resource", + "gpu_resource", + "agent", + "queue_time", + "model", + "input_token_count", + "output_token_count", + "token_count", + "errors", + "input_cache_tokens", + "input_cache_write_tokens", +] + + +def snapshot_execution_fields(redis, future_id): + """Read the execution-related fields currently stored for a future.""" + data = redis.hgetall(f"future:{future_id}") + return {k: v for k, v in data.items() if k in EXECUTION_FIELDS} + + +def merge_execution_snapshot(redis, future_id, snapshot): + """Merge a remote execution snapshot into this node's future:{future_id} hash. + + Only EXECUTION_FIELDS are applied -- identity fields in the incoming + snapshot (if any) are ignored so a remote node can never clobber the + origin's own record of what the future is. None values are dropped + rather than merged, so an absent/unset field on the sender doesn't + stomp a value already present on the receiver. + """ + filtered = { + k: v + for k, v in snapshot.items() + if k in EXECUTION_FIELDS and v is not None + } + if filtered: + redis.hset_multiple(f"future:{future_id}", filtered) diff --git a/ventis/controller/utils/sqlalchemy.py b/ventis/controller/utils/sqlalchemy.py index 2220880..b08a91d 100644 --- a/ventis/controller/utils/sqlalchemy.py +++ b/ventis/controller/utils/sqlalchemy.py @@ -157,12 +157,16 @@ def _get_engine(database_url): def pull_runtime_information(redis_client): """Scan node Redis for future execution metrics. - Each future's metrics live at future:{future_id}:metrics, written entirely by - whichever node executed it -- never split across nodes like the main - future:{future_id} key (used for the result hand-off) can be. + Each future's identity and execution metrics both live at future:{future_id} + now that the two have been consolidated into a single hash. Sibling keys + (future:{future_id}:children, future:{future_id}:consumers) share the same + prefix but are bookkeeping, not metrics -- skip them explicitly since Redis + glob patterns can't express "no suffix". """ rows = [] - for key in redis_client.scan_keys("future:*:metrics"): + for key in redis_client.scan_keys("future:*"): + if key.endswith(":children") or key.endswith(":consumers"): + continue data = redis_client.hgetall(key) if data: data["future_id"] = data.get("id") or key.split(":")[1] @@ -192,11 +196,9 @@ def send_runtime_information( session_id = raw.get("request_id") if not session_id: continue - # A future without finished_at is still executing. Now that metrics live - # entirely on the executing node (future:{future_id}:metrics is only ever - # written by the one process that runs it), "incomplete" genuinely means - # "still running" -- skip it and let a later poll, once it has actually - # finished, write the real measurements instead. + # A future without finished_at is still executing -- skip it and let a + # later poll, once it has actually finished, write the real measurements + # instead. if not raw.get("finished_at"): continue start = float(raw.get("created_at") or 0) diff --git a/ventis/future.py b/ventis/future.py index 598bc4f..6335e04 100644 --- a/ventis/future.py +++ b/ventis/future.py @@ -118,8 +118,8 @@ def _submit_request(self): ) except Exception as e: logger.error("gRPC call failed for %s.%s: %s", self.service, self.method, e) - self.redis.hset(f"future:{self.id}", "error", str(e)) - self.redis.hset_multiple(f"future:{self.id}:metrics", { + self.redis.hset_multiple(f"future:{self.id}", { + "error": str(e), "failed": 1, "error_message": str(e), }) @@ -153,10 +153,10 @@ def value(self, timeout=None): Returns immediately if the result is already available locally. Polls Redis periodically to check for computed results. """ - failed = self.redis.hget(f"future:{self.id}:metrics", "failed") + failed = self.redis.hget(self._key(), "failed") if str(failed) == "1": raise RuntimeError( - self.redis.hget(f"future:{self.id}:metrics", "error_message") + self.redis.hget(self._key(), "error_message") or "Unknown error" ) diff --git a/ventis/llm/bedrock.py b/ventis/llm/bedrock.py index b843234..1c5f7bb 100644 --- a/ventis/llm/bedrock.py +++ b/ventis/llm/bedrock.py @@ -15,7 +15,7 @@ def call_bedrock(model_id: str, messages: list, inference_config: dict, region: str = "us-east-1") -> dict: """Call Bedrock's converse() API and log token/error telemetry onto the - currently executing future's metrics hash (future::metrics).""" + currently executing future's hash (future:).""" import boto3 client = boto3.client("bedrock-runtime", region_name=region) @@ -36,7 +36,7 @@ def call_bedrock(model_id: str, messages: list, inference_config: dict, region: if metrics_key: _redis.hincrby(metrics_key, "error_count", 1) if future_id: - _redis.hset_multiple(f"future:{future_id}:metrics", { + _redis.hset_multiple(f"future:{future_id}", { "failed": 1, "error_message": str(e), }) @@ -44,7 +44,7 @@ def call_bedrock(model_id: str, messages: list, inference_config: dict, region: finally: if future_id: usage = (response or {}).get("usage", {}) - _redis.hset_multiple(f"future:{future_id}:metrics", { + _redis.hset_multiple(f"future:{future_id}", { "model": model_id, "input_token_count": str(usage.get("inputTokens", "")), "output_token_count": str(usage.get("outputTokens", "")), diff --git a/ventis/stub_generator.py b/ventis/stub_generator.py index 64ef1fc..f072432 100644 --- a/ventis/stub_generator.py +++ b/ventis/stub_generator.py @@ -319,6 +319,10 @@ def generate_docker( os.path.join(script_dir, "controller", "utils", "gpu_metrics.py"), "gpu_metrics.py", ), + ( + os.path.join(script_dir, "controller", "utils", "future_schema.py"), + "future_schema.py", + ), (os.path.join(script_dir, "llm", "bedrock.py"), "bedrock.py"), ] @@ -435,6 +439,10 @@ def generate_workflow_docker( os.path.join(script_dir, "controller", "utils", "gpu_metrics.py"), "gpu_metrics.py", ), + ( + os.path.join(script_dir, "controller", "utils", "future_schema.py"), + "future_schema.py", + ), ( os.path.join(script_dir, "controller", "utils", "session_store.py"), "session_store.py", From 1e617e259991b85f8e5196d4af0bd0661fbb1c70 Mon Sep 17 00:00:00 2001 From: Saaketh Sodanapalli Date: Wed, 5 Aug 2026 15:05:19 -0700 Subject: [PATCH 7/8] removed duplicate error fields slimmed down useless function removed useless file removed some useless comments --- FUTURE_PLANS.md | 212 ------------------ tests/test_error_propagation.py | 14 +- tests/test_local_controller_metrics.py | 13 +- ventis/FUTURE_SCHEMA.md | 34 +++ ventis/controller/local_controller.py | 26 +-- .../controller/local_controller_frontend.py | 8 +- ventis/controller/utils/future_schema.py | 60 ----- ventis/controller/utils/telemetry_logging.py | 5 - ventis/future.py | 3 +- ventis/llm/bedrock.py | 10 +- ventis/stub_generator.py | 6 +- 11 files changed, 62 insertions(+), 329 deletions(-) delete mode 100644 FUTURE_PLANS.md create mode 100644 ventis/FUTURE_SCHEMA.md delete mode 100644 ventis/controller/utils/future_schema.py diff --git a/FUTURE_PLANS.md b/FUTURE_PLANS.md deleted file mode 100644 index 746ba73..0000000 --- a/FUTURE_PLANS.md +++ /dev/null @@ -1,212 +0,0 @@ -# Future State Synchronization Plan - -## Purpose - -Ensure that the origin controller's `future:{future_id}` record contains the -complete future state, including execution metrics and failure details, while -preserving the executor-local metrics used by the local controller and runtime -reporting. - -## Current architecture - -The origin creates a future record in its Redis instance: - -```text -future:{future_id} -``` - -When a request is forwarded to another EC2 instance, the executor receives the -future ID and request data, not the origin's Redis hash or a live Redis object. -Because Redis keys are scoped to a Redis server, the executor's: - -```text -future:{future_id} -``` - -is a different Redis hash from the origin's identically named key. - -The executor also creates: - -```text -future:{future_id}:metrics -``` - -This hash contains execution-specific data, including: - -- `id` -- `request_id` -- `service` -- `method` -- `args` -- `created_at` -- `finished_at` -- `failed` -- `error_message` -- `cpu_resource` -- `gpu_resource` -- `agent` -- queue timing -- LLM model and token usage fields, when applicable - -The executor additionally updates its controller-level metrics hash: - -```text -controller:{host}:{port}:metrics -``` - -That hash contains aggregate controller data such as requests served and full -failures. It is separate from the per-future metrics hash and should remain so. - -## Current result handoff - -The executor currently sends this callback payload to the origin: - -```json -{ - "future_id": "", - "result": "" -} -``` - -The callback is created in `LocalController._send_result_callback()` and -received by `WriteResult()` in the local controller frontend. The origin writes -the received result into its local `future:{future_id}` hash. - -The callback currently does not send: - -- `failed` -- `error_message` -- the `future:{future_id}:metrics` hash -- CPU/GPU/timing fields -- LLM usage fields - -Therefore, the executor's metrics remain only in the executor's Redis instance. - -## Current failure behavior - -Execution starts by initializing the per-future metrics hash with `failed = 0` -and an empty `error_message`. - -Failure metadata is written for: - -- exceptions raised by the agent method -- Bedrock call failures -- request-processing failures with a known future ID -- forwarding failures -- result-callback failures - -The failure record is: - -```text -future:{future_id}:metrics.failed = 1 -future:{future_id}:metrics.error_message = str(exception) -``` - -`Future.value()` checks this metrics hash and raises the recorded error when -the metrics are available in the Redis instance being queried. - -For remote execution with separate Redis instances, the origin cannot see the -executor's metrics hash. It receives only the failure result string, so the -origin may return that string instead of raising the original `error_message`. - -Malformed JSON cannot be associated with a future because no reliable future ID -is available. Failures before a future ID exists have the same limitation. - -## Important callback ordering issue - -The executor currently sends the success or failure callback before it writes -all final metrics. In particular, fields such as these are written afterward: - -- `finished_at` -- `cpu_resource` -- `gpu_resource` -- `agent` -- `queue_time` - -Consequently, sending a metrics snapshot from the callback's current location -would produce an incomplete snapshot. The callback must be sent only after the -final metrics writes are complete. - -## Proposed future-state synchronization - -Treat the future ID as the logical identity, while explicitly synchronizing a -serialized snapshot between Redis instances. - -### Request path - -1. The origin creates `future:{future_id}`. -2. The origin sends the future state/request data and future ID to the executor. -3. The executor executes the request using that future ID. - -### Completion path - -1. The executor initializes and updates its local per-future metrics hash. -2. The executor writes the result or failure state. -3. The executor writes all final timing and resource metrics. -4. The executor reads the completed future state and metrics fields. -5. The executor sends the complete serialized snapshot to the origin. -6. The origin merges the returned fields into its own `future:{future_id}` hash. -7. `Future.value()` on the origin can read the synchronized failure fields and - raise the original `error_message`. - -A callback payload could contain: - -```json -{ - "future_id": "", - "future": { - "result": "", - "error": "" - }, - "metrics": { - "failed": 0, - "error_message": "", - "created_at": 0, - "finished_at": 0, - "cpu_resource": 0, - "gpu_resource": 0, - "agent": "..." - } -} -``` - -Alternatively, the origin can receive one flattened field map and merge all -fields into `future:{future_id}`. The important requirement is that the -metrics fields be explicitly transferred; matching key names across different -Redis instances does not synchronize them. - -## Data ownership - -After synchronization: - -- The executor keeps `future:{future_id}:metrics` locally for runtime polling - and local-controller reporting. -- The origin's `future:{future_id}` contains the result plus the synchronized - metrics needed by callers. -- The controller-level aggregate metrics hash remains local to each controller. -- Child and consumer bookkeeping should not be copied as part of the completed - future snapshot unless a separate requirement establishes ownership and merge - semantics for those sets. - -## Implementation considerations - -- Move success and failure callbacks until after final metrics writes. -- Extend the callback payload to include the completed metrics/state snapshot. -- Merge returned fields into the origin's `future:{future_id}` hash. -- Preserve the executor-local `future:{future_id}:metrics` hash. -- Define whether origin fields or executor fields win if the same field appears - in both snapshots. -- Ensure callback retries or duplicate callbacks are idempotent. -- Add tests for local execution and separate-origin/executor Redis instances. -- Verify that `Future.value()` raises the synchronized `error_message` for - remote failures. - -## Non-goals - -- Making identically named Redis keys automatically shared across instances. -- Moving aggregate controller metrics into the future record. -- Copying `children` or `consumers` bookkeeping without explicit merge rules. -- Raising failures from the local controller based on the metrics flag; failure - recording belongs in the controller, while failure surfacing belongs in - `Future.value()`. - diff --git a/tests/test_error_propagation.py b/tests/test_error_propagation.py index a29d2bf..82da262 100644 --- a/tests/test_error_propagation.py +++ b/tests/test_error_propagation.py @@ -104,7 +104,7 @@ def test_future_value_raises_when_metrics_mark_it_failed(self): redis = _FakeRedis() redis.hset_multiple( "future:future-1", - {"failed": 1, "error_message": "agent exploded"}, + {"failed": 1, "error": "agent exploded"}, ) future = SimpleNamespace( redis=redis, @@ -139,9 +139,9 @@ def test_result_callback_sends_error_separately_from_result(self): json.loads(payload), { "future_id": "future-1", - "result": None, + "result": "", "failed": 1, - "error_message": "agent exploded", + "error": "agent exploded", }, ) @@ -153,7 +153,7 @@ def test_write_result_persists_remote_error_as_terminal_failure(self): { "future_id": "future-1", "failed": 1, - "error_message": "remote exploded", + "error": "remote exploded", } ) ) @@ -165,7 +165,7 @@ def test_write_result_persists_remote_error_as_terminal_failure(self): redis.hget("future:future-1", "failed"), 1 ) self.assertEqual( - redis.hget("future:future-1", "error_message"), + redis.hget("future:future-1", "error"), "remote exploded", ) @@ -236,7 +236,7 @@ def capture_write_result(request): self.assertEqual(origin_redis.hget("future:future-1", "failed"), 1) self.assertEqual( - origin_redis.hget("future:future-1", "error_message"), + origin_redis.hget("future:future-1", "error"), "executor exploded", ) self.assertIn("cpu_resource", origin_redis.hashes["future:future-1"]) @@ -255,7 +255,7 @@ def capture_write_result(request): # Executor's own local copy is untouched by the origin-side merge. self.assertEqual( - executor_redis.hget("future:future-1", "error_message"), + executor_redis.hget("future:future-1", "error"), "executor exploded", ) diff --git a/tests/test_local_controller_metrics.py b/tests/test_local_controller_metrics.py index 6e6ce59..6dbe332 100644 --- a/tests/test_local_controller_metrics.py +++ b/tests/test_local_controller_metrics.py @@ -26,7 +26,7 @@ def _bind_failure_marker(controller): LocalController._mark_future_failed(controller, future_id, error, origin) ) controller._send_result_callback = ( - lambda origin, future_id, result=None, failed=0, error_message="": LocalController._send_result_callback( + lambda origin, future_id, result="", failed=0, error_message="": LocalController._send_result_callback( controller, origin, future_id, result, failed, error_message ) ) @@ -194,9 +194,6 @@ def boom(name): self.assertEqual( redis.hget("future:future-2", "failed"), 1 ) - self.assertEqual( - redis.hget("future:future-2", "error_message"), "nope" - ) self.assertNotIn("future:future-2:metrics", redis.hashes) self.assertEqual( redis.hget("controller:localhost:50051:metrics", "requests_served"), 1 @@ -227,10 +224,6 @@ def test_execute_locally_marks_missing_agent_as_failed(self): redis.hget("future:future-3", "error"), "No agent loaded" ) self.assertEqual(redis.hget("future:future-3", "failed"), 1) - self.assertEqual( - redis.hget("future:future-3", "error_message"), - "No agent loaded", - ) self.assertNotIn("future:future-3:metrics", redis.hashes) def test_remote_execution_failure_sends_error_callback(self): @@ -266,9 +259,9 @@ def boom(): self.assertEqual(redis.hget("future:future-4", "error"), "remote nope") payload = json.loads(stub.WriteResult.call_args.args[0].resonse) self.assertEqual(payload["future_id"], "future-4") - self.assertIsNone(payload["result"]) + self.assertEqual(payload["result"], "") self.assertEqual(payload["failed"], 1) - self.assertEqual(payload["error_message"], "remote nope") + self.assertEqual(payload["error"], "remote nope") # The callback fires only after the finally block writes final metrics, # so the snapshot sent to origin carries the full execution record. self.assertEqual(payload["agent"], "agent-1") diff --git a/ventis/FUTURE_SCHEMA.md b/ventis/FUTURE_SCHEMA.md new file mode 100644 index 0000000..360b1af --- /dev/null +++ b/ventis/FUTURE_SCHEMA.md @@ -0,0 +1,34 @@ +# `future:{future_id}` Redis hash schema + +Both directions (origin -> executor request, executor -> origin completion +callback) send the future's full hash. Whichever node last wrote a field +wins for most fields (e.g. `args` as re-serialized by the executor) -- the +one exception is `created_at`, which only the origin ever writes, so it +always reflects the future's true submission time. + +Fields currently written into `future:{future_id}`, and where: + +| Field | Written by | +|----------------------------|------------| +| `id` | `future.py` (`Future.__init__`), `local_controller.py` (`_execute_locally`) | +| `request_id` | `future.py`, `local_controller.py` | +| `parent` | `future.py`, `local_controller.py` | +| `service` | `future.py`, `local_controller.py` | +| `method` | `future.py`, `local_controller.py` | +| `args` | `future.py`, `local_controller.py` (json-encoded) | +| `created_at` | `future.py` only (origin submission time) | +| `result` | `future.py`, `local_controller.py` | +| `failed` | `future.py`, `local_controller.py` | +| `error` | `future.py` (`_submit_request`), `local_controller.py` (`_mark_future_failed`) -- the sole failure-message field; `bedrock.py` deliberately never writes it | +| `finished_at` | `local_controller.py` (`_execute_locally` finally block) | +| `cpu_resource` | `local_controller.py` | +| `gpu_resource` | `local_controller.py` | +| `agent` | `local_controller.py` (agent_id that executed this step) | +| `queue_time` | `local_controller.py` (only when `submitted_at` is known) | +| `model` | `llm/bedrock.py` (`call_bedrock`) | +| `input_token_count` | `llm/bedrock.py` | +| `output_token_count` | `llm/bedrock.py` | +| `token_count` | `llm/bedrock.py` | +| `errors` | `llm/bedrock.py` (Bedrock call error count) | +| `input_cache_tokens` | `llm/bedrock.py` | +| `input_cache_write_tokens` | `llm/bedrock.py` | diff --git a/ventis/controller/local_controller.py b/ventis/controller/local_controller.py index 1b6eba5..5e40580 100644 --- a/ventis/controller/local_controller.py +++ b/ventis/controller/local_controller.py @@ -18,12 +18,10 @@ try: from ventis.controller.local_controller_frontend import start_server from ventis.controller.utils.gpu_metrics import read_gpu_percent - from ventis.controller.utils.future_schema import snapshot_execution_fields from ventis.utils.redis_client import RedisClient except ImportError: from gpu_metrics import read_gpu_percent from local_controller_frontend import start_server - from future_schema import snapshot_execution_fields from redis_client import RedisClient # Add local generated grpc_stubs to path (Docker context copies them directly to /app) @@ -312,7 +310,7 @@ def _mark_future_failed(self, future_id, error, origin=None): error_message = str(error) or "Unknown error" self.redis.hset_multiple( f"future:{future_id}", - {"error": error_message, "failed": 1, "error_message": error_message}, + {"error": error_message, "failed": 1}, ) if origin and origin != self._my_endpoint: @@ -493,7 +491,7 @@ def _resolve_future_args(self, args, poll_interval=0.01, timeout=300): failed = self.redis.hget(future_key, "failed") if str(failed) == "1": raise RuntimeError( - self.redis.hget(future_key, "error_message") + self.redis.hget(future_key, "error") or "Unknown error" ) # print("Waiting for result for future next iteration %s", value) @@ -529,10 +527,7 @@ def _execute_locally( thread_cpu_start = time.thread_time() # Write a complete, self-contained execution record for this step entirely - # to this node's own Redis. When this node is the origin, this overlaps - # with the identity fields Future.__init__ already wrote; when it's a - # remote executor, these fields live only here until the completion - # callback merges them back into the origin's future:{future_id}. + # to this node's own Redis. self.redis.hset_multiple( f"future:{future_id}", { @@ -543,9 +538,8 @@ def _execute_locally( "service": service, "method": function, "args": json.dumps(args), - "created_at": wall_start, "failed": 0, - "error_message": "", + "error": "", }, ) if request_id: @@ -600,8 +594,6 @@ def _execute_locally( except Exception as e: logger.error("Failed to execute %s.%s: %s", service, function, e) - # Persist the failure locally now; the callback to origin (if any) - # is sent below, after the finally block writes final metrics. self._mark_future_failed(future_id, e) self.redis.hincrby(self._metrics_key, "full_failures", 1) finally: @@ -636,7 +628,7 @@ def _execute_locally( origin, future_id, result=serialized, failed=0, error_message="" ) else: - error_message = self.redis.hget(f"future:{future_id}", "error_message") + error_message = self.redis.hget(f"future:{future_id}", "error") self._send_result_callback( origin, future_id, failed=1, error_message=error_message or "" ) @@ -675,9 +667,9 @@ def _forward_request(self, endpoint, data): self._mark_future_failed(data.get("future_id"), e) def _send_result_callback( - self, origin, future_id, result=None, failed=0, error_message="" + self, origin, future_id, result="", failed=0, error_message="" ): - """Send the future's full execution snapshot to the originating controller.""" + """Send the future's full Redis hash to the originating controller.""" if not result: logger.warning( "Agent '%s' is sending an empty/None result for future %s to origin %s, result: %s", @@ -688,13 +680,13 @@ def _send_result_callback( ) stub = self._get_remote_stub(origin) - snapshot = snapshot_execution_fields(self.redis, future_id) + snapshot = self.redis.hgetall(f"future:{future_id}") snapshot.update( { "future_id": future_id, "result": result, "failed": int(bool(failed)), - "error_message": str(error_message or ""), + "error": str(error_message or ""), } ) payload = json.dumps(snapshot) diff --git a/ventis/controller/local_controller_frontend.py b/ventis/controller/local_controller_frontend.py index d876503..9d2f25a 100644 --- a/ventis/controller/local_controller_frontend.py +++ b/ventis/controller/local_controller_frontend.py @@ -18,11 +18,6 @@ import local_controler_pb2 import local_controler_pb2_grpc -try: - from ventis.controller.utils.future_schema import merge_execution_snapshot -except ImportError: - from future_schema import merge_execution_snapshot - logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) @@ -66,7 +61,8 @@ def WriteResult(self, request, context): ) if future_id: - merge_execution_snapshot(self.redis, future_id, data) + if data: + self.redis.hset_multiple(f"future:{future_id}", data) if failed: logger.info("WriteResult: wrote error for future %s", future_id) elif result is not None: diff --git a/ventis/controller/utils/future_schema.py b/ventis/controller/utils/future_schema.py deleted file mode 100644 index c035380..0000000 --- a/ventis/controller/utils/future_schema.py +++ /dev/null @@ -1,60 +0,0 @@ -# Shared schema for the consolidated future:{future_id} Redis hash. -# -# IDENTITY_FIELDS are written once by the future's creator (the origin) and -# must never be overwritten by a remote execution snapshot. EXECUTION_FIELDS -# are produced during/after execution and are always last-write-wins when -# merged in from a remote node's callback -- the origin has no independent -# opinion about e.g. cpu_resource or finished_at. - -IDENTITY_FIELDS = [ - "id", - "request_id", - "parent", - "service", - "method", - "args", - "created_at", -] - -EXECUTION_FIELDS = [ - "result", - "error", - "failed", - "error_message", - "finished_at", - "cpu_resource", - "gpu_resource", - "agent", - "queue_time", - "model", - "input_token_count", - "output_token_count", - "token_count", - "errors", - "input_cache_tokens", - "input_cache_write_tokens", -] - - -def snapshot_execution_fields(redis, future_id): - """Read the execution-related fields currently stored for a future.""" - data = redis.hgetall(f"future:{future_id}") - return {k: v for k, v in data.items() if k in EXECUTION_FIELDS} - - -def merge_execution_snapshot(redis, future_id, snapshot): - """Merge a remote execution snapshot into this node's future:{future_id} hash. - - Only EXECUTION_FIELDS are applied -- identity fields in the incoming - snapshot (if any) are ignored so a remote node can never clobber the - origin's own record of what the future is. None values are dropped - rather than merged, so an absent/unset field on the sender doesn't - stomp a value already present on the receiver. - """ - filtered = { - k: v - for k, v in snapshot.items() - if k in EXECUTION_FIELDS and v is not None - } - if filtered: - redis.hset_multiple(f"future:{future_id}", filtered) diff --git a/ventis/controller/utils/telemetry_logging.py b/ventis/controller/utils/telemetry_logging.py index b471d3d..7d911e2 100644 --- a/ventis/controller/utils/telemetry_logging.py +++ b/ventis/controller/utils/telemetry_logging.py @@ -99,12 +99,7 @@ def _get_engine(database_url): def pull_runtime_information(redis_client): """Scan node Redis for future execution metrics. - Each future's identity and execution metrics both live at future:{future_id} - now that the two have been consolidated into a single hash. Sibling keys - (future:{future_id}:children, future:{future_id}:consumers) share the same - prefix but are bookkeeping, not metrics -- skip them explicitly since Redis - glob patterns can't express "no suffix". """ rows = [] for key in redis_client.scan_keys("future:*"): diff --git a/ventis/future.py b/ventis/future.py index 6335e04..62bf07d 100644 --- a/ventis/future.py +++ b/ventis/future.py @@ -121,7 +121,6 @@ def _submit_request(self): self.redis.hset_multiple(f"future:{self.id}", { "error": str(e), "failed": 1, - "error_message": str(e), }) def _key(self): @@ -156,7 +155,7 @@ def value(self, timeout=None): failed = self.redis.hget(self._key(), "failed") if str(failed) == "1": raise RuntimeError( - self.redis.hget(self._key(), "error_message") + self.redis.hget(self._key(), "error") or "Unknown error" ) diff --git a/ventis/llm/bedrock.py b/ventis/llm/bedrock.py index 114a9de..f350b69 100644 --- a/ventis/llm/bedrock.py +++ b/ventis/llm/bedrock.py @@ -32,11 +32,11 @@ def call_bedrock(model_id: str, messages: list, inference_config: dict, region: metrics_key = ventis_context.get_current_metrics_key() if metrics_key: _redis.hincrby(metrics_key, "error_count", 1) - if future_id: - _redis.hset_multiple(f"future:{future_id}", { - "failed": 1, - "error_message": str(e), - }) + # Deliberately does not write "error"/"failed" onto the future here -- + # that's owned by LocalController._mark_future_failed, which only + # fires if this exception propagates all the way up uncaught. If a + # caller catches and recovers (e.g. a fallback summary), the future + # succeeds, and writing a failure here would falsely mark it failed. raise finally: if future_id: diff --git a/ventis/stub_generator.py b/ventis/stub_generator.py index bccb6b1..8c956ef 100644 --- a/ventis/stub_generator.py +++ b/ventis/stub_generator.py @@ -319,10 +319,6 @@ def generate_docker( os.path.join(script_dir, "controller", "utils", "gpu_metrics.py"), "gpu_metrics.py", ), - ( - os.path.join(script_dir, "controller", "utils", "future_schema.py"), - "future_schema.py", - ), (os.path.join(script_dir, "llm", "bedrock.py"), "bedrock.py"), ] @@ -437,7 +433,7 @@ def generate_workflow_docker( (os.path.join(script_dir, "utils", "redis_client.py"), "redis_client.py"), *[ (os.path.join(script_dir, "controller", "utils", name), name) - for name in ("gpu_metrics.py", "future_schema.py", "session_logging.py") + for name in ("gpu_metrics.py", "session_logging.py") ], ] From bdade78df634c5dc29f3c492e36ab826216f4a10 Mon Sep 17 00:00:00 2001 From: Saaketh Sodanapalli Date: Wed, 5 Aug 2026 16:28:45 -0700 Subject: [PATCH 8/8] forgot to send the full future from origin to executor --- tests/test_future.py | 3 +++ ventis/controller/local_controller.py | 32 +++++++++++++++------------ ventis/future.py | 22 +++++++++--------- 3 files changed, 31 insertions(+), 26 deletions(-) diff --git a/tests/test_future.py b/tests/test_future.py index c277358..e190426 100644 --- a/tests/test_future.py +++ b/tests/test_future.py @@ -31,6 +31,9 @@ def hset(self, name, field, value): def hget(self, name, field): return self.hashes.get(name, {}).get(field) + def hgetall(self, name): + return dict(self.hashes.get(name, {})) + def sadd(self, name, *values): self.sets.setdefault(name, set()).update(values) diff --git a/ventis/controller/local_controller.py b/ventis/controller/local_controller.py index 5e40580..d9d327b 100644 --- a/ventis/controller/local_controller.py +++ b/ventis/controller/local_controller.py @@ -335,6 +335,7 @@ def _process_request(self, data): future_id = data.get("future_id") origin = data.get("origin") # endpoint of the LC that originated this request request_id = data.get("request_id") # tracing ID from deploy module + created_at = data.get("created_at") # origin's true submission time baggage = data.get("baggage", {}) # 1. Unpack context from baggage (or fall back to local Redis) @@ -405,6 +406,7 @@ def _process_request(self, data): request_id, submitted_at, parent, + created_at, ) else: # Register the target as a consumer for any Future args @@ -521,6 +523,7 @@ def _execute_locally( request_id=None, submitted_at=None, parent=None, + created_at=None, ): """Execute a request on the local agent and write the result to Redis.""" wall_start = time.time() @@ -528,20 +531,21 @@ def _execute_locally( # Write a complete, self-contained execution record for this step entirely # to this node's own Redis. - self.redis.hset_multiple( - f"future:{future_id}", - { - "id": future_id, - "request_id": request_id or "", - "result": "", - "parent": parent or "", - "service": service, - "method": function, - "args": json.dumps(args), - "failed": 0, - "error": "", - }, - ) + initial_fields = { + "id": future_id, + "request_id": request_id or "", + "result": "", + "parent": parent or "", + "service": service, + "method": function, + "args": json.dumps(args), + "failed": 0, + "error": "", + } + + if created_at is not None: + initial_fields["created_at"] = created_at + self.redis.hset_multiple(f"future:{future_id}", initial_fields) if request_id: self.redis.sadd(f"request:{request_id}:futures", future_id) ventis_context.set_request_id(request_id) diff --git a/ventis/future.py b/ventis/future.py index 62bf07d..b071da7 100644 --- a/ventis/future.py +++ b/ventis/future.py @@ -99,18 +99,16 @@ def __init__(self, parent, service, method, args=None): def _submit_request(self): """Send the gRPC request to the local controller.""" stub = self._get_stub() - request_payload = json.dumps( - { - "service": self.service, - "function": self.method, - "args": self.args, - "future_id": self.id, - "request_id": self.request_id, - "parent": self.parent, - } - ) - request = local_controler_pb2.JsonResponse(resonse=request_payload) - self.redis.hset(f"future:{self.id}", "created_at", time.time()) + self.redis.hset(self._key(), "created_at", time.time()) + + # Send the whole future hash as the request payload -- args and the id/method fields + # are overridden below since the hash stores args are JSON-encoded (the executor needs the real dict) + request_data = dict(self.redis.hgetall(self._key())) + request_data["future_id"] = request_data.pop("id") + request_data["function"] = request_data.pop("method") + request_data["args"] = self.args + + request = local_controler_pb2.JsonResponse(resonse=json.dumps(request_data)) try: self.response = stub.Execute(request) logger.debug(