fix(ci): memoize Mission Control build cascade (stop 55-min timeout)

.github/workflows/ci.yml

@@ -34,7 +34,9 @@ jobs:
           python -m pip install --upgrade pip
           pip install -c requirements.lock -e ".[dev,secrets,cloud]"
           # Headless Linux runners have no OS keyring; vault tests need a backend.
-          pip install ruff keyrings.alt
+          # pytest-timeout turns any pathological hang into a fast, named failure
+          # instead of silently consuming the whole job budget.
+          pip install ruff keyrings.alt pytest-timeout
       - name: Syntax check (compileall)
         run: python -m compileall -q aethos_core aethos_sdk
       - name: Lint (real errors only)
@@ -47,7 +49,7 @@ jobs:
         env:
           USE_REAL_LLM: "false"
           ACTIVE_PROVIDER: "none"
-        run: python -m pytest tests/test_beta_smoke_harness.py -q -p no:cacheprovider
+        run: python -m pytest tests/test_beta_smoke_harness.py -q -p no:cacheprovider --timeout=120 --timeout-method=signal
       - name: Behavioral corpus (masterpiece gates)
         env:
           USE_REAL_LLM: "false"
@@ -57,12 +59,15 @@ jobs:
         env:
           USE_REAL_LLM: "false"
           ACTIVE_PROVIDER: "none"
-        run: python -m pytest tests/test_chat_20_turn.py -q -p no:cacheprovider
+        run: python -m pytest tests/test_chat_20_turn.py -q -p no:cacheprovider --timeout=120 --timeout-method=signal
       - name: Pytest (full suite)
         env:
           USE_REAL_LLM: "false"
           ACTIVE_PROVIDER: "none"
-        run: python -m pytest tests -q -p no:cacheprovider
+        # Per-test timeout is a safety net: no single test may hang the job. A test
+        # exceeding the limit fails with a stack trace naming it, rather than the
+        # whole suite silently timing out at the job level.
+        run: python -m pytest tests -q -p no:cacheprovider --timeout=120 --timeout-method=signal
 
   web:
     name: Web typecheck + tests

aethos_core/mission_control/agent_execution_quality_throughput_metrics/agent_execution_quality_throughput_metrics_service.py

@@ -3,6 +3,8 @@
 
 from __future__ import annotations
 
+from aethos_core.mission_control.build_memoization import scoped_build
+
 from dataclasses import dataclass, field
 from datetime import UTC, datetime
 from typing import Any
@@ -289,6 +291,7 @@ def _end_to_end_throughput_score(
     }
 
 
+@scoped_build
 def build_agent_execution_quality_throughput_metrics(
     *, session_id: str
 ) -> AgentExecutionQualityThroughputMetricsResult:

aethos_core/mission_control/approval_inbox/approval_inbox_service.py

@@ -3,6 +3,8 @@
 
 from __future__ import annotations
 
+from aethos_core.mission_control.build_memoization import scoped_build
+
 from dataclasses import dataclass, field
 from typing import Any
 
@@ -616,6 +618,7 @@ def _collect_operational_deployment_approvals() -> list[dict[str, Any]]:
     return items
 
 
+@scoped_build
 def build_approval_inbox(*, session_id: str) -> ApprovalInboxResult:
     from aethos_core.mission_control.cross_lane.snapshot_service import load_mission_control_config
 
@@ -657,6 +660,7 @@ def build_approval_inbox(*, session_id: str) -> ApprovalInboxResult:
     )
 
 
+@scoped_build
 def approval_inbox_payload(*, session_id: str) -> dict[str, Any]:
     result = build_approval_inbox(session_id=session_id)
     if not result.ok:

aethos_core/mission_control/atlas_trader_pilot_arc_orchestrator/atlas_trader_pilot_arc_orchestrator_service.py

@@ -3,6 +3,8 @@
 
 from __future__ import annotations
 
+from aethos_core.mission_control.build_memoization import scoped_build
+
 from dataclasses import dataclass, field
 from datetime import UTC, datetime
 from pathlib import Path
@@ -269,6 +271,7 @@ def _pilot_readiness_summary(*, gates: dict[str, Any], arc_state: str) -> dict[s
     }
 
 
+@scoped_build
 def build_atlas_trader_pilot_arc_orchestrator(*, session_id: str) -> AtlasTraderPilotArcOrchestratorResult:
     sid = (session_id or "default").strip()[:64] or "default"
     exported_at = _exported_at()

aethos_core/mission_control/atlas_trader_trust_report_freeze/atlas_trader_trust_report_freeze_service.py

@@ -3,6 +3,8 @@
 
 from __future__ import annotations
 
+from aethos_core.mission_control.build_memoization import scoped_build
+
 from dataclasses import dataclass, field
 from datetime import UTC, datetime
 from typing import Any
