CNexus-COS-v2-Core

Patch CNexus v2 COS kernel.py to a 6-step cognitive loop with Skill Registry, Context Fusion Layer, and Cross-Skill Interaction Graph, enhancing context-awar...

CNexus COS Runtime Patch

Upgrade a CNexus v2 Cognitive OS kernel.py through 6 upgrade layers (P0.1 → P2-C) without altering external APIs, boot sequence, or kernel run loop structure.

Layer Stack

P0.3  Cognitive Loop (6-step: OBSERVE → COGNIZE → DECIDE → SPEAK → STORE → REFLECT)
  ↓
P1    Goal/Planner/Utility (baseline, untouched by this patch)
  ↓
P2-A  Skill Registry (178 skills, boot-registered Skill objects)
  ↓
P2-B  Context Fusion Layer (intent + memory + state → policy selection)
  ↓
P2-C  Cross-Skill Interaction Graph (influence propagation with distance decay)

Target

Single file: D:\类脑记忆\cnexus\cnexus_os\kernel.py

Boot data (read-only, generated by CNexus Compiler v2 Stage 5):

• D:\类脑记忆\cnexus\cnexus_final_skill_classification.json
• D:\类脑记忆\cnexus\cnexus_router_ready_skillset.json
• D:\类脑记忆\cnexus\cnexus_skill_graph_clean.json

Constraints (hard)

• ❌ No changes to P0.3 cognition loop structure (6-step remains)
• ❌ No introduction of CNexus original source code
• ❌ No real skill logic implementation
• ❌ No external dependencies (LLM / DB / embedding)
• ❌ No changes to boot.py or API interface (run(), status(), memory_dump(), reset(), shutdown())
• ❌ No governance injection
• ❌ No linear pipeline → only 6-step cognitive loop

Upgrade Sequence

Apply in order. Each layer is verified before proceeding to the next.

---

P0.1 — Structural Reorganization (6-step Cognitive Loop)

Replace the 4-step pipeline (route → DAG → execute → writeback) with a 6-step cognitive loop.

What to change in kernel.py:

Add self.cog canonical state object in __init__:

self.cog = {
    'observation':   None,
    'context':       None,
    'decision':      None,
    'last_response': None,
    'trace':         [],
    'iteration':     0,
    'cog_state': {
        'active_intent':      None,
        'last_content_hash':  '',
        'accumulated_weight': 0.0,
        'recall_strength':    0.0,
        'total_observations': 0,
        'last_intent':        None,
        'consecutive_same_intent': 0,
    },
}

Add 6 step methods:

1. _observe(input_text) — Normalize input, detect empty. Returns {type, raw, normalized, is_empty}. Sets self.cog['observation'].

2. _cognize(observation) — Intent extraction (store/recall/execute/converse via keyword matching). Weighted recall from self.memory_store. Updates cog_state fields.

3. _decide(context) — Strategy selection (IDLE vs SPEAK vs RECALL). Sets self.cog['decision'].

4. _speak(decision) — Route via self.route(), build DAG via self.build_dag(), execute via _execute_skills(). Returns response dict. Sets self.cog['last_response'].

5. _store(response) — Write memory blocks. Writes skill_execution_trace block + individual memory_block per skill. Updates self.memory_store.

6. _reflect() — Compute belief adjustments, narrative entry, consecutive intent tracking. Increments self.cog['iteration'].

Rewrite run() from:

# old: ctx = self._cognize(input_text) → dec = self._decide(ctx) → ... (4-step)

to:

self.cog['trace'] = []
obs = self._observe(input_text)
ctx = self._cognize(obs)
dec = self._decide(ctx)
resp = self._speak(dec)
stored = self._store(resp)
ref = self._reflect()

Trace detail formatting (in final return of run()):

