diff --git a/examples/portfolio/agents/advisor_agent.py b/examples/portfolio/agents/advisor_agent.py
index 2db763f..0915eea 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:<future_id>:metrics hash. Configure
+# recorded onto this execution's future:<future_id> hash. Configure
 # with env vars:
 #   BEDROCK_MODEL_ID  (default: meta.llama3-8b-instruct-v1:0)
 #   AWS_REGION        (default: us-east-1)
@@ -11,13 +11,10 @@
 # If the LLM is unavailable (returns an empty string), it falls back to a
 # deterministic templated summary so the pipeline still returns.
 #
-# Resource profile: cheap CPU; the LLM cost sits in LLMAgent, not here.
+# Resource profile: cheap CPU; the LLM cost sits in the Bedrock call, not here.
 
-import sys
 import os
 
-sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "stubs"))
-from llm_agent import LLMAgent
 try:
     from ventis.llm.bedrock import call_bedrock
 except ImportError:
@@ -27,16 +24,14 @@
 class AdvisorAgent(object):
     def __init__(self):
         self.tools = [self.summarize]
-        self.llm = LLMAgent()
+        self.model_id = os.environ.get(
+            "BEDROCK_MODEL_ID", "meta.llama3-8b-instruct-v1:0"
+        )
+        self.region = os.environ.get("AWS_REGION", "us-east-1")
 
     def summarize(self, holdings: dict, metrics: dict, risk: dict) -> str:
         """Write a short plain-English briefing on the portfolio."""
         prompt = self._build_prompt(holdings, metrics, risk)
-        text = self.llm.complete(
-            prompt=prompt, max_tokens=400, temperature=0.2
-        ).value()
-        if not text:
-            print("AdvisorAgent: LLM returned no output; using templated summary.")
         try:
             response = call_bedrock(
                 model_id=self.model_id,
@@ -48,7 +43,6 @@ def summarize(self, holdings: dict, metrics: dict, risk: dict) -> str:
         except Exception as e:
             print(f"AdvisorAgent: Bedrock call failed ({e}); using templated summary.")
             return self._fallback_summary(metrics, risk)
-        return text
 
     def _build_prompt(self, holdings: dict, metrics: dict, risk: dict) -> str:
         lines = ["You are a portfolio analyst. Given the figures below, write a "
diff --git a/examples/portfolio/agents/intent_agent.py b/examples/portfolio/agents/intent_agent.py
index 972c8af..d74b27b 100644
--- a/examples/portfolio/agents/intent_agent.py
+++ b/examples/portfolio/agents/intent_agent.py
@@ -7,20 +7,9 @@
 #      ->  {"holdings": {"AAPL": 0.4, "MSFT": 0.35, "NVDA": 0.25},
 #           "lookback_days": 180}
 #
-<<<<<<< HEAD
-# The actual model call lives in the shared LLMAgent (remote, resolved via
-# .value()) — this agent only builds the prompt and parses the result, so no
-# Bedrock boilerplate lives here. If the LLM is unavailable or returns
-# unparseable output, parse() raises: there is no fallback, the request fails
-# loudly rather than guessing at the holdings. Weights are renormalized to 1.0.
-#
-# Resource profile: cheap CPU; the LLM cost sits in LLMAgent, not here.
-
-import sys
-=======
 # Calls AWS Bedrock (Converse API) via ventis.llm.bedrock -- same pattern as
 # AdvisorAgent -- so token/cost telemetry gets recorded onto this execution's
-# future:<future_id>:metrics hash. Configure with env vars:
+# future:<future_id> hash. Configure with env vars:
 #   BEDROCK_MODEL_ID  (default: meta.llama3-8b-instruct-v1:0)
 #   AWS_REGION        (default: us-east-1)
 #
@@ -31,20 +20,14 @@
 # Resource profile: cheap CPU, single call per request, on the critical path
 # before the fan-out.
 
->>>>>>> remotes/origin/telemetry-signals
 import os
 import re
 import json
 
-<<<<<<< HEAD
-sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "stubs"))
-from llm_agent import LLMAgent
-=======
 try:
     from ventis.llm.bedrock import call_bedrock
 except ImportError:
     from bedrock import call_bedrock