@@ -402,6 +404,7 @@ def _evidence_index(
     return entries
 
 
+@scoped_build
 def build_atlas_trader_trust_report_freeze(*, session_id: str) -> AtlasTraderTrustReportFreezeResult:
     sid = (session_id or "default").strip()[:64] or "default"
     exported_at = _exported_at()

aethos_core/mission_control/autonomous_application_lifecycle_management/autonomous_application_lifecycle_management_service.py

@@ -3,6 +3,8 @@
 
 from __future__ import annotations
 
+from aethos_core.mission_control.build_memoization import scoped_build
+
 from dataclasses import dataclass, field
 from datetime import UTC, datetime
 from typing import Any
@@ -414,6 +416,7 @@ def _lifecycle_memory(
     ]
 
 
+@scoped_build
 def build_autonomous_application_lifecycle_management(
     *, session_id: str
 ) -> AutonomousApplicationLifecycleManagementResult:

aethos_core/mission_control/autonomous_business_operating_system/autonomous_business_operating_system_service.py

@@ -3,6 +3,8 @@
 
 from __future__ import annotations
 
+from aethos_core.mission_control.build_memoization import scoped_build
+
 from dataclasses import dataclass, field
 from datetime import UTC, datetime
 from typing import Any
@@ -512,6 +514,7 @@ def _business_operating_memory(
     ]
 
 
+@scoped_build
 def build_autonomous_business_operating_system(*, session_id: str) -> AutonomousBusinessOperatingSystemResult:
     sid = (session_id or "default").strip()[:64] or "default"
     business_records = list_autonomous_business_operating_system_records()

aethos_core/mission_control/autonomous_capability_registry/autonomous_capability_registry_service.py

@@ -3,6 +3,8 @@
 
 from __future__ import annotations
 
+from aethos_core.mission_control.build_memoization import scoped_build
+
 import inspect
 import re
 import shutil
@@ -495,6 +497,7 @@ def _self_awareness_report(
     ]
 
 
+@scoped_build
 def build_autonomous_capability_registry(*, session_id: str) -> AutonomousCapabilityRegistryResult:
     sid = (session_id or "default").strip()[:64] or "default"
     capability_records = list_autonomous_capability_registry_records()

aethos_core/mission_control/autonomous_product_stewardship/autonomous_product_stewardship_service.py

@@ -3,6 +3,8 @@
 
 from __future__ import annotations
 
+from aethos_core.mission_control.build_memoization import scoped_build
+
 from dataclasses import dataclass, field
 from datetime import UTC, datetime
 from typing import Any
@@ -305,6 +307,7 @@ def _stewardship_memory(
     ]
 
 
+@scoped_build
 def build_autonomous_product_stewardship(*, session_id: str) -> AutonomousProductStewardshipResult:
     sid = (session_id or "default").strip()[:64] or "default"
     stewardship_records = list_autonomous_product_stewardship_records()

aethos_core/mission_control/billing_entitlements_foundation/billing_entitlements_foundation_service.py

@@ -3,6 +3,8 @@
 
 from __future__ import annotations
 
+from aethos_core.mission_control.build_memoization import scoped_build
+
 from dataclasses import dataclass, field
 from datetime import UTC, datetime
 from typing import Any
@@ -81,6 +83,7 @@ def _usage_snapshot(*, org_id: str) -> dict[str, int]:
     }
 
 
+@scoped_build
 def build_billing_entitlements_foundation(*, session_id: str) -> BillingEntitlementsFoundationResult:
     sid = (session_id or "default").strip()[:64] or "default"
     records = list_billing_entitlements_foundation_records()

aethos_core/mission_control/bounded_execution_participation/bounded_execution_participation_service.py

@@ -3,6 +3,8 @@
 
 from __future__ import annotations
 
+from aethos_core.mission_control.build_memoization import scoped_build
+
 from dataclasses import dataclass, field
 from datetime import UTC, datetime
 from typing import Any
@@ -304,6 +306,7 @@ def _participation_integrity_scoring(
     ]
 
 
+@scoped_build
 def build_bounded_execution_participation(*, session_id: str) -> BoundedExecutionParticipationResult:
     sid = (session_id or "default").strip()[:64] or "default"
 

aethos_core/mission_control/bounded_multi_agent_delivery_execution/bounded_multi_agent_delivery_execution_service.py

@@ -3,6 +3,8 @@
 
 from __future__ import annotations
 
+from aethos_core.mission_control.build_memoization import scoped_build
+
 from dataclasses import dataclass, field
 from datetime import UTC, datetime
 from typing import Any
@@ -195,6 +197,7 @@ def _execution_readiness(*, pipeline_state: str, gates: dict[str, Any]) -> dict[
     }
 
 
+@scoped_build
 def build_bounded_multi_agent_delivery_execution(
     *, session_id: str
 ) -> BoundedMultiAgentDeliveryExecutionResult:

aethos_core/mission_control/bounded_multi_agent_delivery_work_packages/bounded_multi_agent_delivery_work_packages_service.py

@@ -3,6 +3,8 @@
 
 from __future__ import annotations
 
+from aethos_core.mission_control.build_memoization import scoped_build
+
 from dataclasses import dataclass, field
 from datetime import UTC, datetime
 from typing import Any
@@ -281,6 +283,7 @@ def _delivery_integrity_scoring(
     ]
 
 
