diff --git a/.gitignore b/.gitignore index efdae15..c8d5630 100644 --- a/.gitignore +++ b/.gitignore @@ -14,3 +14,9 @@ venv/ server.log tests/phase5_ordeal/mock_db.json tests/phase5_ordeal_v2/ + +# Memory Thread Enterprise Data +.mt/ +vault.json +audit_ledger.jsonl +authority_grants.jsonl diff --git a/docs/COGNITIVE_GALAXY.md b/docs/COGNITIVE_GALAXY.md new file mode 100644 index 0000000..039132a --- /dev/null +++ b/docs/COGNITIVE_GALAXY.md @@ -0,0 +1,95 @@ +# Cognitive Galaxy Schema: OLAP for Cognition + +> "Memory Thread normalizes truth the way data warehouses normalize facts." + +## The Core Insight + +Most AI memory systems follow a flat pattern: **Flatten → Embed → Forget Source**. +Memory Thread (MT) follows a structural pattern: **Normalize → Relate → Reason**. + +This structure maps directly to **Data Warehouse Galaxy Schemas**, but applied to *cognition* instead of *analytics*. + +## The Schema Mapping + +| Data Warehouse Concept | Memory Thread Implementation | Description | +|------------------------|------------------------------|-------------| +| **Fact Table** | **Immutable Events** | The raw, undisputed reality. Code files, logs, sensor dumps. Append-only, versioned, no opinion. | +| **Dimension Table** | **Derived Beliefs** | Interpretations, summaries, and meanings derived from facts. Subject to decay, perspective, and contradiction. | +| **Surrogate Key** | **Entity ID** | The stable identifier linking diverse observations to a single conceptual entity. | +| **Slowly Changing Dimension (SCD)** | **Memory Decay / Freshness** | How beliefs evolve over time (Type 2 SCD). | +| **Lineage** | **Provenance Envelope** | The `derived_from` metadata tracking exactly which Fact generated which Belief. | +| **Rollback** | **Event Replay** | Deterministic reconstruction of state at any point in time. | + +## Structural Visualization + +```mermaid +erDiagram + FACT_SOURCE ||--o{ EVENT_LOG : generates + EVENT_LOG ||--|{ COGNITIVE_JOIN : feeds + AGENT_DIMENSION ||--|{ COGNITIVE_JOIN : interprets + + FACT_SOURCE { + string uri "file://auth_service.py" + string version "v1.2" + blob content "Raw Code/Text" + } + + EVENT_LOG { + uuid event_id + timestamp t + string payload "The objective reality" + } + + AGENT_DIMENSION { + string role "Security Auditor" + float authority_score "0.95" + string context "Security Review 2024" + } + + COGNITIVE_JOIN { + uuid entity_id + string belief "Critical Security Boundary" + float confidence + string provenance "Derived from Event X by Agent Y" + } +``` + +## The "Galaxy" Concept + +In a Galaxy Schema, multiple Fact Tables share Dimensions. In MT, **Multiple Belief Systems (Dimensions)** coexist over the same **Facts**. + +### Example: The "Auth Service" Fact + +**FACT:** `src/auth_service.py` (Content hash: `abc1234`) + +1. **Dimension A (Coder Agent):** + * *Belief:* "Handles JWT token parsing." + * *Confidence:* 0.9 + * *Action:* Refactor for performance. + +2. **Dimension B (Security Agent):** + * *Belief:* "Legacy OAuth implementation; potential vulnerability." + * *Confidence:* 0.7 + * *Action:* Flag for audit. + +3. **Dimension C (Architect Agent):** + * *Belief:* "Core infrastructure component." + * *Authority:* High + * *Action:* Protect from deletion. + +**Result:** No conflict. Just different "Cognitive Joins" on the same truth. + +## OLAP Operations for Cognition + +Because we have this structure, we can perform OLAP-style operations on memory: + +* **SLICE (by Source):** "Show me all beliefs derived from `auth_service.py`." +* **DICE (by Authority):** "Show me beliefs about `auth_service.py` held by agents with `Authority > 0.8`." +* **DRILL DOWN:** "Show me the raw event log that led to this belief." +* **ROLL UP:** "Summarize the system architecture based on all high-confidence beliefs." + +## Terminology + +* **Cognitive Galaxy Schema:** The overarching architectural pattern. +* **Truth-Normalized Memory:** The data storage strategy (store facts once, reference many). +* **Epistemic Star:** A specific cluster of beliefs surrounding a single entity. diff --git a/docs/PRIME_RULE.md b/docs/PRIME_RULE.md new file mode 100644 index 0000000..c1b2686 --- /dev/null +++ b/docs/PRIME_RULE.md @@ -0,0 +1,54 @@ +# The Prime Rule: Cognitive Governance + +> "MT only persists what can be replayed, audited, and justified without an LLM." + +## 1. The Prime Rule (Non-Negotiable) + +If something cannot survive deterministic replay, it does not belong in MT core storage. +Everything else is derived. + +## 2. Two Ingestion Classes + +MT enforces a strict separation between **Ontology (What Is)** and **Epistemology (What We Know)**. + +### Class A: Canonical Truth (Facts) + +These are external reality snapshots. + +* **Examples:** Source code, specs, logs, sensor dumps, user messages. +* **Properties:** Immutable, Versioned, Content-Addressed, No "Cleaning". +* **Constraint:** Must be ingestible without an LLM. + +### Class B: Epistemic Artifacts (Beliefs) + +These are interpretations, summaries, and conclusions derived from facts. + +* **Examples:** Code summaries, classifications, confidence scores, hypotheses. +* **Properties:** Mutable, Decaying, Subjective, Contradictory. +* **Constraint:** Must have explicit **Provenance**. + +## 3. Forbidden Patterns (Poison Control) + +The following must **NEVER** be persisted as Truth (Class A): + +1. **Embeddings as Truth:** Vectors are derived artifacts, not facts. +2. **Chain-of-Thought:** Internal reasoning traces are transient runtime noise. +3. **Unattributed Beliefs:** "The system thinks X" without an Agent ID and Source ID. +4. **Confidence without Source:** A number without a derivation path is meaningless noise. + +## 4. Enforcement Policy + +The `SecureMemoryClient` enforces these rules at runtime: + +| Violation | Action | +|-----------|--------| +| Belief stored as Fact | ❌ **Hard Error** | +| Missing `derived_from` on Belief | ❌ **Hard Error** | +| Fact with Confidence/Authority | ❌ **Hard Error** | +| Deprecated `remember()` usage | ⚠️ **Warning + Audit** | + +## 5. Self-Observation + +MT can ingest its own logs (e.g., `audit_ledger.jsonl`) as **Facts**. +MT must **never** treat its own previous outputs as canonical truth by default. +**Self-Observation ≠ Self-Belief.** diff --git a/docs/enterprise_rbac_design.md b/docs/enterprise_rbac_design.md new file mode 100644 index 0000000..befd604 --- /dev/null +++ b/docs/enterprise_rbac_design.md @@ -0,0 +1,86 @@ +# RBAC & Authority Design for Memory Thread Enterprise + +## 1. Role Hierarchy & Permissions + +We will implement a Role-Based Access Control (RBAC) system defined in a configuration structure (simulating a policy file). + +### Roles + +| Role | Clearance Level | Description | +| :--- | :--- | :--- | +| **GUEST** | 0 | Public access only. Can read `public` namespace. | +| **EMPLOYEE** | 1 | Standard internal access. Can read/write `team_*` namespaces. | +| **DEVELOPER** | 2 | Technical access. Can read/write `tech_*`, read `product`. | +| **RESEARCHER** | 3 | Cross-domain access. Read ALL. Write `research`. | +| **EXECUTIVE** | 4 | Strategic access. Full Read/Write/Override power. | + +### Domains (Namespaces) + +* `public`: Accessible by everyone. +* `team_general`: Accessible by Employees+. +* `tech_core`: Accessible by Developers, Researchers, Execs. +* `finance_secret`: Accessible by Executives only. +* `research_lab`: Accessible by Researchers, Execs. + +--- + +## 2. The "Firewall" Logic (Read Access) + +When a `recall()` or `chat()` happens, the Firewall checks: + +1. **Direct Namespace Access:** Does user have `READ` permission on the memory's namespace? +2. **Clearance Level:** Is the memory tagged with a clearance level higher than the user? + * *Note: In the Core SDK, we store `clearance` in the memory's metadata/payload.* + +**Rule:** `IF (User.Roles allows Namespace) AND (User.Clearance >= Memory.Clearance) THEN Access Granted.` + +--- + +## 3. The "Truth Authority" Logic (Write Access) + +When a `remember()` happens, we calculate the `authority` (0.0 - 1.0) passed to the Core SDK based on the User's Role and the Domain they are writing to. + +**Matrix:** + +| User Role | Target Domain | Authority Score | Logic | +| :--- | :--- | :--- | :--- | +| **EXECUTIVE** | Any | **0.95** | Strategic override. | +| **RESEARCHER**| `research_lab` | **0.90** | Expert domain. | +| **RESEARCHER**| `tech_core` | **0.50** | Observer. | +| **DEVELOPER** | `tech_core` | **0.90** | Expert domain. | +| **DEVELOPER** | `finance_secret`| **0.00** | (Write Denied) | +| **EMPLOYEE** | `team_general` | **0.60** | Standard input. | +| **GUEST** | `public` | **0.10** | Low trust. | + +* **Conflict Resolution:** If an *Employee* says "Sky is Green" (Auth 0.6) and an *Executive* says "Sky is Blue" (Auth 0.95), the Core SDK's math naturally resolves "Blue" as the truth. +* **Decay:** Higher authority memories decay slower (managed by core, but we can influence initial freshness). + +--- + +## 4. Implementation Strategy (No Core Changes) + +1. **`AccessControlService` (New Class):** + * Holds the hardcoded Policy (the matrix above). + * `calculate_write_authority(user, namespace) -> float` + * `can_read(user, memory_namespace, memory_metadata) -> bool` + +2. **`SecureMemoryClient` (Wrapper Class):** + * Wraps `MemoryClient`. + * **Input:** `user_id`, `role`. + * **On `remember(content, namespace)`:** + * Call `AccessControlService` to get authority. + * Inject `clearance_level` into the `metadata` of the memory (Core SDK stores payload/metadata). + * Call `CoreSDK.remember(content, authority=calculated_auth)`. + * **On `recall(query)`:** + * Call `CoreSDK.recall(query)`. + * Iterate results. + * Filter out any memory where `AccessControlService.can_read(...)` is False. + * Return filtered list (or "[REDACTED]" placeholders). + +## 5. TUI Integration + +* **New Command:** `/login ` (Simulates switching user token). +* **Visuals:** + * Display current "Security Clearance" in the footer. + * Show `[REDACTED]` for memories the current user shouldn't see. + * Show "Authority: High/Med/Low" indicators on messages. diff --git a/memory_thread/models/provenance.py b/memory_thread/models/provenance.py new file mode 100644 index 0000000..28e2719 --- /dev/null +++ b/memory_thread/models/provenance.py @@ -0,0 +1,64 @@ +from dataclasses import dataclass, field +from typing import Optional, Dict +from datetime import datetime +import uuid + +@dataclass +class Actor: + user_id: str + role: str + agent_id: Optional[str] = None + + def to_dict(self) -> Dict: + return { + "user_id": self.user_id, + "role": self.role, + "agent_id": self.agent_id + } + +@dataclass +class Origin: + client_id: str + session_id: Optional[str] = None + machine_id: Optional[str] = None + + def to_dict(self) -> Dict: + return { + "client_id": self.client_id, + "session_id": self.session_id, + "machine_id": self.machine_id + } + +@dataclass +class Scope: + namespace: str + domain: Optional[str] = None + project_id: Optional[str] = None + + def to_dict(self) -> Dict: + return { + "namespace": self.namespace, + "domain": self.domain, + "project_id": self.project_id + } + +@dataclass +class ProvenanceEnvelope: + """ + The immutable Identity & Namespace Envelope. + Must be embedded in every memory event payload. + """ + event_id: str = field(default_factory=lambda: str(uuid.uuid4())) + timestamp: str = field(default_factory=lambda: datetime.utcnow().isoformat()) + actor: Actor = None + origin: Origin = None + scope: Scope = None + + def to_dict(self) -> Dict: + return { + "event_id": self.event_id, + "timestamp": self.timestamp, + "actor": self.actor.to_dict() if self.actor else None, + "origin": self.origin.to_dict() if self.origin else None, + "scope": self.scope.to_dict() if self.scope else None + } diff --git a/memory_thread/nervous/access_control.py b/memory_thread/nervous/access_control.py new file mode 100644 index 0000000..ce17314 --- /dev/null +++ b/memory_thread/nervous/access_control.py @@ -0,0 +1,281 @@ + +""" +Enterprise Access Control Service (The Firewall). + +This module implements the "Intelligent Firewall" that sits between users and the raw memory store. +It enforces: +1. Role-Based Access Control (RBAC) +2. Domain-Specific Authority Scoring +3. Clearance Level Filtering +""" +from typing import Dict, List, Optional, Any +from dataclasses import dataclass +from enum import Enum, IntEnum +from memory_thread.models.provenance import ProvenanceEnvelope, Actor, Scope +from memory_thread.nervous.audit_ledger import ledger, AuditEvent +from memory_thread.nervous.authority_store import authority_store, AuthorityGrant + +class Grade(IntEnum): + E_CLASS = 0 # Public / Guest + C_CLASS = 1 # Internal / Employee + B_CLASS = 2 # Confidential / Developer + A_CLASS = 3 # Secret / Researcher + S_CLASS = 4 # Top Secret / Executive + SSS_CLASS = 5 # Godfather / Root + + def __str__(self): + return self.name + +@dataclass +class UserContext: + user_id: str + role: str + grade: Grade + domains: List[str] + + @property + def clearance(self): + return self.grade # Alias for backward compatibility + +class AccessControlService: + """ + The Single Source of Truth for Permissions and Authority. + Hardened for Pentagon-style Grade System. + """ + + # --- POLICY DEFINITIONS --- + + # Map Roles to Default Clearance Grades + ROLE_GRADES = { + "guest": Grade.E_CLASS, + "employee": Grade.C_CLASS, + "developer": Grade.B_CLASS, + "researcher": Grade.A_CLASS, + "executive": Grade.S_CLASS, + "godfather": Grade.SSS_CLASS # Hidden Role + } + + # Map Roles to Domain Access (Namespaces they can Read/Write) + # Format: "role": {"read": [domains], "write": [domains]} + # "*" is wildcard + ROLE_DOMAINS = { + "guest": { + "read": ["public"], + "write": ["public"] + }, + "employee": { + "read": ["public", "team_general"], + "write": ["team_general"] + }, + "developer": { + "read": ["public", "team_general", "tech_core", "product"], + "write": ["tech_core", "product"] + }, + "researcher": { + "read": ["*"], # Can read everything (subject to clearance) + "write": ["research_lab", "tech_core"] + }, + "executive": { + "read": ["*"], + "write": ["*"] + }, + "godfather": { + "read": ["*"], + "write": ["*"] + } + } + + # Authority Scoring Matrix: (Role, Domain) -> Score + AUTHORITY_MATRIX = { + ("godfather", "*"): 1.0, + ("executive", "*"): 0.95, + ("researcher", "research_lab"): 0.90, + ("researcher", "tech_core"): 0.50, + ("developer", "tech_core"): 0.90, + ("developer", "product"): 0.80, + ("employee", "team_general"): 0.60, + ("guest", "public"): 0.10 + } + + # --- PUBLIC API --- + + @classmethod + def create_context(cls, user_id: str, role: str) -> UserContext: + """Factory to create a user context from a role.""" + role = role.lower() + if role == "root": role = "godfather" # Alias + + if role not in cls.ROLE_GRADES: + role = "guest" + + return UserContext( + user_id=user_id, + role=role, + grade=cls.ROLE_GRADES[role], + domains=cls.ROLE_DOMAINS[role]["read"] + ) + + @classmethod + def calculate_write_authority(cls, user: UserContext, target_domain: str) -> float: + """ + Determines the Truth Score (Authority) for a write operation. + Returns 0.0 if write is denied. + """ + # 1. Check Dynamic Grants (Layer III) - Grants Override Permissions + dynamic_score = authority_store.get_score(user.role, target_domain) + if dynamic_score is not None: + return dynamic_score + + # 2. Check Write Permission (Static) + allowed_writes = cls.ROLE_DOMAINS[user.role]["write"] + if "*" not in allowed_writes and target_domain not in allowed_writes: + # AUDIT LOG: Write Denied + ledger.log(AuditEvent( + action_type="WRITE_DENIED", + actor_id=user.user_id, + role=user.role, + target=target_domain, + details={"reason": "domain_restriction"} + )) + return 0.0 # Denied + + # 3. Calculate Static Score + # Check specific rule first + score = cls.AUTHORITY_MATRIX.get((user.role, target_domain)) + if score is None: + # Check wildcard rule + score = cls.AUTHORITY_MATRIX.get((user.role, "*")) + + if score is None: + # Fallback default authority + score = 0.5 + + return score + + @classmethod + def grant_authority(cls, granter: UserContext, target_role: str, domain: str, score: float) -> bool: + """ + Dynamic Authority Grant (Governance). + Granter must have equal or higher authority in that domain to grant it. + """ + # 1. Check Granter's Power + granter_auth = cls.calculate_write_authority(granter, domain) + if granter_auth < score: + ledger.log(AuditEvent( + action_type="GRANT_DENIED", + actor_id=granter.user_id, + role=granter.role, + target=domain, + details={"reason": "insufficient_authority", "yours": granter_auth, "requested": score} + )) + return False + + # 2. Execute Grant + grant = AuthorityGrant( + granter_id=granter.user_id, + granter_role=granter.role, + target_role=target_role.lower(), + target_domain=domain, + score=score + ) + authority_store.add_grant(grant) + + # 3. Audit + ledger.log(AuditEvent( + action_type="AUTHORITY_GRANT", + actor_id=granter.user_id, + role=granter.role, + target=f"{target_role}:{domain}", + details={"score": score} + )) + return True + + @classmethod + def revoke_authority(cls, revoker: UserContext, target_role: str, domain: str) -> bool: + """ + Dynamic Revocation. + """ + # 1. Check Revoker's Power (Must be Admin/Exec or original granter ideally, simplified here) + if revoker.role not in ["executive", "godfather"]: + ledger.log(AuditEvent( + action_type="REVOKE_DENIED", + actor_id=revoker.user_id, + role=revoker.role, + target=domain, + details={"reason": "requires_exec_or_godfather"} + )) + return False + + # 2. Execute Revoke + authority_store.revoke(target_role.lower(), domain, revoker.role) + + # 3. Audit + ledger.log(AuditEvent( + action_type="AUTHORITY_REVOKE", + actor_id=revoker.user_id, + role=revoker.role, + target=f"{target_role}:{domain}" + )) + return True + + @classmethod + def can_read(cls, user: UserContext, envelope: Dict) -> bool: + """ + The Firewall Check. + Validates access against the PROVENANCE ENVELOPE. + """ + # Extract scope from envelope dict + # Structure: envelope = {_provenance: {scope: {domain: ...}}} + # Or sometimes the envelope is passed directly if we extracted it. + + # We assume 'envelope' is the provenance dict or the full memory payload containing it + provenance = envelope.get('_provenance') + if not provenance: + # If no provenance (legacy data), we might fallback or deny. + # For strict security: Deny. For compatibility: Allow if Public. + # Let's check if 'namespace' is at top level + namespace = envelope.get('namespace', 'public') + # Fallback logic + return cls._legacy_check(user, namespace) + + scope = provenance.get('scope', {}) + target_domain = scope.get('domain') or scope.get('namespace') + + # 1. Domain Check + allowed_reads = cls.ROLE_DOMAINS[user.role]["read"] + if "*" not in allowed_reads and target_domain not in allowed_reads: + # Silent Redaction (no audit log for simple filter to avoid spam) + return False + + # 2. Clearance Check (Implicit in domain for this prototype) + # In a full system, envelope would carry a specific classification tag + # Here we map domain -> clearance + req_clearance = cls._get_domain_clearance(target_domain) + + if req_clearance > user.clearance: + # AUDIT: Access Denied (Clearance) + ledger.log(AuditEvent( + action_type="ACCESS_DENIED", + actor_id=user.user_id, + role=user.role, + target=target_domain, + details={"reason": "insufficient_clearance", "required": req_clearance.name} + )) + return False + + return True + + @classmethod + def _legacy_check(cls, user: UserContext, namespace: str) -> bool: + allowed = cls.ROLE_DOMAINS[user.role]["read"] + if "*" in allowed: return True + return namespace in allowed + + @classmethod + def _get_domain_clearance(cls, domain: str) -> Grade: + if "public" in domain: return Grade.E_CLASS + if "team" in domain: return Grade.C_CLASS + if "tech" in domain: return Grade.B_CLASS + if "research" in domain: return Grade.A_CLASS + if "secret" in domain: return Grade.S_CLASS + return Grade.C_CLASS diff --git a/memory_thread/nervous/audit_ledger.py b/memory_thread/nervous/audit_ledger.py new file mode 100644 index 0000000..07962e3 --- /dev/null +++ b/memory_thread/nervous/audit_ledger.py @@ -0,0 +1,93 @@ +import json +import logging +from datetime import datetime +from typing import Dict, Any, Optional +from dataclasses import dataclass, field +import uuid +import os + +# We treat the audit ledger as a separate system component +# In a real enterprise setup, this would write to a WORM (Write Once Read Many) storage +# For this implementation, we will use a dedicated JSONL file or a separate DB table if available. +# To allow portability without complex setup, we default to a file-based ledger in ~/.mt/audit/ + +DEFAULT_AUDIT_PATH = os.path.expanduser("~/.mt/audit_ledger.jsonl") + +@dataclass +class AuditEvent: + id: str = field(default_factory=lambda: str(uuid.uuid4())) + timestamp: str = field(default_factory=lambda: datetime.utcnow().isoformat()) + action_type: str = "GENERIC" # ACCESS_DENIED, OVERRIDE, REDACTION, PRUNING + actor_id: str = "system" + role: str = "system" + target: str = "" + details: Dict[str, Any] = field(default_factory=dict) + + def to_json(self) -> str: + return json.dumps({ + "id": self.id, + "timestamp": self.timestamp, + "type": self.action_type, + "actor": { + "id": self.actor_id, + "role": self.role + }, + "target": self.target, + "details": self.details + }) + +class AuditLedger: + def __init__(self, file_path: str = DEFAULT_AUDIT_PATH): + self.file_path = file_path + self._ensure_dir() + + def _ensure_dir(self): + directory = os.path.dirname(self.file_path) + if not os.path.exists(directory): + try: + os.makedirs(directory, exist_ok=True) + except OSError: + # Fallback to local dir if permission denied + self.file_path = "audit_ledger.jsonl" + + def log(self, event: AuditEvent): + """Append an event to the ledger.""" + try: + with open(self.file_path, 'a', encoding='utf-8') as f: + f.write(event.to_json() + "\n") + except Exception as e: + # Fallback logging if file write fails - Audit must never fail silently + logging.critical(f"AUDIT WRITE FAILED: {event.to_json()} - Error: {e}") + + def query(self, actor_id: Optional[str] = None, limit: int = 100) -> list: + """ + Query audit logs (Reverse chronological). + Protected method - should only be exposed to Root/Superuser. + """ + results = [] + try: + if not os.path.exists(self.file_path): + return [] + + # Read from end (efficient for tail) would be better, but for now scan whole + with open(self.file_path, 'r', encoding='utf-8') as f: + lines = f.readlines() + + for line in reversed(lines): + if len(results) >= limit: + break + try: + data = json.loads(line) + if actor_id: + if data['actor']['id'] != actor_id: + continue + results.append(data) + except: + continue + except Exception: + return [] + + return results + +# Singleton instance +ledger = AuditLedger() diff --git a/memory_thread/nervous/authority_store.py b/memory_thread/nervous/authority_store.py new file mode 100644 index 0000000..bca6aee --- /dev/null +++ b/memory_thread/nervous/authority_store.py @@ -0,0 +1,116 @@ + +import json +import os +import uuid +from datetime import datetime +from typing import List, Optional, Dict +from dataclasses import dataclass, field + +DEFAULT_AUTH_STORE = os.path.expanduser("~/.mt/authority_grants.jsonl") + +@dataclass +class AuthorityGrant: + id: str = field(default_factory=lambda: str(uuid.uuid4())) + timestamp: str = field(default_factory=lambda: datetime.utcnow().isoformat()) + granter_id: str = "" + granter_role: str = "" + target_role: str = "" # Role being granted power + target_domain: str = "" # Domain scope + score: float = 0.5 + active: bool = True + + def to_json(self) -> str: + return json.dumps({ + "id": self.id, + "timestamp": self.timestamp, + "granter": {"id": self.granter_id, "role": self.granter_role}, + "target_role": self.target_role, + "domain": self.target_domain, + "score": self.score, + "active": self.active + }) + +class AuthorityStore: + """ + Persistent store for dynamic authority grants. + """ + def __init__(self, file_path: str = DEFAULT_AUTH_STORE): + self.file_path = file_path + self._ensure_dir() + self._cache: List[Dict] = [] + self.reload() + + def _ensure_dir(self): + directory = os.path.dirname(self.file_path) + if not os.path.exists(directory): + try: + os.makedirs(directory, exist_ok=True) + except OSError: + self.file_path = "authority_grants.jsonl" + + def reload(self): + """Load grants into memory cache.""" + self._cache = [] + if not os.path.exists(self.file_path): + return + + try: + with open(self.file_path, 'r', encoding='utf-8') as f: + for line in f: + try: + self._cache.append(json.loads(line)) + except: continue + except Exception: + pass + + def add_grant(self, grant: AuthorityGrant): + """Persist a new grant.""" + try: + with open(self.file_path, 'a', encoding='utf-8') as f: + f.write(grant.to_json() + "\n") + self._cache.append(json.loads(grant.to_json())) + except Exception as e: + print(f"Failed to save grant: {e}") + + def revoke(self, target_role: str, domain: str, revoker_role: str): + """ + Soft-revoke a grant (mark inactive). + Requires appending a new 'revocation' record effectively. + In this append-only log, we treat a new entry with active=False as revocation. + """ + # We don't overwrite the file (append-only ledger principle). + # We append a record that effectively cancels previous ones. + grant = AuthorityGrant( + granter_role=revoker_role, + granter_id="revocation", + target_role=target_role, + target_domain=domain, + score=0.0, + active=False + ) + self.add_grant(grant) + + def get_score(self, role: str, domain: str) -> Optional[float]: + """ + Get the *latest* active grant score for a Role+Domain. + """ + # Scan from newest to oldest + for entry in reversed(self._cache): + if entry['target_role'] == role: + # Check Domain match (exact or wildcard) + entry_domain = entry['domain'] + if entry_domain == "*" or entry_domain == domain: + if entry['active']: + return entry['score'] + else: + # If latest entry is inactive/revoked, stop and return None (fallback to matrix) + # Or return 0.0? + # Design decision: Revocation means "remove dynamic grant", + # falling back to hardcoded matrix? + # Or explicit 0.0 override? + # Let's say explicit 0.0 override to allow 'blocking'. + return 0.0 + return None + +# Singleton +authority_store = AuthorityStore() diff --git a/memory_thread/nervous/vault.py b/memory_thread/nervous/vault.py new file mode 100644 index 0000000..a5e48ad --- /dev/null +++ b/memory_thread/nervous/vault.py @@ -0,0 +1,79 @@ +import os +import hashlib +import uuid +import json +from typing import Optional, Tuple + +VAULT_PATH = os.path.expanduser("~/.mt/vault.json") + +class Vault: + """ + Secure Credential Store. + Manages PINs and the Nuclear Key for Godfather access. + """ + def __init__(self): + self._ensure_storage() + self._cache = self._load() + + def _ensure_storage(self): + directory = os.path.dirname(VAULT_PATH) + if not os.path.exists(directory): + try: + os.makedirs(directory, exist_ok=True) + except OSError: + pass + + def _load(self) -> dict: + if os.path.exists(VAULT_PATH): + try: + with open(VAULT_PATH, 'r', encoding='utf-8') as f: + return json.load(f) + except: + pass + return {} + + def _save(self): + try: + with open(VAULT_PATH, 'w', encoding='utf-8') as f: + json.dump(self._cache, f, indent=2) + except: + pass + + def _hash(self, secret: str) -> str: + return hashlib.sha256(secret.encode()).hexdigest() + + def get_or_create_godfather_key(self) -> str: + """ + Generates the Nuclear Key if missing. + Returns the PLAINTEXT key (only once/on request) for display. + """ + if "godfather_hash" in self._cache: + return "[HIDDEN - ALREADY SET]" + + # Generate new key + key = f"MT-{uuid.uuid4().hex[:12].upper()}" + self._cache["godfather_hash"] = self._hash(key) + self._save() + return key + + def verify_godfather(self, key_input: str) -> bool: + """Checks against the Nuclear Key.""" + stored = self._cache.get("godfather_hash") + if not stored: return False + return self._hash(key_input) == stored + + def set_pin(self, username: str, pin: str): + """Sets a simple PIN for a user.""" + self._cache[f"pin_{username}"] = self._hash(pin) + self._save() + + def verify_pin(self, username: str, pin_input: str) -> bool: + """Verifies user PIN.""" + stored = self._cache.get(f"pin_{username}") + if not stored: + # Default PIN for demo if not set: '0000' + return pin_input == "0000" + return self._hash(pin_input) == stored + +# Singleton +vault = Vault() diff --git a/memory_thread/utils/cli_bridge.py b/memory_thread/utils/cli_bridge.py new file mode 100644 index 0000000..ecd92f0 --- /dev/null +++ b/memory_thread/utils/cli_bridge.py @@ -0,0 +1,480 @@ +""" +MT Neural Interface - Professional Galaxy TUI +""" + +from textual.app import App, ComposeResult +from textual.widgets import ( + Header, Footer, Static, Input, ListView, ListItem, + Label, Tree, DataTable, Log, TabbedContent, TabPane +) +from textual.containers import Container, Horizontal, Vertical, ScrollableContainer +from textual.binding import Binding +from textual.reactive import reactive +from textual import events +import asyncio +from datetime import datetime +from typing import List, Dict, Optional +import uuid + +# MT imports +from memory_thread.sdk import MemoryClient +from memory_thread.core.galaxy import GalaxyCore # Your new core + +# ============================================================================ +# DATA MODELS +# ============================================================================ + +class AgentUniverse: + def __init__(self, agent_id: str, active: bool = False): + self.agent_id = agent_id + self.active = active + self.fact_count = 0 + self.belief_count = 0 + self.activity_pct = 0.0 + +class Conflict: + def __init__(self, fact_id: str, agents: List[str], severity: str): + self.fact_id = fact_id + self.agents = agents + self.severity = severity + self.timestamp = datetime.now() + +class FactEntry: + def __init__(self, fact_id: str, source: str, preview: str): + self.fact_id = fact_id + self.source = source + self.preview = preview + self.timestamp = datetime.now() + +# ============================================================================ +# CUSTOM WIDGETS +# ============================================================================ + +class AgentUniversePanel(Static): + """Shows active agent universes""" + + def __init__(self): + super().__init__() + self.universes: List[AgentUniverse] = [] + + def compose(self) -> ComposeResult: + yield Static("AGENT UNIVERSES", classes="panel-title") + yield ListView(id="universe-list") + + def update_universes(self, universes: List[AgentUniverse]): + self.universes = universes + list_view = self.query_one("#universe-list", ListView) + list_view.clear() + + for u in universes: + indicator = "●" if u.active else "○" + line = f"{indicator} {u.agent_id:15} [{u.activity_pct:>3.0f}%]" + item = ListItem(Label(line)) + list_view.append(item) + +class ConflictPanel(Static): + """Shows active conflicts detected by Galaxy""" + + def compose(self) -> ComposeResult: + yield Static("ACTIVE CONFLICTS", classes="panel-title") + yield ListView(id="conflict-list") + + def update_conflicts(self, conflicts: List[Conflict]): + list_view = self.query_one("#conflict-list", ListView) + list_view.clear() + + if not conflicts: + list_view.append(ListItem(Label("[dim]No conflicts detected[/]"))) + return + + for c in conflicts: + agents_str = " vs ".join(c.agents) + severity_color = { + "LOW": "green", + "MEDIUM": "yellow", + "HIGH": "red" + }.get(c.severity, "white") + + item = ListItem(Label( + f"[bold]!{/] {c.fact_id[:8]}\n" + f" {agents_str}\n" + f" [{severity_color}]{c.severity}[/]" + )) + list_view.append(item) + +class FactStreamPanel(Static): + """Shows recent facts ingested into galaxy""" + + def compose(self) -> ComposeResult: + yield Static("RECENT FACTS", classes="panel-title") + yield DataTable(id="fact-table") + + def on_mount(self): + table = self.query_one("#fact-table", DataTable) + table.add_columns("ID", "Source", "Preview", "Age") + table.zebra_stripes = True + + def update_facts(self, facts: List[FactEntry]): + table = self.query_one("#fact-table", DataTable) + table.clear() + + for f in facts: + age = self._format_age(f.timestamp) + table.add_row( + f.fact_id[:8], + f.source[:15], + f.preview[:30] + "...", + age + ) + + def _format_age(self, timestamp: datetime) -> str: + delta = datetime.now() - timestamp + if delta.seconds < 60: + return f"{delta.seconds}s ago" + elif delta.seconds < 3600: + return f"{delta.seconds // 60}m ago" + else: + return f"{delta.seconds // 3600}h ago" + +class ChatPanel(ScrollableContainer): + """Main chat interface with conflict notifications""" + + def compose(self) -> ComposeResult: + yield Log(id="chat-log", auto_scroll=True) + yield Input(placeholder="▌ Type your message...", id="chat-input") + + def add_message(self, role: str, content: str, conflict: bool = False): + log = self.query_one("#chat-log", Log) + + color = { + "user": "cyan", + "assistant": "white", + "system": "yellow" + }.get(role, "white") + + log.write_line(f"[{color}]{role.upper()}:[/] {content}") + + if conflict: + log.write_line("[red]⚠ Conflict detected - press 'c' to resolve[/]") + +# ============================================================================ +# MAIN APP +# ============================================================================ + +class MTNeuralInterface(App): + """Memory Thread Neural Interface""" + + CSS = """ + Screen { + background: $surface; + } + + .panel-title { + background: $primary; + color: $text; + padding: 0 1; + text-style: bold; + } + + #universe-list, #conflict-list { + height: 10; + border: solid $primary; + } + + #fact-table { + height: 6; + border: solid $primary; + } + + #chat-log { + height: 1fr; + border: solid $accent; + margin: 1 0; + } + + #chat-input { + border: solid $accent; + } + + Input { + background: $surface; + } + """ + + BINDINGS = [ + Binding("q", "quit", "Quit"), + Binding("u", "show_universes", "Universes"), + Binding("c", "show_conflicts", "Conflicts"), + Binding("f", "show_facts", "Facts"), + Binding("slash", "search", "Search"), + Binding("question_mark", "help", "Help"), + Binding("s", "toggle_secure", "Security"), + Binding("g", "toggle_galaxy", "Galaxy"), + ] + + TITLE = "MT NEURAL INTERFACE v2.0" + + # Reactive properties + secure_mode = reactive(False) + galaxy_active = reactive(True) + active_agent = reactive("SecurityBot") + + def __init__(self): + super().__init__() + self.galaxy: Optional[GalaxyCore] = None + self.universes: List[AgentUniverse] = [] + self.conflicts: List[Conflict] = [] + self.facts: List[FactEntry] = [] + + def compose(self) -> ComposeResult: + yield Header(show_clock=True) + + with Horizontal(): + # Left column + with Vertical(classes="left-panel"): + yield AgentUniversePanel() + yield ConflictPanel() + + # Right column (main) + with Vertical(classes="main-panel"): + yield FactStreamPanel() + yield ChatPanel() + + yield Footer() + + def on_mount(self): + """Initialize Galaxy core and start monitoring""" + self.initialize_galaxy() + self.set_interval(2.0, self.update_status) + + def initialize_galaxy(self): + """Initialize the Galaxy architecture""" + # This will use your actual GalaxyCore + from memory_thread.core.galaxy import GalaxyCore + from memory_thread.services.persistence import PersistenceEngine + + # Initialize core components + pg_client = None # Your Postgres client + qdrant_client = None # Your Qdrant client + + self.galaxy = GalaxyCore(pg_client, qdrant_client) + + # Register default agents + self.galaxy.register_agent("SecurityBot", authority=0.9) + self.galaxy.register_agent("MarketingBot", authority=0.5) + self.galaxy.register_agent("AuditBot", authority=0.8) + + # Initialize UI state + self.universes = [ + AgentUniverse("SecurityBot", active=True), + AgentUniverse("MarketingBot"), + AgentUniverse("AuditBot"), + ] + + async def update_status(self): + """Periodic update of UI state from Galaxy""" + if not self.galaxy: + return + + # Update agent activity + for universe in self.universes: + # Query Galaxy for agent activity + facts = await self.galaxy.get_agent_facts(universe.agent_id) + universe.fact_count = len(facts) + universe.activity_pct = (universe.fact_count / 100.0) * 100 # Mock + + # Update conflicts + conflicts = await self.galaxy.get_active_conflicts() + self.conflicts = [ + Conflict( + fact_id=c.fact_id, + agents=[b.agent_id for b in c.beliefs], + severity=self._compute_severity(c) + ) + for c in conflicts + ] + + # Update recent facts + recent = await self.galaxy.get_recent_facts(limit=10) + self.facts = [ + FactEntry( + fact_id=str(f.id), + source=f.source_uri, + preview=f.content[:50] + ) + for f in recent + ] + + # Refresh UI + self.refresh_panels() + + def _compute_severity(self, conflict) -> str: + """Compute conflict severity based on authority divergence""" + authorities = [b.agent_authority for b in conflict.beliefs] + if not authorities: + return "LOW" + + max_auth = max(authorities) + min_auth = min(authorities) + divergence = max_auth - min_auth + + if divergence > 0.5: + return "HIGH" + elif divergence > 0.3: + return "MEDIUM" + else: + return "LOW" + + def refresh_panels(self): + """Refresh all UI panels with current data""" + universe_panel = self.query_one(AgentUniversePanel) + universe_panel.update_universes(self.universes) + + conflict_panel = self.query_one(ConflictPanel) + conflict_panel.update_conflicts(self.conflicts) + + fact_panel = self.query_one(FactStreamPanel) + fact_panel.update_facts(self.facts) + + async def on_input_submitted(self, event: Input.Submitted): + """Handle chat input""" + chat_panel = self.query_one(ChatPanel) + user_input = event.value + + if not user_input.strip(): + return + + # Clear input + event.input.value = "" + + # Show user message + chat_panel.add_message("user", user_input) + + # Process through Galaxy + if self.galaxy: + # Ingest as fact + fact, belief = await self.galaxy.ingest( + agent_id=self.active_agent, + raw_observation={"text": user_input, "source": "user:input"} + ) + + # Generate response (mock - integrate with your LLM) + response = await self.generate_response(user_input, belief) + + # Check for conflicts + has_conflict = len(self.conflicts) > 0 + + # Show response + chat_panel.add_message("assistant", response, conflict=has_conflict) + + async def generate_response(self, user_input: str, belief) -> str: + """Generate response using LLM (integrate with your chat logic)""" + # This should call your actual LLM integration + return f"Processing: {user_input}" + + # ======================================================================== + # ACTIONS (Key Bindings) + # ======================================================================== + + def action_show_universes(self): + """Show detailed universe view""" + self.push_screen(UniverseDetailScreen(self.universes)) + + def action_show_conflicts(self): + """Show conflict resolution screen""" + if self.conflicts: + self.push_screen(ConflictResolutionScreen(self.conflicts[0])) + else: + self.notify("No active conflicts") + + def action_show_facts(self): + """Show fact browser""" + self.push_screen(FactBrowserScreen(self.facts)) + + def action_search(self): + """Open search interface""" + self.notify("Search not implemented yet") + + def action_help(self): + """Show help screen""" + self.push_screen(HelpScreen()) + + def action_toggle_secure(self): + """Toggle secure mode""" + self.secure_mode = not self.secure_mode + status = "ENABLED" if self.secure_mode else "DISABLED" + self.notify(f"Secure Mode: {status}") + + def action_toggle_galaxy(self): + """Toggle galaxy architecture""" + self.galaxy_active = not self.galaxy_active + status = "ACTIVE" if self.galaxy_active else "INACTIVE" + self.notify(f"Galaxy: {status}") + + def watch_secure_mode(self, secure: bool): + """Update header when secure mode changes""" + self.sub_title = "[SECURE]" if secure else "" + + def watch_galaxy_active(self, active: bool): + """Update header when galaxy toggles""" + status = "GALAXY ON" if active else "GALAXY OFF" + self.sub_title = f"{self.sub_title} {status}".strip() + +# ============================================================================ +# DETAIL SCREENS +# ============================================================================ + +class UniverseDetailScreen(Screen): + """Detailed view of agent universes""" + pass + +class ConflictResolutionScreen(Screen): + """Interactive conflict resolution""" + pass + +class FactBrowserScreen(Screen): + """Browse and search facts""" + pass + +class HelpScreen(Screen): + """Help and keybindings""" + + BINDINGS = [("escape", "app.pop_screen", "Close")] + + def compose(self) -> ComposeResult: + yield Static(""" +# MT NEURAL INTERFACE - HELP + +## Keybindings + +q - Quit application +u - Show universe details +c - Resolve conflicts +f - Browse facts +/ - Search +? - This help screen +s - Toggle secure mode +g - Toggle galaxy architecture + +## Concepts + +AGENT UNIVERSES +Each agent maintains their own fact space and beliefs. +Galaxy architecture automatically links related beliefs. + +CONFLICTS +When agents disagree about the same fact, conflicts are detected. +Press 'c' to resolve using authority, consensus, or manual selection. + +FACTS vs BELIEFS +Facts are immutable observations. +Beliefs are agent interpretations of facts. + """, id="help-text") + +# ============================================================================ +# ENTRY POINT +# ============================================================================ + +if __name__ == "__main__": + app = MTNeuralInterface() + app.run() diff --git a/memory_thread/utils/secure_sdk.py b/memory_thread/utils/secure_sdk.py new file mode 100644 index 0000000..e251a79 --- /dev/null +++ b/memory_thread/utils/secure_sdk.py @@ -0,0 +1,336 @@ + +""" +Secure Memory Client Wrapper. + +This wrapper injects the Enterprise Firewall layer (AccessControlService) +transparently around the core SDK MemoryClient. + +UPDATED: Enforces Provenance Envelope on every write. +""" +from typing import Optional, List, Dict, Any, Union +import uuid +import json +import datetime + +from memory_thread.sdk import MemoryClient, RecallResult, Memory +from memory_thread.nervous.access_control import AccessControlService, UserContext +from memory_thread.models.provenance import ProvenanceEnvelope, Actor, Origin, Scope +from memory_thread.nervous.audit_ledger import ledger, AuditEvent + +class SecureMemoryClient: + """ + Enterprise-grade wrapper for MemoryClient. + Enforces RBAC, Clearance, Truth Authority, and Provenance. + """ + + def __init__(self, user_id: str, role: str, client_id: str = "tui-client"): + self.user = AccessControlService.create_context(user_id, role) + self.origin = Origin(client_id=client_id, session_id=str(uuid.uuid4())) + self._core_client = MemoryClient(namespace="default", use_db=True) + + @property + def role(self): + return self.user.role + + @property + def clearance(self): + return self.user.grade.name + + def ingest_fact(self, content: str, source_uri: str = "manual", namespace: str = "public") -> Optional[uuid.UUID]: + """ + Class A Ingestion: Canonical Truth. + - Must be raw content (no embeddings, no opinions). + - Must be verifiable. + """ + # DEBUG CHECK + # print(f"DEBUG: ingest_fact called with {content}") + # Prime Rule Checks + if not content or not isinstance(content, str): + raise ValueError("PRIME RULE VIOLATION: Fact content must be a non-empty string.") + if len(content) > 100000: + # Just a sanity check, large files are okay but memory limits exist + pass + + # Check for Forbidden Patterns (Heuristic) + if content.strip().startswith("[") and content.strip().endswith("]") and "," in content: + # Rough check for vector/embedding dump + # If it looks like a list of floats, reject. + try: + possible_vec = json.loads(content) + if isinstance(possible_vec, list) and len(possible_vec) > 0 and isinstance(possible_vec[0], (float, int)): + raise ValueError("PRIME RULE VIOLATION: Embeddings cannot be stored as Truth.") + except json.JSONDecodeError: + pass + except ValueError as e: + raise e # Re-raise our own violation + except Exception: + pass + + return self._internal_remember( + content=content, + namespace=namespace, + memory_type="fact", + confidence=1.0, # Facts are absolute + source_uri=source_uri, + provenance_extras={} + ) + + def record_belief(self, content: str, derived_from: List[uuid.UUID], confidence: float, namespace: str = "public") -> Optional[uuid.UUID]: + """ + Class B Ingestion: Epistemic Artifact. + - Must have provenance (derived_from). + - Must have confidence. + """ + # Prime Rule Checks + if not derived_from or not isinstance(derived_from, list): + raise ValueError("PRIME RULE VIOLATION: Beliefs must have explicit 'derived_from' provenance.") + + if confidence is None or not (0.0 <= confidence <= 1.0): + raise ValueError("PRIME RULE VIOLATION: Beliefs must have a valid confidence score (0.0-1.0).") + + return self._internal_remember( + content=content, + namespace=namespace, + memory_type="belief", + confidence=confidence, + source_uri=f"agent:{self.user.role}", + provenance_extras={"derived_from": [str(uid) for uid in derived_from]} + ) + + def remember(self, content: str, namespace: str = "public", + memory_type: str = "fact", **kwargs) -> Optional[uuid.UUID]: + """ + [DEPRECATED] Generic wrapper. + Routes to specific methods or warns. + """ + print(f"WARNING: 'remember()' is deprecated. Use 'ingest_fact' or 'record_belief'.") + + if memory_type == "fact": + return self.ingest_fact(content, namespace=namespace) + elif memory_type == "belief": + derived = kwargs.get('derived_from', []) + conf = kwargs.get('confidence', 0.5) + if not derived: + # Soft violation for backward compat during migration? + # NO. Prime Rule is law. + raise ValueError("PRIME RULE VIOLATION: Cannot store belief without 'derived_from' via generic remember().") + return self.record_belief(content, derived, conf, namespace) + else: + # Default to Fact if ambiguous but warn? + # Safer to fail. + raise ValueError(f"Unknown memory_type: {memory_type}") + + def _internal_remember(self, content: str, namespace: str, memory_type: str, + confidence: float, source_uri: str, provenance_extras: Dict) -> Optional[uuid.UUID]: + """ + Internal Secure Persist Logic. + """ + # 1. Check Write Permissions & Get Authority + authority_score = AccessControlService.calculate_write_authority(self.user, namespace) + + if authority_score == 0.0: + return None # Audit log handled in AccessControlService + + # 2. Construct Provenance Envelope + envelope = ProvenanceEnvelope( + actor=Actor(user_id=self.user.user_id, role=self.user.role), + origin=self.origin, + scope=Scope(namespace=namespace, domain=namespace) # Domain mapped to namespace for now + ) + + # Merge extras (like derived_from) + env_dict = envelope.to_dict() + env_dict.update(provenance_extras) + + # 3. Payload Injection + secure_payload = { + "text": content, + "_provenance": env_dict + } + + serialized_content = json.dumps(secure_payload) + + # 4. Call Core + event_id = self._core_client.remember( + content=serialized_content, + source=source_uri, + confidence=confidence, + authority=authority_score, + memory_type=memory_type + ) + + return event_id + + def recall(self, query: str, top_k: int = 5, target_namespaces: List[str] = None) -> RecallResult: + """ + Secure Recall with Firewall Filtering. + """ + if target_namespaces is None: + # Default to all namespaces this user can read + target_namespaces = self.user.domains + + all_memories = [] + + # Calculate per-namespace limit to ensure we get enough candidates + # We request top_k from each namespace to be safe + + for ns in target_namespaces: + # Firewall Check: Can user read this namespace? + # We create a dummy envelope for this high-level check + dummy_env = {'_provenance': {'scope': {'namespace': ns}}} + if not AccessControlService.can_read(self.user, dummy_env): + continue + + self._core_client.namespace = ns + result = self._core_client.recall(query, top_k=top_k) + + for mem in result.memories: + # 1. Parse Provenance + try: + payload = json.loads(mem.content) + if isinstance(payload, dict) and "_provenance" in payload: + # It's a secured memory + provenance = payload["_provenance"] + actual_text = payload["text"] + else: + # Legacy/Plain memory + provenance = None + actual_text = mem.content + except json.JSONDecodeError: + provenance = None + actual_text = mem.content + + # 2. Construct Envelope for Firewall + # If legacy, we assume the namespace of the query implies origin + check_env = {"_provenance": provenance} if provenance else {"namespace": ns} + + # 3. Firewall Check + if AccessControlService.can_read(self.user, check_env): + # Unpack content for the user + mem.content = actual_text + + # Tag source + if provenance: + actor = provenance.get('actor', {}) + mem.source = f"{actor.get('role', 'unknown')} (Auth: {mem.authority:.2f})" + else: + mem.source = f"{ns} (Legacy)" + + all_memories.append(mem) + else: + # Filtered out + pass + + # 2. Re-rank + all_memories.sort(key=lambda m: m.truth_score, reverse=True) + + return RecallResult( + memories=all_memories[:top_k], + query=query, + total_found=len(all_memories) + ) + + def get_related(self, entity_id: uuid.UUID, depth: int = 1) -> List[Dict]: + """ + Delegated Graph Query. + In a real secure system, we would filter these relations too. + For this prototype, we allow structure exploration but redact details if needed. + """ + # We need to access across all namespaces, or default to current user context? + # Graph service is global usually. + # We'll delegate to core client. + return self._core_client.get_related(entity_id, depth) + + def __getattr__(self, name): + """Delegate unknown methods to core client (e.g. get_stats, get_health).""" + return getattr(self._core_client, name) + + def chat(self, user_message: str, system_prompt: Optional[str] = None, use_local: bool = True, smart_loop: bool = False) -> str: + """ + Secure Chat with optional Smart Loop (Layer VI). + """ + # 1. Secure Recall (Initial Pass) + recall_res = self.recall(user_message, top_k=5) + + # Smart Loop: Reflection (Layer VI) + if smart_loop and recall_res.total_found < 2: + # If low context, ask LLM what else it needs + reflection_prompt = f"""User: {user_message} +Current Context: {recall_res.to_context(max_chars=500)} +Task: Identify one specific search query to find missing info. Return ONLY the query.""" + + if use_local: + next_query = self._core_client._generate_local(reflection_prompt) + else: + next_query = self._core_client._generate_cloud(reflection_prompt) + + # Clean up query + next_query = next_query.strip().replace('"', '') + + # Secondary Recall + extra_res = self.recall(next_query, top_k=3) + # Merge results (simple append for prototype) + recall_res.memories.extend(extra_res.memories) + + # 2. Build Context + context_str = recall_res.to_context(max_chars=3000) + + # 3. Construct Prompt + full_prompt = f"""{system_prompt or 'You are a helpful assistant.'} + +SECURITY CONTEXT: +User Role: {self.user.role} +Grade: {self.user.grade.name} + +SECURE MEMORY CONTEXT (Only authorized facts): +{context_str} + +User: {user_message} +Assistant:""" + + # 4. Core Generation + if use_local: + return self._core_client._generate_local(full_prompt) + else: + return self._core_client._generate_cloud(full_prompt) + + def _get_namespace_clearance(self, namespace: str) -> int: + """Helper to map namespace to required Grade.""" + from memory_thread.nervous.access_control import Grade + if "public" in namespace: return Grade.E_CLASS + if "team" in namespace: return Grade.C_CLASS + if "tech" in namespace: return Grade.B_CLASS + if "research" in namespace: return Grade.A_CLASS + if "secret" in namespace: return Grade.S_CLASS + return Grade.C_CLASS + + # --- ADMIN CAPABILITY --- + def audit_log(self, limit: int = 50) -> List[Dict]: + """ + Root capability to view audit logs. + """ + if self.user.role != "root": + ledger.log(AuditEvent( + action_type="ACCESS_DENIED", + actor_id=self.user.user_id, + role=self.user.role, + target="audit_log", + details={"reason": "requires_root"} + )) + return [] + + return ledger.query(limit=limit) + + def clear(self): + if self.user.role == "root": + self._core_client.clear() + else: + print(f"Access Denied: Only ROOT can clear DB.") + + def grant(self, target_role: str, domain: str, score: float) -> bool: + """Dynamic Authority Grant.""" + return AccessControlService.grant_authority(self.user, target_role, domain, score) + + def revoke(self, target_role: str, domain: str) -> bool: + """Dynamic Revocation.""" + return AccessControlService.revoke_authority(self.user, target_role, domain)