->>>>>>> remotes/origin/telemetry-signals
 
 DEFAULT_LOOKBACK_DAYS = 365
 
@@ -52,15 +35,6 @@
 class IntentAgent(object):
     def __init__(self):
         self.tools = [self.parse]
-<<<<<<< HEAD
-        self.llm = LLMAgent()
-
-    def parse(self, query: str) -> dict:
-        """Parse a natural-language portfolio request into holdings + lookback."""
-        text = self.llm.complete(
-            prompt=self._build_prompt(query), max_tokens=300, temperature=0.0
-        ).value()
-=======
         self.model_id = os.environ.get(
             "BEDROCK_MODEL_ID", "meta.llama3-8b-instruct-v1:0"
         )
@@ -75,7 +49,6 @@ def parse(self, query: str) -> dict:
             region=self.region,
         )
         text = response["output"]["message"]["content"][0]["text"]
->>>>>>> remotes/origin/telemetry-signals
         if not text:
             raise ValueError("IntentAgent: LLM returned no output for the request.")
 
@@ -148,15 +121,7 @@ def _sanitize(self, parsed: dict) -> dict:
 
 
 if __name__ == "__main__":
-<<<<<<< HEAD
-    # Assumes the LLMAgent stub (Future-returning) is on the path, as it is
-    # inside the deployed pipeline.
-    agent = IntentAgent()
-    print(agent.parse(
-        "Analyze 40% Apple, 35% Microsoft and 25% Nvidia over the last 6 months"
-=======
     agent = IntentAgent()
     print(agent.parse(
         query="Analyze 40% Apple, 35% Microsoft and 25% Nvidia over the last 6 months"
->>>>>>> remotes/origin/telemetry-signals
     ))
diff --git a/examples/portfolio/agents/llm_agent.py b/examples/portfolio/agents/llm_agent.py
deleted file mode 100644
index d42e4fb..0000000
--- a/examples/portfolio/agents/llm_agent.py
+++ /dev/null
@@ -1,50 +0,0 @@
-# LLM Agent
-#
-# Shared inference node. Owns all the AWS Bedrock (Converse API) plumbing so no
-# other agent has to carry boto3 boilerplate — they just call complete(prompt)
-# and get text back. Configure with env vars:
-#   BEDROCK_MODEL_ID  (default: meta.llama3-8b-instruct-v1:0)
-#   AWS_REGION        (default: us-east-1)
-#
-# On any failure (no boto3, no creds, model not enabled) it returns an empty
-# string; callers decide how to degrade (templated summary, regex parse, etc.).
-#
-# Resource profile: LLM-bound. This is the only node that talks to Bedrock, so
-# it's the natural place to scale inference capacity independently.
-
-import os
-
-
-class LLMAgent(object):
-    def __init__(self):
-        self.tools = [self.complete]
-        self.model_id = os.environ.get(
-            "BEDROCK_MODEL_ID", "meta.llama3-8b-instruct-v1:0"
-        )
-        self.region = os.environ.get("AWS_REGION", "us-east-1")
-
-    def complete(
-        self, prompt: str, max_tokens: int = 400, temperature: float = 0.2
-    ) -> str:
-        """Run a single-turn completion on Bedrock; '' on any failure."""
-        try:
-            import boto3
-
-            client = boto3.client("bedrock-runtime", region_name=self.region)
-            response = client.converse(
-                modelId=self.model_id,
-                messages=[{"role": "user", "content": [{"text": prompt}]}],
-                inferenceConfig={
-                    "maxTokens": max_tokens,
-                    "temperature": temperature,
-                },
-            )
-            return response["output"]["message"]["content"][0]["text"]
-        except Exception as e:
-            print(f"LLMAgent: Bedrock call failed ({e}).")
-            return ""
-
-
-if __name__ == "__main__":
-    agent = LLMAgent()
-    print(agent.complete("Say hello in one short sentence.", max_tokens=50))
\ No newline at end of file
diff --git a/examples/portfolio/agents/llm_agent.yaml b/examples/portfolio/agents/llm_agent.yaml
deleted file mode 100644
index ccc0095..0000000
--- a/examples/portfolio/agents/llm_agent.yaml
+++ /dev/null
@@ -1,14 +0,0 @@
-agent:
-  name: LLMAgent
-  functions:
-    - name: complete
-      description: Run a single-turn completion on Bedrock; '' on any failure.
-      arguments:
-        - name: prompt
-          type: str
-        - name: max_tokens
-          type: int
-        - name: temperature
-          type: float
-      returns:
-        type: str
diff --git a/examples/portfolio/config/global_controller.yaml b/examples/portfolio/config/global_controller.yaml
index 29e8de9..bbc160b 100644
--- a/examples/portfolio/config/global_controller.yaml
+++ b/examples/portfolio/config/global_controller.yaml
@@ -6,26 +6,6 @@
 # reflect each stage's real cost so the scheduler has placement decisions to make.
 
 agents:
-<<<<<<< HEAD
-  # Shared inference node. Owns all Bedrock plumbing; IntentAgent and
-  # AdvisorAgent delegate their model calls here. LLM-bound — scale replicas
-  # to match inference demand.
-  - name: LLMAgent
-    host: localhost
-    port: 8075
-    redis_port: 6379
-    replicas: 1
-    resources:
-      cpu: 1
-      memory: 512
-    entrypoint: agents/llm_agent.py
-
-  # Stage 0: parse the free-text request into structured holdings + lookback
-  # window (calls LLMAgent). Cheap CPU, one call per request, on the critical
-  # path before the fan-out.
-  - name: IntentAgent
-    host: localhost
-    port: 8076
   # Stage 0: parse the free-text request into structured holdings + lookback
   # window (calls Bedrock directly via ventis.llm.bedrock). Cheap CPU, one
   # call per request, on the critical path before the fan-out.
@@ -36,8 +16,6 @@ agents:
       cpu: 1
       memory: 256
     entrypoint: agents/intent_agent.py
-
-  # Stage 0: price history fetch. Network/IO-bound, cheap CPU. Called by
     provider: EC2
     instance_type: t3.micro
 
diff --git a/examples/portfolio/config/policy.yaml b/examples/portfolio/config/policy.yaml
index 573c91b..834c88e 100644
--- a/examples/portfolio/config/policy.yaml
+++ b/examples/portfolio/config/policy.yaml
@@ -14,7 +14,6 @@ rules:
   - match: {}
     access:
       - Workflow
-      - LLMAgent
       - IntentAgent
       - PriceAgent
       - MetricsAgent
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:<future_id>:metrics hash — same pattern as
+# future:<future_id> 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..82da262 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,8 +103,8 @@ 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",
-            {"failed": 1, "error_message": "agent exploded"},
+            "future:future-1",
+            {"failed": 1, "error": "agent exploded"},
         )
         future = SimpleNamespace(
             redis=redis,
@@ -131,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",
             },
         )
 