+@scoped_build
 def build_bounded_delivery_work_packages(*, session_id: str) -> BoundedDeliveryWorkPackagesResult:
     sid = (session_id or "default").strip()[:64] or "default"
 

aethos_core/mission_control/build_memoization.py

@@ -0,0 +1,99 @@
+# SPDX-License-Identifier: Apache-2.0
+"""Request-scoped memoization for the Mission Control build_* cascade.
+
+Many Mission Control ``build_*`` services form a diamond dependency: a top-level
+builder transitively invokes shared lower builders many times, each of which
+re-runs the entire stack beneath it. With no caching this is exponential — a
+single top-level build (e.g. the pilot-arc orchestrator) can take minutes, which
+in turn pushes the full test suite past the CI time budget.
+
+These builders are pure, read-only compositions: for a given ``session_id`` (and
+the persisted state on disk, which does not change during a single logical build)
+they return an equivalent value every time. This module provides a
+context-scoped cache so that within one logical build each ``(builder, args)`` is
+computed once. Nested re-requests return a ``deepcopy`` of the cached value, so
+every caller still receives an independent object — behavior is identical to the
+un-memoized version, only without the exponential recomputation. The cache lives
+only for the duration of the outermost decorated call and is discarded on return,
+so no state leaks across calls, requests, or tests.
+"""
+
+from __future__ import annotations
+
+import contextvars
+import copy
+import functools
+from typing import Any, Callable, TypeVar
+
+_CACHE: contextvars.ContextVar[dict[Any, Any] | None] = contextvars.ContextVar(
+    "mission_control_build_cache", default=None
+)
+
+# Sentinel: key is currently being computed. Prevents infinite recursion if a
+# builder re-enters itself (directly or via a cycle) before the first result is stored.
+_IN_PROGRESS: object = object()
+
+F = TypeVar("F", bound=Callable[..., Any])
+
+_SCALAR = (str, int, float, bool)
+
+
+def _cache_key(fn: Callable[..., Any], kwargs: dict[str, Any]) -> tuple[Any, ...]:
+    scalar_kwargs = tuple(
+        sorted((k, v) for k, v in kwargs.items() if v is None or isinstance(v, _SCALAR))
+    )
+    return (fn.__module__, fn.__qualname__, scalar_kwargs)
+
+
+def _clone(value: Any) -> Any:
+    try:
+        return copy.deepcopy(value)
+    except (RecursionError, TypeError):
+        # Circular / un-copyable graphs: prefer sharing the cached object over failing.
+        return value
+
+
+def scoped_build(fn: F) -> F:
+    """Memoize a pure ``build_*`` service within one logical build tree.
+
+    Only the canonical keyword-only call shape (no positional args) is cached; any
+    other call shape falls through to the wrapped function unchanged, so this can
+    never silently mis-key a call.
+    """
+
+    @functools.wraps(fn)
+    def wrapper(*args: Any, **kwargs: Any) -> Any:
+        if args:
+            return fn(*args, **kwargs)
+        key = _cache_key(fn, kwargs)
+        store = _CACHE.get()
+        if store is not None:
+            if key in store:
+                cached = store[key]
+                if cached is _IN_PROGRESS:
+                    # Re-entrant / cyclic call while computing — run uncached.
+                    return fn(**kwargs)
+                return _clone(cached)
+            store[key] = _IN_PROGRESS
+            try:
+                result = fn(**kwargs)
+            except BaseException:
+                store.pop(key, None)
+                raise
+            store[key] = result
+            return _clone(result)
+        store = {}
+        token = _CACHE.set(store)
+        try:
+            store[key] = _IN_PROGRESS
+            try:
+                result = fn(**kwargs)
+            except BaseException:
+                store.pop(key, None)
+                raise
+            store[key] = result
+            return result
+        finally:
+            _CACHE.reset(token)
+
+    return wrapper  # type: ignore[return-value]

aethos_core/mission_control/capability_registry_runtime_integration/capability_registry_runtime_integration_service.py

@@ -3,6 +3,8 @@
 
 from __future__ import annotations
 
+from aethos_core.mission_control.build_memoization import scoped_build
+
 from dataclasses import dataclass, field
 from datetime import UTC, datetime
 from typing import Any
@@ -112,6 +114,7 @@ def _names(*cap_ids: str) -> list[str]:
     }
 
 
+@scoped_build
 def build_capability_registry_runtime_integration(
     *, session_id: str
 ) -> CapabilityRegistryRuntimeIntegrationResult:

aethos_core/mission_control/channel_integration_foundation/channel_integration_foundation_service.py

@@ -3,6 +3,8 @@
 
 from __future__ import annotations
 
+from aethos_core.mission_control.build_memoization import scoped_build
+
 from dataclasses import dataclass, field
 from datetime import UTC, datetime
 from typing import Any
@@ -110,6 +112,7 @@ def _supported_actions(channel: str) -> list[str]:
     return []
 
 
+@scoped_build
 def build_channel_integration_foundation(*, session_id: str) -> ChannelIntegrationFoundationResult:
     sid = (session_id or "default").strip()[:64] or "default"
     user_id = _resolve_user(session_id=sid)