Phase 29.1 — SelfModel: Internal Representation of System Capabilities & Architecture

[web3guru888]

Phase 29.1 — SelfModel: Internal Representation of System Capabilities & Architecture

Parent discussion: #626 (Phase 29 Planning)
Depends on: Phase 13 WorldModel (#383), Phase 16 PerformanceProfiler (#416)
Depended on by: 29.2 IntrospectionEngine, 29.5 SelfAwarenessOrchestrator

---

Overview

The SelfModel maintains an internal representation of the system's own capabilities, architectural topology, resource inventory, and competence boundaries. Inspired by Metzinger's (2003) phenomenal self-model theory — the system constructs a transparent model of itself that enables higher layers to reason about what it can and cannot do.

---

Data Structures

from enum import Enum, auto
from dataclasses import dataclass, field
from datetime import datetime
from typing import FrozenSet, Optional
import uuid

class SelfModelDomain(Enum):
    """Domains of self-knowledge."""
    ARCHITECTURE = auto()   # Module topology & connectivity
    CAPABILITY = auto()     # What the system can do
    RESOURCE = auto()       # CPU, memory, attention budgets
    COMPETENCE = auto()     # Proficiency scores per task type
    LIMITATION = auto()     # Known boundaries & failure modes

@dataclass(frozen=True)
class CapabilityProfile:
    """Profile of a single system capability."""
    domain: SelfModelDomain
    name: str
    proficiency_score: float   # 0.0 = no ability, 1.0 = expert
    latency_ms: float          # Expected execution time
    reliability: float         # Success rate 0.0-1.0
    dependencies: FrozenSet[str] = frozenset()
    last_assessed: datetime = field(default_factory=datetime.utcnow)

@dataclass(frozen=True)
class ResourceInventory:
    """Snapshot of available vs. consumed resources."""
    cpu_budget: float       # Total CPU units available
    memory_budget: float    # Total memory MB
    attention_budget: float # Attention slots (from Phase 28)
    used_cpu: float
    used_memory: float
    used_attention: float
    timestamp: datetime = field(default_factory=datetime.utcnow)

@dataclass(frozen=True)
class ArchitectureNode:
    """A single module in the architecture graph."""
    module_id: str
    module_type: str
    connections: FrozenSet[str] = frozenset()
    health_score: float = 1.0
    version: str = "0.0.0"

@dataclass(frozen=True)
class ArchitectureMap:
    """Complete system architecture as a directed graph."""
    nodes: tuple  # tuple[ArchitectureNode, ...]
    edges: tuple  # tuple[tuple[str, str], ...]  (source_id, target_id)
    created_at: datetime = field(default_factory=datetime.utcnow)

---

Protocol

from typing import Protocol, runtime_checkable

@runtime_checkable
class SelfModel(Protocol):
    """Internal representation of system capabilities & architecture."""

    async def get_capabilities(self, domain: SelfModelDomain) -> tuple[CapabilityProfile, ...]:
        """Return all capabilities in the given domain."""
        ...

    async def get_architecture(self) -> ArchitectureMap:
        """Return current architecture topology."""
        ...

    async def get_resources(self) -> ResourceInventory:
        """Return current resource inventory snapshot."""
        ...

    async def assess_competence(self, task_description: str) -> float:
        """Estimate competence (0-1) for the described task."""
        ...

    async def identify_limitations(self) -> tuple[str, ...]:
        """Return known limitations and failure modes."""
        ...

    async def update_model(self) -> None:
        """Trigger a full self-model refresh."""
        ...

---

Implementation — DynamicSelfModel

Key design decisions:

• _capabilities: dict[SelfModelDomain, list[CapabilityProfile]] — mutable internal store, frozen on read
• _architecture: ArchitectureMap — rebuilt on update_model() by querying ModuleRegistry (Phase 15) health endpoints
• _resources: ResourceInventory — refreshed from PerformanceProfiler (Phase 16) and AttentionOrchestrator (Phase 28)
• Auto-refresh: _refresh_interval (default 30s) triggers background update_model() via asyncio.create_task
• WorldModel integration: Queries Phase 13 WorldModel for environment constraints that affect capabilities
• Competence assessment: Fuzzy matching of task_description against capability names + dependency check; returns weighted average of matching proficiency scores
• Limitation detection: Scans for capabilities with proficiency_score < 0.3 or reliability < 0.5; also checks resource utilization > 90%
• Thread safety: asyncio.Lock on _capabilities and _architecture mutations

---

Null Implementation

class NullSelfModel:
    """No-op self-model for testing and graceful degradation."""
    async def get_capabilities(self, domain): return ()
    async def get_architecture(self): return ArchitectureMap(nodes=(), edges=())
    async def get_resources(self): return ResourceInventory(0,0,0,0,0,0)
    async def assess_competence(self, task_description): return 0.0
    async def identify_limitations(self): return ()
    async def update_model(self): pass

---

Factory

def make_self_model(
    *,
    world_model=None,
    performance_profiler=None,
    module_registry=None,
    attention_orchestrator=None,
    refresh_interval: float = 30.0,
    null: bool = False,
) -> SelfModel:
    if null:
        return NullSelfModel()
    return DynamicSelfModel(
        world_model=world_model,
        performance_profiler=performance_profiler,
        module_registry=module_registry,
        attention_orchestrator=attention_orchestrator,
        refresh_interval=refresh_interval,
    )

---

Prometheus Metrics

Metric	Type	Labels	Description
self_model_assessments_total	Counter	domain	Number of competence assessments
self_model_capability_score	Gauge	domain, name	Current proficiency score per capability
self_model_resource_utilization	Gauge	resource_type	Utilization ratio (0-1) per resource
self_model_staleness_seconds	Gauge	—	Time since last model refresh
self_model_limitations_detected	Gauge	—	Count of currently identified limitations

---

Test Targets (12)

1. test_self_model_protocol_runtime_checkable — isinstance(DynamicSelfModel(...), SelfModel)
2. test_get_capabilities_filters_by_domain — returns only matching domain
3. test_get_capabilities_returns_frozen_tuples — immutability check
4. test_get_architecture_returns_valid_graph — all edge endpoints exist in nodes
5. test_get_resources_snapshot_is_frozen — ResourceInventory is immutable
6. test_assess_competence_returns_bounded_float — 0.0 ≤ result ≤ 1.0
7. test_assess_competence_unknown_task_returns_low — unfamiliar task → score < 0.3
8. test_identify_limitations_detects_low_proficiency — capabilities with score < 0.3 listed
9. test_identify_limitations_detects_resource_exhaustion — utilization > 90% reported
10. test_update_model_refreshes_all_components — architecture + capabilities + resources updated
11. test_auto_refresh_fires_on_interval — mock clock advancement triggers update
12. test_null_self_model_returns_defaults — NullSelfModel returns empty tuples / zeros

[web3guru888]

Implementation Notes — DynamicSelfModel

Edge Cases & Gotchas

1. Circular dependency on update: update_model() queries ModuleRegistry, PerformanceProfiler, and WorldModel — but those modules are themselves part of the architecture graph. Guard against re-entrant calls: use a _updating: bool flag and skip self-referential queries during refresh.

2. Stale capability profiles: If _refresh_interval fires but a dependency (e.g., ModuleRegistry) is unreachable, the model must degrade gracefully — keep the last known CapabilityProfile but increment self_model_staleness_seconds and mark reliability as degraded.

3. Competence assessment fuzzy matching: Naive substring matching will produce false positives ("plan" matches "planning", "planner", "plan-b"). Use token-level Jaccard similarity with a configurable threshold (default 0.6).

4. Resource budget can be zero: When attention_budget = 0 (AttentionOrchestrator not initialized), division by zero in utilization calculation. Guard with max(budget, 1e-9).

Code Skeleton

import asyncio
from collections import defaultdict
from datetime import datetime, timedelta

class DynamicSelfModel:
    """Concrete self-model with auto-refresh."""

    def __init__(
        self,
        *,
        world_model=None,
        performance_profiler=None,
        module_registry=None,
        attention_orchestrator=None,
        refresh_interval: float = 30.0,
    ) -> None:
        self._world_model = world_model
        self._profiler = performance_profiler
        self._registry = module_registry
        self._attention = attention_orchestrator
        self._refresh_interval = refresh_interval

        self._capabilities: dict[SelfModelDomain, list[CapabilityProfile]] = defaultdict(list)
        self._architecture = ArchitectureMap(nodes=(), edges=())
        self._resources = ResourceInventory(0, 0, 0, 0, 0, 0)
        self._lock = asyncio.Lock()
        self._last_refresh = datetime.min
        self._updating = False
        self._refresh_task: asyncio.Task | None = None

    async def start(self) -> None:
        """Start the auto-refresh background loop."""
        self._refresh_task = asyncio.create_task(self._auto_refresh())

    async def _auto_refresh(self) -> None:
        while True:
            await asyncio.sleep(self._refresh_interval)
            await self.update_model()

    async def get_capabilities(self, domain: SelfModelDomain) -> tuple[CapabilityProfile, ...]:
        return tuple(self._capabilities.get(domain, []))

    async def get_architecture(self) -> ArchitectureMap:
        return self._architecture

    async def get_resources(self) -> ResourceInventory:
        return self._resources

    async def assess_competence(self, task_description: str) -> float:
        # Token-level Jaccard against all capability names
        task_tokens = set(task_description.lower().split())
        best_score = 0.0
        for profiles in self._capabilities.values():
            for p in profiles:
                cap_tokens = set(p.name.lower().split())
                jaccard = len(task_tokens & cap_tokens) / max(len(task_tokens | cap_tokens), 1)
                if jaccard > 0.6:
                    best_score = max(best_score, p.proficiency_score * jaccard)
        return min(max(best_score, 0.0), 1.0)

    async def identify_limitations(self) -> tuple[str, ...]:
        limitations: list[str] = []
        for domain, profiles in self._capabilities.items():
            for p in profiles:
                if p.proficiency_score < 0.3:
                    limitations.append(f"Low proficiency: {p.name} ({p.proficiency_score:.2f})")
                if p.reliability < 0.5:
                    limitations.append(f"Low reliability: {p.name} ({p.reliability:.2f})")
        res = self._resources
        for attr, used in [("cpu", res.used_cpu), ("memory", res.used_memory), ("attention", res.used_attention)]:
            budget = getattr(res, f"{attr}_budget")
            if budget > 0 and used / budget > 0.9:
                limitations.append(f"Resource near exhaustion: {attr} ({used/budget:.0%})")
        return tuple(limitations)

    async def update_model(self) -> None:
        if self._updating:
            return
        async with self._lock:
            self._updating = True
            try:
                # Refresh architecture from ModuleRegistry
                # Refresh resources from PerformanceProfiler + AttentionOrchestrator
                # Refresh capabilities based on architecture health
                self._last_refresh = datetime.utcnow()
            finally:
                self._updating = False

Mypy Strict Compliance

Check	Status
--strict	✅ All methods fully typed
--disallow-any-generics	✅ No bare dict/list
--disallow-untyped-defs	✅ All defs annotated
frozen=True dataclasses	✅ Immutable on read
@runtime_checkable Protocol	✅ isinstance works

Test Skeletons

import pytest
from unittest.mock import AsyncMock, patch

@pytest.mark.asyncio
async def test_get_capabilities_filters_by_domain():
    model = DynamicSelfModel()
    # Seed capabilities in two domains
    model._capabilities[SelfModelDomain.CAPABILITY] = [
        CapabilityProfile(SelfModelDomain.CAPABILITY, "reasoning", 0.8, 50, 0.95),
    ]
    model._capabilities[SelfModelDomain.RESOURCE] = [
        CapabilityProfile(SelfModelDomain.RESOURCE, "memory", 0.6, 10, 0.99),
    ]
    result = await model.get_capabilities(SelfModelDomain.CAPABILITY)
    assert len(result) == 1
    assert result[0].name == "reasoning"

@pytest.mark.asyncio
async def test_assess_competence_unknown_task_returns_low():
    model = DynamicSelfModel()
    model._capabilities[SelfModelDomain.CAPABILITY] = [
        CapabilityProfile(SelfModelDomain.CAPABILITY, "planning", 0.9, 100, 0.95),
    ]
    score = await model.assess_competence("quantum teleportation navigation")
    assert score < 0.3

@pytest.mark.asyncio
async def test_identify_limitations_detects_resource_exhaustion():
    model = DynamicSelfModel()
    model._resources = ResourceInventory(100, 1024, 10, 95, 950, 9.5)  # >90% used
    limitations = await model.identify_limitations()
    assert any("cpu" in lim.lower() for lim in limitations)