Step	Detail
OBSERVE	type=... raw='...'
COGNIZE	intent=... recall_strength=... recall=N entries
DECIDE	strategy=... confidence=...
SPEAK	skills=N items ex=N miss=N mode=...
STORE	blocks_written=N
REFLECT	iter#N belief=... intent=... consec=N

Verify: Run 3 inputs (empty, normal, repeated). check trace has all 6 steps.

---

P0.2 — Weighted Cognitive State Update

Upgrade _cognize() recall weighting:

# Matching rules (in priority order):
if entry_hash[:4] == input_prefix:
    weight = 0.8    # content similarity
elif entry.get('skill') == intent:
    weight = 0.6    # skill/intent match
else:
    age = iteration - entry.get('iteration', 0)
    weight = max(0.1, 0.5 - age * 0.05)  # time decay

recall_strength = sum(weights) / max(len(recall), 1)
accumulated_weight += recall_strength * 0.1

Verify: Run 3 same-intent inputs; verify recall_strength shows meaningful differentiation.

---

P2-A — Skill Registry + Stub Execution

Replace the hash-based execution in _execute_skills() with a proper registry.

Skill class (add before CNexusOSKernel):

class Skill:
    def __init__(self, name, module=None, hint=None):
        self.name = name
        self.module = module
        self.hint = hint

    def run(self, context):
        return {
            'skill': self.name,
            'status': 'executed_stub',
            'module': self.module,
            ...
        }

In boot(), populate self.skill_registry:

self.skill_registry = {}
for name, entry in self.skills.items():
    self.skill_registry[name] = Skill(
        name=name,
        module=entry.get('source_module', ''),
        hint=entry.get('recovery_hint', ''),
    )

Rewrite _execute_skills() to iterate DAG nodes:

for node in dag['nodes']:
    skill_obj = self.skill_registry.get(skill_name)
    if skill_obj is None:
        ...  # missing entry
    run_result = skill_obj.run(context)

Add skill_execution_trace block in _store():

exec_key = hashlib.md5(f"exec_trace:{iteration}:{input}".encode()).hexdigest()[:8]
self.memory_store[exec_key] = {
    'block_id': exec_key,
    'type': 'skill_execution_trace',
    'content': {'execution_results': [...]},
    ...
}

Verify: status() shows execution: {skill_count, executed, missing, mode}.

---

P0.3 — REFLECT Enhancement

Upgrade _reflect() to produce 3 adjustment dimensions:

Belief — accumulated_weight adjustment:

if recall_strength > 0:
    belief_delta = recall_strength * 0.05
else:
    belief_delta = -0.01
accumulated_weight = clamp(accumulated_weight + belief_delta, 0.0, 2.0)

Narrative — Iteration log:

narrative_entry = {
    'iteration': iteration,
    'intent': intent,
    'strategy': strategy,
    'recall_strength': recall_strength,
    'latency': compute_latency(),
}
self._narrative_log.append(narrative_entry)  # keep last 20

Self-model — Consecutive intent tracking:

if prev_intent == intent:
    consecutive_same_intent = count_consecutive(narrative_log, intent)
else:
    consecutive_same_intent = 0

Verify: 4+ iterations with repeated intent show increasing belief and consecutive_same_intent.

---

P2-B — Context Fusion Layer

Upgrade Skill.run() from stub execution to context-aware fusion. The kernel loop uses Skill.run() — this is the only change point.

Replace Skill.run():

def run(self, context):
    intent = context.get('intent', 'unknown')
    state = context.get('cog_state', {})
    memory = context.get('memory_snapshot', [])
    goal = context.get('goal', None)

    alignment_score = self._compute_alignment(intent, goal)
    memory_weight = min(len(memory) * 0.05, 0.5)
    state_weight = state.get('accumulated_weight', 0) * 0.3
    total_score = alignment_score + memory_weight + state_weight

    if total_score > 0.8:          mode = 'high_confidence_execution'
    elif total_score > 0.4:        mode = 'contextual_execution'
    else:                          mode = 'low_confidence_stub'

    return {skill, module, mode, alignment_score, memory_weight,
            state_weight, total_score, intent, goal, output}

