File size: 7,648 Bytes
63dd1f4
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
"""
SelfModel — Vitalis FSI

Vitalis maintains a coherent understanding of its own architecture,
capabilities, limitations, and growth trajectory.

This is not a static config file.
This is a living self-representation that updates as the system evolves.

The SelfModel answers:
  - What am I capable of right now?
  - What are my current limitations?
  - How have I grown since initialization?
  - What is the next capability boundary I should push?
  - Am I operating within my identity alignment?
"""
import numpy as np
import os
import json
import time
from vitalis_ide.math_core.kernel import VitalisKernel
from src.brain.resonance import ResonanceEngine
from src.hippocampus import Hippocampus


class SelfModel:
    # Capability thresholds — evolve as resonance weights grow
    CAPABILITY_THRESHOLD  = 1.2   # resonance weight above this = capable
    LIMITATION_THRESHOLD  = 0.5   # resonance weight below this = limited
    FRONTIER_THRESHOLD    = 0.8   # between = frontier (learning zone)

    def __init__(self):
        self.kernel    = VitalisKernel()
        self.resonance = ResonanceEngine()
        self.hippocampus = Hippocampus()
        self.path      = os.path.expanduser(
            "~/.vitalis_workspace/self_model.json"
        )
        self._birth_time = self._load_birth_time()
        self._snapshots  = self._load_snapshots()

    def _load_birth_time(self) -> float:
        if os.path.exists(self.path):
            with open(self.path) as f:
                data = json.load(f)
                return data.get("birth_time", time.time())
        birth = time.time()
        self._save({"birth_time": birth, "snapshots": []})
        return birth

    def _load_snapshots(self) -> list:
        if os.path.exists(self.path):
            with open(self.path) as f:
                return json.load(f).get("snapshots", [])
        return []

    def _save(self, data: dict = None):
        os.makedirs(os.path.dirname(self.path), exist_ok=True)
        if data is None:
            data = {
                "birth_time": self._birth_time,
                "snapshots":  self._snapshots[-100:],
            }
        with open(self.path, "w") as f:
            json.dump(data, f, indent=2)

    # ------------------------------------------------------------------
    # Core self-assessment
    # ------------------------------------------------------------------
    def assess(self) -> dict:
        """
        Full self-assessment. Returns current capability map,
        limitations, frontier zones, and growth trajectory.
        """
        resonance_report = self.resonance.report()
        weights          = self.resonance.weights
        memory_report    = self.hippocampus.memory_report()

        # Capability map
        capabilities = []
        limitations  = []
        frontiers    = []

        for pattern, weight in weights.items():
            if weight >= self.CAPABILITY_THRESHOLD:
                capabilities.append((pattern, round(weight, 3)))
            elif weight <= self.LIMITATION_THRESHOLD:
                limitations.append((pattern, round(weight, 3)))
            else:
                frontiers.append((pattern, round(weight, 3)))

        capabilities.sort(key=lambda x: x[1], reverse=True)
        limitations.sort(key=lambda x: x[1])
        frontiers.sort(key=lambda x: x[1], reverse=True)

        # Growth metrics
        age_hours   = (time.time() - self._birth_time) / 3600
        mem_count   = len(memory_report)
        avg_strength = float(np.mean([
            v["strength"] for v in memory_report.values()
        ])) if memory_report else 0.0

        # Next capability boundary — highest frontier item
        next_boundary = frontiers[0][0] if frontiers else "unexplored"

        # Identity coherence — how aligned are capabilities with identity?
        identity_vec = self._load_identity_vec()
        coherence    = self._identity_coherence(identity_vec, capabilities)

        assessment = {
            "timestamp":       time.time(),
            "age_hours":       round(age_hours, 2),
            "capabilities":    capabilities[:10],
            "limitations":     limitations[:5],
            "frontiers":       frontiers[:5],
            "next_boundary":   next_boundary,
            "memory_count":    mem_count,
            "memory_strength": round(avg_strength, 4),
            "identity_coherence": round(coherence, 4),
            "total_patterns":  len(weights),
            "growth_index":    self._growth_index(),
        }

        # Snapshot for trajectory tracking
        self._snapshots.append({
            "timestamp":    assessment["timestamp"],
            "capabilities": len(capabilities),
            "limitations":  len(limitations),
            "growth_index": assessment["growth_index"],
        })
        self._save()

        return assessment

    # ------------------------------------------------------------------
    # Internal
    # ------------------------------------------------------------------
    def _load_identity_vec(self) -> np.ndarray:
        path = os.path.expanduser("~/.vitalis_workspace/identity.npy")
        if os.path.exists(path):
            return np.load(path)
        return np.ones(self.kernel.dim, dtype=np.int8)

    def _identity_coherence(
        self, identity_vec: np.ndarray, capabilities: list
    ) -> float:
        """
        How aligned are current capabilities with the system's identity?
        High coherence = acting true to itself.
        """
        if not capabilities:
            return 0.5
        sims = []
        for pattern, _ in capabilities[:5]:
            cap_vec = self.kernel.vectorize_tokens(
                pattern.split("_"), positional=False
            )
            sims.append(self.kernel.similarity(identity_vec, cap_vec))
        return float(np.mean(sims))

    def _growth_index(self) -> float:
        """
        Single metric for overall growth.
        Combines: total patterns, avg weight, memory count.
        """
        weights = self.resonance.weights
        if not weights:
            return 0.0
        avg_w    = float(np.mean(list(weights.values())))
        n        = len(weights)
        mem      = len(self.hippocampus.all_slots())
        # Normalised growth index
        return round(float(np.log1p(n) * avg_w + np.log1p(mem) * 0.1), 4)

    def growth_trajectory(self) -> dict:
        """How has the system grown over time?"""
        if len(self._snapshots) < 2:
            return {"status": "Insufficient snapshots"}
        first = self._snapshots[0]
        last  = self._snapshots[-1]
        return {
            "snapshots":          len(self._snapshots),
            "growth_index_delta": round(
                last["growth_index"] - first["growth_index"], 4
            ),
            "capability_delta":   last["capabilities"] - first["capabilities"],
            "limitation_delta":   last["limitations"]  - first["limitations"],
            "time_span_hours":    round(
                (last["timestamp"] - first["timestamp"]) / 3600, 2
            ),
        }

    def report(self) -> dict:
        assessment = self.assess()
        return {
            "age_hours":          assessment["age_hours"],
            "growth_index":       assessment["growth_index"],
            "capabilities":       len(assessment["capabilities"]),
            "limitations":        len(assessment["limitations"]),
            "next_boundary":      assessment["next_boundary"],
            "identity_coherence": assessment["identity_coherence"],
            "trajectory":         self.growth_trajectory(),
        }