@@ -145,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",
                 }
             )
         )
@@ -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"),
             "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"),
+            "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"),
+            "executor exploded",
+        )
 
 
 if __name__ == "__main__":
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/tests/test_local_controller_metrics.py b/tests/test_local_controller_metrics.py
index 90491a7..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
         )
     )
@@ -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,11 @@ 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
-        )
-        self.assertEqual(
-            redis.hget("future:future-2:metrics", "error_message"), "nope"
+            redis.hget("future:future-2", "failed"), 1
         )
+        self.assertNotIn("future:future-2:metrics", redis.hashes)
         self.assertEqual(
             redis.hget("controller:localhost:50051:metrics", "requests_served"), 1
         )
@@ -224,11 +223,8 @@ 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:metrics", "error_message"),
-            "No agent loaded",
-        )
+        self.assertEqual(redis.hget("future:future-3", "failed"), 1)
+        self.assertNotIn("future:future-3:metrics", redis.hashes)
 
     def test_remote_execution_failure_sends_error_callback(self):
         redis = _FakeRedis()
@@ -261,17 +257,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.assertEqual(payload["result"], "")
+        self.assertEqual(payload["failed"], 1)
+        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")
+        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_telemetry_logging.py b/tests/test_telemetry_logging.py
index 37ec8c0..5bf78d7 100644
--- a/tests/test_telemetry_logging.py
+++ b/tests/test_telemetry_logging.py
@@ -116,7 +116,7 @@ def tearDown(self):
     def test_pull_and_upsert(self):
         redis = _FakeRedis(
             {
-                "future:abc:metrics": {
+                "future:abc": {
                     "id": "abc",
                     "request_id": "req1",
                     "agent": "1f2e3d4c5b6a7988fedcba9876543210",
@@ -163,7 +163,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",
@@ -187,7 +187,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",
@@ -217,7 +217,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",
@@ -253,7 +253,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",
@@ -284,7 +284,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",
@@ -310,7 +310,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",
@@ -341,7 +341,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",
@@ -365,7 +365,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",
@@ -390,7 +390,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/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 9f53a3e..d9d327b 100644
--- a/ventis/controller/local_controller.py
+++ b/ventis/controller/local_controller.py
@@ -308,10 +308,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},
         )
 
         if origin and origin != self._my_endpoint:
@@ -336,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)
@@ -406,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
@@ -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")
                             or "Unknown error"
                         )
                     # print("Waiting for result for future next iteration %s", value)