And _compute_alignment():

def _compute_alignment(self, intent, goal):
    if goal is None:                       return 0.2
    if intent == 'store' and 'memory' in str(goal): return 0.9
    if intent == 'recall' and 'retrieval' in str(goal): return 0.85
    if intent == 'execute':                return 0.6
    return 0.3

Update _execute_skills() context construction:

memory_items = list(self.memory_store.values())
skill_context = {
    'intent': self.cog.get('cog_state', {}).get('active_intent'),
    'cog_state': self.cog.get('cog_state'),
    'memory_snapshot': memory_items[-5:],
    'goal': None,  # placeholder — goal system not yet deployed
}

Verify: 3 scenarios (store/recall/execute) with 0, 3, 10 memory entries → verify total_score and mode change.

---

P2-C — Cross-Skill Interaction Graph

Add skill-to-skill influence propagation between _execute_skills() context construction and Skill.run().

Method _build_influence_map(parallel_groups):

all_skills = []
for group in parallel_groups:
    for sk in group['skills']:
        all_skills.append(sk)
influence_map = {}
for i in range(len(all_skills) - 1):
    a, b = all_skills[i], all_skills[i + 1]
    influence_map.setdefault(a, []).append({
        'target': b,
        'weight': round(0.5 * (0.7 ** i), 3),  # distance decay
        'type': 'sequential_dependency',
    })
self._influence_map = influence_map

Method _apply_cross_skill_influence(skill_name, context):

influence_score = 0.0
influence_tags = []
for src_sf, targets in self._influence_map.items():
    for t in targets:
        if t['target'] == skill_name:
            influence_score += t['weight']
            influence_tags.append(src_sf)
context['cross_skill'] = {
    'influence_score': round(influence_score, 3),
    'influenced_by': influence_tags,
}
return context

Wire into _execute_skills():

self._build_influence_map(dag.get('parallel_groups', []))
# ... build skill_context ...
for node in dag['nodes']:
    node_context = self._apply_cross_skill_influence(skill_name, dict(skill_context))
    result = skill_obj.run(node_context)

Initialize self._influence_map = {} in __init__ and reset().

Verify: 5-skill DAG shows influence propagation chain 0.5 → 0.35 → 0.245 → 0.171.

---

Verification Protocol

After each layer, run this verification:

k = CNexusOSKernel()
k.boot(class_file, router_file, graph_file)
r = k.run('store a memory about my project')
assert len(r['trace']) == 6, f"Expected 6 trace entries, got {len(r['trace'])}"
assert r['execution']['skill_count'] >= 1
assert r['memory']['total'] >= 1

Output Files

• Modified: D:\类脑记忆\cnexus\cnexus_os\kernel.py (single file only)
• Test scripts (local temp): C:\Users\<user>\AppData\Local\Temp\test_p*.py
• State persistence: D:\类脑记忆\cnexus\cnexus_state_store.json

Publish Package

发布包路径：D:\类脑记忆\cnexus\publish\cnexus-cos-core-v2\

cnexus-cos-core-v2/
├── kernel.py               ← 最终修正版内核
├── README.md               ← 接口说明书 + 快速启动
└── test_benchmark.py       ← 标准化验证脚本（6轮基准测试）

运行 test_benchmark.py 确认：

条件	预期
Belief 6轮后	0.0 ~ 0.5（无双重累积）
consec 意图切换时	精确清零
store 策略	SPEAK
recall 策略	RECALL
store 块 metadata	含 strategy/iteration/weight/decay_factor/reference_count

Signal Values for a Healthy Patch

Signal Values for a Healthy Patch

Metric	Expected
BOOT time	< 50ms
Skills loaded	178
Pure skills	~90
Trace steps per run	6
Latency per run	< 10ms
Memory blocks after 3 runs	15-25