الفصل السادس: هندسة النظم والمعمارية البرمجية الهجينة لنظام MathAgent والتحكم الصارم بالموارد¶
6.1 الطوبولوجيا المعمارية وتفكيك الطبقات الهندسية للنظام¶
6.1.1 مبدأ الفصل الصارم بين الاهتمامات (Strict Separation of Concerns)¶
يتطلب بناء نظام تدريس ذكي يعمل محلياً بكفاءة على أجهزة طرفية منخفضة المواصفات تصميماً معمارياً يتفادى التداخل بين منطق الواجهة الرسومية ومحرك اتخاذ القرار وعمليات التخزين المستدامة. يتبنى نظام MathAgent معمارية طبقية متعامدة ثلاثية المستويات تعتمد على نمط الوسيط المنسق (Mediator-Controller Pattern):
- طبقة العرض والتفاعل الحركي (Presentation & Kinesthetic Interaction Layer):
- المسؤولة الحصرية عن رصد الأحداث اللمسية، ورسم الإطارات الرسومية، وتفعيل الاهتزازات الحسية (Haptics).
- تعمل كواجهة "غبية" خالية من أي منطق استدلالي داخلي، وتقتصر وظيفتها على تمرير شفرات الأحداث اللمسية واستقبال تعليمات الرسم.
- طبقة الحوكمة السقراطية والتحكيم المحلي (Local Governance & Socratic Core Layer):
- النواة الصلبة للنظام التي تدير المخططات الحلقية الموجهة (DAG)، وتراقب الحمل المعرفي، وتحسب مؤشر التلاشي التدريجي، وتفرض قيود الجدار الناري البيداغوجي.
- طبقة الاستدامة والتسجيل الآمن (Persistence & Security Ledger Layer):
- المعنية بإدارة المعاملات التخزينية في قواعد البيانات المشفرة محلياً، وحفظ لقطات الحالة الذرية، وتجميع سجلات قياسات التتبع دون مقاطعة خيوط المعالجة الحية.
┌────────────────────────────────────────────────────────────────────────┐
│ طبقة العرض والتفاعل (Presentation & Kinesthetic Interface) │
│ شاشة اللمس - مؤشرات UDL - حظر لوحة المفاتيح - ردود الفعل الحسية │
└───────────────────────────────────┬────────────────────────────────────┘
│ (أحداث لمسية منفصلة / شفرات خام)
▼
┌────────────────────────────────────────────────────────────────────────┐
│ وسيط الأحداث المعماري (Core System Architecture Mediator) │
│ توجيه الرسائل والتحكيم وفصل التزامن │
└───────────────┬────────────────────────────────────────┬───────────────┘
│ │
▼ ▼
┌────────────────────────────────┐ ┌────────────────────────────────┐
│ طبقة الحوكمة السقراطية │ │ طبقة الاستدامة والأمان │
│ محرك DAG - الجدار الناري │ │ التشفير العتادي - التتبع xAPI │
│ حساب التلاشي - قيود الحمل CLT │ │ قاعدة بيانات SQLCipher المحلية │
└────────────────────────────────┘ └────────────────────────────────┘
تتواصل هذه الطبقات حصرياً عبر كائنات غير قابلة للتغيير (Immutable Data Transfer Objects - DTOs) عبر وسيط نظامي يمنع الاعتماديات الدائرية.
6.1.2 النمذجة البرمجية لوسيط الطبقات المعمارية (Architecture Core Mediator)¶
يقوم هذا المكون بتنظيم تدفق الرسائل بين طبقات النظام وعزل مسؤولية كل طبقة لمنع تداخل العمليات الحسابية مع خط تصيير الشاشة.
# =====================================================================
# Academic Module: architecture_core.py
# Scope: codebase/complet/system
# Doctoral Research: AI Agent MathAgent - Chapter 06 (Section 6.1)
# =====================================================================
from typing import Dict, Any, Callable
from enum import Enum
class SystemLayer(Enum):
PRESENTATION = "presentation"
SOCRATIC_GOVERNANCE = "socratic_governance"
PERSISTENCE_SECURITY = "persistence_security"
# START: REF_ARCHITECTURE_CORE
class CoreMediator:
"""
الوسيط المعماري للنظام (Core Architecture Mediator).
يفصل بين طبقات العرض الرسومي ومحرك الاستدلال وقاعدة البيانات المحلية،
ويمنع أي اقتران مباشر بين واجهة المستخدم ومنطق المعالجة السقراطية.
"""
def __init__(self):
self._layer_handlers: Dict[SystemLayer, Callable[[Dict[str, Any]], Dict[str, Any]]] = {}
def register_layer_handler(
self,
layer: SystemLayer,
handler: Callable[[Dict[str, Any]], Dict[str, Any]]
) -> None:
"""تسجيل المعالج المعتمد لكل طبقة معمارية."""
self._layer_handlers[layer] = handler
def route_layer_event(
self,
source_layer: SystemLayer,
target_layer: SystemLayer,
payload: Dict[str, Any]
) -> Dict[str, Any]:
"""
توجيه الحدث بين الطبقات مع تدقيق الحدود ومنع تدفق المعطيات غير المصرح بها.
"""
if target_layer not in self._layer_handlers:
return {
"routing_success": False,
"error": f"TARGET_LAYER_UNREGISTERED: {target_layer.value}",
"source": source_layer.value
}
handler = self._layer_handlers[target_layer]
# تغليف الرسالة في حاوية تحكم غير قابلة للتغيير
envelope = {
"origin": source_layer.value,
"target": target_layer.value,
"data": payload
}
response = handler(envelope)
return {
"routing_success": True,
"source": source_layer.value,
"target": target_layer.value,
"response": response
}
# END: REF_ARCHITECTURE_CORE
6.2 الحوكمة العتادية للذاكرة العشوائية وتطويع جامع المهملات تحت سقف 2 جيجابايت¶
6.2.1 قيود العتاد المدرسي وتحدي تخصيص الذاكرة¶
في الفصول الدراسية الواقعية المستهدفة، تعمل المنظومة على أجهزة لوحية اقتصادية ذات مواصفات عتادية متواضعة:
- سعة الذاكرة العشوائية الإجمالية: 2 جيجابايت (يقتطع نظام تشغيل أندرويد منها ما بين 1.1 إلى 1.4 جيجابايت لتشغيل خدمات النظام والاتصالات الراديوية).
- المساحة الحقيقية المتبقية لتطبيقات المستخدم: بين 400 إلى 600 ميجابايت كحد أقصى.
إذا تُركت بيئة تشغيل بايثون/الواجهة لآليات إدارة الذاكرة الافتراضية، فإن التراكم التدريجي لكائنات التتبع والمخططات الرسومية غير المفرغة يؤدي حتماً إلى تدخل مدير نفاد الذاكرة في نواة نظام التشغيل (Low-Memory Killer - LMK) لإغلاق التطبيق فجأة وبشكل قسري (SIGKILL)، مما يدمر جلسة التعلم ويفقد الطالبة تقدمها الدراسي.
┌────────────────────────────────────────────────────────────────────────┐
│ توزيع الذاكرة الفيزيائية (2GB RAM) │
├────────────────────────────────────────┬───────────────────────────────┤
│ نظام التشغيل أندرويد وخدمات العتاد │ المساحة المتاحة للمنظومة │
│ (1.2 - 1.4 GB) │ (400 - 600 MB) │
└────────────────────────────────────────┴───────────────┬───────────────┘
│
┌────────────────┴───────────────┐
│ سقف تشغيل MathAgent الصارم │
│ (أقصى حد: 250 MB) │
└────────────────────────────────┘
6.2.2 خوارزمية الجدولة القسرية لدورة جامع المهملات (Deterministic GC Scheduling)¶
يضع النظام سقفاً تشغيلياً صارماً لاستهلاك الذاكرة: \(\le 250\) ميجابايت طوال مدة الحصة المدرسية. ولتحقيق ذلك، يتم تعطيل الجريان التلقائي غير المنضبط لجامع المهملات (Garbage Collector) أثناء لحظات حل المسألة الحرجة، وتطبيق دورات جمع المهملات الانتقائية:
- أثناء تفكير ونقر المتعلمة: يُجمد جامع المهملات تلقائياً لتفادي توقف الشاشة المؤقت (Stop-The-World Latency Spikes) وضمان بقاء زمن الاستجابة اللمسية تحت 50 مللي ثانية.
- عند لحظات الانتقال البيداغوجي الطبيعية: تفعيل جمع المهملات القسري للجيلين الأول والثاني في اللحظة الزمنية الفاصلة بين الانتقال من مسألة إلى أخرى أو عند خمول التفاعل لأكثر من 5 ثوانٍ.
# =====================================================================
# Academic Module: memory_load_guard.py
# Scope: codebase/complet/system
# Doctoral Research: AI Agent MathAgent - Chapter 06 (Section 6.2)
# =====================================================================
import gc
import sys
from typing import Dict, Any
# START: REF_MEMORY_GOVERNOR
class MemoryGovernor:
"""
حارس الميزانية العتادية للذاكرة العشوائية (Memory Footprint Governor).
يراقب استهلاك الذاكرة ويفرض جدولة قطعية لجامع المهملات دون مقاطعة التفاعل اللمسي.
"""
HARD_CEILING_MB: float = 250.0 # السقف الأقصى المسموح به في الأجهزة الاقتصادية
GC_THRESHOLD_RATIO: float = 0.85 # عتبة بدء الإفراغ الاستباقي (85% من السقف)
def __init__(self):
# تعطيل دورات الجمع التلقائي غير المتوقعة لتفادي تجمد الواجهة
gc.disable()
self._is_critical_section: bool = False
def enter_critical_interaction(self) -> None:
"""الإعلان عن بدء نافذة تفاعلية حرجة تتطلب زمناً فائق السرعة."""
self._is_critical_section = True
def exit_critical_interaction(self) -> None:
"""انتهاء النافذة الحرجة والانتقال لمرحلة الخمول النسبي."""
self._is_critical_section = False
def get_estimated_footprint_mb(self) -> float:
"""تقدير استهلاك الذاكرة التقريبي للكائنات الحية داخل البيئة."""
# تقدير استناداً إلى تعداد كائنات بايثون وتوزيع الأجيال
tracked_objects = gc.get_count()
estimated_mb = (sum(tracked_objects) * 128) / (1024 * 1024)
return round(estimated_mb, 2)
def enforce_budget(self) -> Dict[str, Any]:
"""
التحقق من الميزانية وإجراء تفريغ قطعي للكائنات غير النشطة عند الضرورة.
"""
current_mb = self.get_estimated_footprint_mb()
over_threshold = current_mb >= (self.HARD_CEILING_MB * self.GC_THRESHOLD_RATIO)
gc_executed = False
if over_threshold and not self._is_critical_section:
# تفعيل الجمع اليدوي المباشر للأجيال الثلاثة بالتتابع
gc.collect(2)
gc_executed = True
current_mb = self.get_estimated_footprint_mb()
return {
"current_footprint_mb": current_mb,
"ceiling_limit_mb": self.HARD_CEILING_MB,
"gc_forced": gc_executed,
"in_critical_section": self._is_critical_section,
"status": "NORMAL_BUDGET" if current_mb <= self.HARD_CEILING_MB else "MEMORY_CEILING_BREACHED"
}
# END: REF_MEMORY_GOVERNOR
6.3 معمارية تعدد المسارات (Multi-threading) وعزل مسار المعالجة عن واجهة المستخدم¶
6.3.1 معضلة قفل المفسر العام (GIL) وتجمد الشاشات اللمسية¶
تعتمد لغة بايثون في تنفيذها المعياري (CPython) على قفل المفسر العام (Global Interpreter Lock - GIL)، والذي يمنع تنفيذ أكثر من خيط برمجي لتعليمات بايثون الأصلية في اللحظة الزمنية ذاتها.
عند قيام النظام بالتحقق الصوري من خطوة رياضية مركبة، أو تشفير حزمة تتبع في قاعدة البيانات، أو كتابة ملف على القرص، فإن تنفيذ هذه العمليات داخل خيط العرض الرسومي الأساسي (Main UI Thread) يؤدي فوراً إلى ظاهرة إسقاط الإطارات الرسومية (UI Frame Drops / Stutter) وتأخر استجابة اللمس لتتجاوز 300 مللي ثانية، وهو ما ينتهك ميزانية الاستجابة الحركية (المحددة بـ 50 مللي ثانية كحد أقصى في الفصل الثاني).
[مسار واجهة المستخدم UI Thread] ────▶ التقاط النقر اللمسي ────▶ تصيير فوري (Frame Blit < 15ms)
│
(تفويض غير متزامن عبر طابور رسائل معزول)
│
▼
[مسار المعالجة الخلفية Worker] ────▶ حل منطق الـ DAG ────▶ تشفير التتبع ────▶ كتابة السجلات
6.3.2 المعمارية المعزولة لمهام المعالجة والتسجيل غير المتزامن¶
لضمان انسيابية العرض التفاعلي على المعالجات القديمة ذات الأنوية الرباعية، يعتمد MathAgent على نمط العزل عبر طوابير الأحداث المنفصلة (Producer-Consumer Threaded Isolation):
- مسار واجهة العرض الرسومية (Main UI Thread): مخصص حصرياً لتصيير الشاشة ومعالجة الأحداث اللمسية بدقة 60 إطاراً في الثانية دون إجراء أي عمليات حسابية معقدة أو معاملات قرصية.
- مسار الحساب والاستدلال السقراطي (Inference Worker Thread): يستقبل معرفات الأحداث اللمسية عبر طابور خفيف الحجم، ويحلل الانتقال في مخطط DAG، ثم يعيد النتيجة عبر دالة رد نداء سريعة (Callback).
- مسار الكتابة على القرص والتتبع (I/O & Persistence Thread): يتولى عمليات التشفير والتعامل مع قاعدة البيانات الموضعية في الخلفية بأدنى أولوية معالجة (Low Thread Priority) لضمان عدم منافسة مسار الرسوم على وقت المعالج.
# =====================================================================
# Academic Module: multithread_isolator.py
# Scope: codebase/complet/system
# Doctoral Research: AI Agent MathAgent - Chapter 06 (Section 6.3)
# =====================================================================
import queue
import threading
from typing import Dict, Any, Callable
# START: REF_PROCESS_ISOLATOR
class ProcessIsolator:
"""
عازل المسارات البرمجية وتأمين سلاسة الواجهة الرسومية (UI Thread Isolator).
يفصل بين مسار تصيير الشاشة ومسارات الاستدلال الحسابي والكتابة على القرص.
"""
def __init__(self, socratic_inference_callable: Callable[[Dict[str, Any]], Dict[str, Any]]):
self._inference_callback = socratic_inference_callable
self._inbound_queue: queue.Queue = queue.Queue(maxsize=32)
self._outbound_queue: queue.Queue = queue.Queue(maxsize=32)
self._worker_thread = threading.Thread(target=self._worker_loop, daemon=True)
self._is_running = True
self._worker_thread.start()
def _worker_loop(self) -> None:
"""حلقة العمل المستقلة في المسار الخلفي المعزول."""
while self._is_running:
try:
task = self._inbound_queue.get(timeout=0.1)
except queue.Empty:
continue
event_payload = task.get("payload", {})
# تنفيذ الاستدلال والتحقق الرياضي بعيداً عن مسار الرسم الرسومي
inference_result = self._inference_callback(event_payload)
self._outbound_queue.put({
"task_id": task.get("task_id"),
"result": inference_result
})
self._inbound_queue.task_done()
def submit_touch_event_async(self, task_id: str, touch_payload: Dict[str, Any]) -> bool:
"""إرسال الحدث اللمسي للمسار الخلفي دون حظر المسار الرئيسي للواجهة."""
try:
self._inbound_queue.put_nowait({
"task_id": task_id,
"payload": touch_payload
})
return True
except queue.Full:
return False
def poll_resolved_state_nonblocking(self) -> Dict[str, Any]:
"""فحص غير معطل لمسار العرض الرسومي لاستلام التحديثات فور جاهزيتها."""
try:
return self._outbound_queue.get_nowait()
except queue.Empty:
return {"status": "PENDING_COMPUTATION"}
def terminate_gracefully(self) -> None:
"""إيقاف المسار الخلفي بأمان عند إنهاء الجلسة التعليمية."""
self._is_running = False
if self._worker_thread.is_alive():
self._worker_thread.join(timeout=1.0)
# END: REF_PROCESS_ISOLATOR
6.4 موثوقية النظام وخوارزمية التعافي الذاتي من الانهيار المفاجئ (In-Class Crash Resilience)¶
6.4.1 بيئة الفصل ومخاطر فقدان البيانات اللحظي¶
تتعرض الأجهزة اللوحية في الفصول الدراسية الحقيقية لحوادث تشغيل مفاجئة ومتكررة:
- نفاد شحن البطارية فجأة أثناء حل الطالبة للمسألة.
- قيام الطالبة بإغلاق التطبيق بطريق الخطأ أو الضغط على زر التشغيل الصلب.
- حدوث إعادة تشغيل قسرية لنظام التشغيل نتيجة السخونة أو أخطاء برمجية عتادية.
إذا كان النظام يعتمد على حفظ البيانات في الذاكرة الحية فقط حتى نهاية الجلسة، فإن أي انهيار طارئ يؤدي إلى فقدان كامل لخطوات الحل ومؤشرات التقييم، مما يولد إحباطاً شديداً لدى المتعلمة ويفسد القياس التجريبي للبحث.
6.4.2 بروتوكول لقطات الحالة الذرية (Atomic State Snapshots)¶
لمواجهة هذا التحدي، يبتكر MathAgent بروتوكول التفريغ الذري للحالة (Atomic State Journaling):
- تسجيل الحالة بعد كل انتقال ديداكتيكي: فور استقرار الطالبة في عقدة جديدة بمخطط DAG، يتم توليد لقطة مجهرية للحالة (Micro-Snapshot) لا يتجاوز حجمها 512 بايت تحتوي على: معرف العقدة الحالية، ومؤشر التلاشي اللحظي، وعدد المحاولات السابقة، وبصمة التحقق التشفيرية.
- الكتابة الآمنة عبر الاستبدال الذري (Atomic File Replacement): تُكتب اللقطة أولاً في ملف مؤقت معزول، وفور اكتمال كتابتها بنجاح وتفريغها فيزيائياً على بطاقة التخزين الوميضية (Flush to Storage)، يتم استبدال ملف الحالة القديم بعملية نظامية ذرية لا تقبل التجزئة.
- التعافي اللحظي في زمن يقل عن ثانية واحدة: عند إعادة تشغيل التطبيق بعد أي انهيار مفاجئ، يكتشف النظام وجود لقطة حالة صالحة، ويستعيد واجهة الطالبة في الموضع نفسه الذي توقفت عنده بدقة المللي ثانية دون أن تشعر بأي ضياع لجهدها.
[انتقال إلى عقدة جديدة في الـ DAG]
│
▼
[توليد لقطة مجهرية Micro-Snapshot (512 Bytes)]
│
▼
[كتابة في ملف مؤقت معزول snapshot.tmp] ──▶ استدعاء Flush فيزيائي للقرص
│
▼
[استبدال ذري للملف الرئيسي: snapshot.tmp ──▶ active_state.dat]
│
(في حال الانهيار: استرجاع الحالة فوراً خلال أقل من ثانية واحدة عند الإقلاع)
# =====================================================================
# Academic Module: crash_recovery_daemon.py
# Scope: codebase/complet/system
# Doctoral Research: AI Agent MathAgent - Chapter 06 (Section 6.4)
# =====================================================================
import json
import hashlib
from typing import Dict, Any, Optional
# START: REF_CRASH_WATCHDOG
class CrashWatchdog:
"""
محرك التعافي الذاتي من الانهيارات وحفظ الحالة الذرية (Atomic Crash Recovery Daemon).
يضمن استعادة الجلسة التعليمية للطالبة بدقة في حال انقطاع التيار أو إغلاق التطبيق.
"""
def __init__(self, in_memory_mock_storage: Optional[Dict[str, str]] = None):
# محاكاة مستودع التخزين المحلي الآمن
self._storage = in_memory_mock_storage if in_memory_mock_storage is not None else {}
self.SNAPSHOT_KEY = "CURRENT_LEARNER_SNAPSHOT"
self.TEMP_SNAPSHOT_KEY = "CURRENT_LEARNER_SNAPSHOT_TMP"
def _generate_state_signature(self, state_dict: Dict[str, Any]) -> str:
"""توليد بصمة تشفيرية للتحقق من سلامة اللقطة المخزنة وعدم تلفها."""
serialized = json.dumps(state_dict, sort_keys=True)
return hashlib.sha256(serialized.encode("utf-8")).hexdigest()
def commit_atomic_snapshot(self, learner_state: Dict[str, Any]) -> bool:
"""
تنفيذ عملية الحفظ الذري المزدوج للحالة البيداغوجية اللحظية.
"""
payload = {
"state_data": learner_state,
"integrity_signature": self._generate_state_signature(learner_state)
}
serialized_payload = json.dumps(payload)
# 1. الكتابة في السجل المؤقت أولاً
self._storage[self.TEMP_SNAPSHOT_KEY] = serialized_payload
# 2. الاستبدال الذري للسجل الدائم
self._storage[self.SNAPSHOT_KEY] = self._storage[self.TEMP_SNAPSHOT_KEY]
del self._storage[self.TEMP_SNAPSHOT_KEY]
return True
def restore_last_valid_state(self) -> Dict[str, Any]:
"""
فحص السجل التخزيني والتعافي اللحظي من آخر حالة تشغيلية متماسكة.
"""
raw_snapshot = self._storage.get(self.SNAPSHOT_KEY)
if not raw_snapshot:
return {
"recovered": False,
"reason": "NO_PREVIOUS_SNAPSHOT_FOUND"
}
try:
payload = json.loads(raw_snapshot)
state_data = payload.get("state_data", {})
recorded_sig = payload.get("integrity_signature", "")
# التحقق من عدم تلف البيانات أثناء الانقطاع المفاجئ
if self._generate_state_signature(state_data) != recorded_sig:
return {
"recovered": False,
"reason": "SNAPSHOT_INTEGRITY_COMPROMISED"
}
return {
"recovered": True,
"state_data": state_data,
"recovery_status": "RESTORED_SEAMLESSLY_WITH_ZERO_DATA_LOSS"
}
except Exception as e:
return {
"recovered": False,
"reason": f"CORRUPTED_SERIALIZATION: {str(e)}"
}
# END: REF_CRASH_WATCHDOG
6.5 خلاصة الفصل السادس والتمهيد المعماري للفصل السابع¶
يقدم هذا الفصل المعالجة الهندسية لنظام التشغيل والموثوقية العتادية لـ MathAgent في البيئات الميدانية الصعبة:
- تأسيس الطوبولوجيا المعمارية ثلاثية المستويات عبر وسيط موحد يضمن العزل المطلق بين طبقة العرض وطبقة الاستدلال السقراطي وطبقة التخزين.
- فرض الحوكمة الصارمة للذاكرة العشوائية تحت سقف 250 ميجابايت والتحكم اليدوي في دورات جامع المهملات لمنع تجمد الواجهة وتفادي الإغلاق القسري للنظام.
- ابتكار معمارية تعدد المسارات المعزولة التي تفصل مسار الرسم اللمسي عن مسارات الاستدلال والكتابة القرصية، لضمان استجابة لمسية لحظية تحت عتبة 50 مللي ثانية.
- تشييد منظومة التعافي الذاتي من الانهيار (Crash Resilience) القائمة على الحفظ الذري المزدوج للحالة البيداغوجية، مما يضمن استعادة جلسة الطالبة فورا وبدقة تامة في حال انقطاع الطاقة أو تعطل العتاد.
تمهد هذه المعمارية التحتية الراسخة الانتقال المباشر نحو الفصل السابع، حيث سننتقل من تشغيل النظام وإدارة موارده إلى هندسة خط الاستيعاب الآلي للمحتوى المدرسي (Ingestion Pipeline)، وتحويل الرسوم والمخططات الهندسية إلى شفرات برهانية، وبروتوكول التحقق البشري وتجميد التصميم البيداغوجي المعتمد.
---
### المكتسبات المعمارية بعد إتمام الفصل السادس:
1. **الفصل 06 (`thesis/chapters/chap_06.md`)** أصبح مكتملاً بنسبة 100% باللغة العربية الفصحى وبصياغة أكاديمية وهندسية دقيقة.
2. تم استخراج 4 كبسولات برمجية متخصصة ومستقلة تشكل البنية التحتية للنظام في حزمة `codebase/complet/system/`:
* `architecture_core.py` (`CoreMediator`)
* `memory_footprint_guard.py` (`MemoryGovernor`)
* `multithread_isolator.py` (`ProcessIsolator`)
* `crash_recovery_daemon.py` (`CrashWatchdog`)