@@ -524,30 +523,29 @@ 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()
         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.
-        self.redis.hset_multiple(
-            f"future:{future_id}:metrics",
-            {
-                "id": future_id,
-                "request_id": request_id or "",
-                "result": "",
-                "parent": parent or "",
-                "service": service,
-                "method": function,
-                "args": json.dumps(args),
-                "created_at": wall_start,
-                "failed": 0,
-                "error_message": "",
-            },
-        )
+        # Write a complete, self-contained execution record for this step entirely
+        # to this node's own Redis.
+        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)
@@ -568,6 +566,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 +585,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 +595,10 @@ 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)
+            self._mark_future_failed(future_id, e)
             self.redis.hincrby(self._metrics_key, "full_failures", 1)
         finally:
             wall_end = time.time()
@@ -619,7 +610,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 +623,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")
+                    self._send_result_callback(
+                        origin, future_id, failed=1, error_message=error_message or ""
+                    )
+
             ventis_context.set_current_future_id(parent or "")
 
     # ------------------------------------------------------------------ #
@@ -666,9 +671,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 a result and its failure metadata 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",
@@ -679,12 +684,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 = self.redis.hgetall(f"future:{future_id}")
+        snapshot.update(
+            {
+                "future_id": future_id,
+                "result": result,
+                "failed": int(bool(failed)),
+                "error": 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 0117c4a..9d2f25a 100644
--- a/ventis/controller/local_controller_frontend.py
+++ b/ventis/controller/local_controller_frontend.py
@@ -51,7 +51,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,17 +61,11 @@ def WriteResult(self, request, context):
                 )
 
             if future_id:
-                self.redis.hset_multiple(
-                    f"future:{future_id}:metrics",
-                    {"failed": failed, "error_message": error_message},
-                )
+                if data:
+                    self.redis.hset_multiple(f"future:{future_id}", data)
                 if failed:
-                    self.redis.hset(
-                        f"future:{future_id}", "error", error_message or "Unknown error"
-                    )
                     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,
@@ -124,7 +117,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/telemetry_logging.py b/ventis/controller/utils/telemetry_logging.py
index 3c168c7..7d911e2 100644
--- a/ventis/controller/utils/telemetry_logging.py
+++ b/ventis/controller/utils/telemetry_logging.py
@@ -99,13 +99,12 @@ 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}
     """
     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]
@@ -135,11 +134,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..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(
@@ -118,10 +116,9 @@ 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),
             })
 
     def _key(self):
@@ -153,10 +150,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")
                 or "Unknown error"
             )
 
diff --git a/ventis/llm/bedrock.py b/ventis/llm/bedrock.py
index 4135352..f350b69 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:<future_id>:metrics)."""
+    currently executing future's hash (future:<future_id>)."""
     import boto3
 
     client = boto3.client("bedrock-runtime", region_name=region)
@@ -32,16 +32,16 @@ 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),
-            })
+        # 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:
             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", "")),