Establish conscious heart core with local-first policy

README.md

@@ -16,6 +16,14 @@ Memory decay for outdated information
 
 Recall, filtering, and summarization
 
+**Phase 1 Conscious Heart**
+
+- `blux_ca.core.heart.ConsciousHeart` orchestrates perception, discernment, and doctrine-backed verdicts.
+- Intent compass classifies along truth, integrity, compassion, and awareness axes.
+- Koan probes surface reflective prompts for orchestrated `bq-cli` sessions.
+- Consent-based memory honors privacy-first defaults and only persists with explicit approval.
+- Immutable audit log supports append-only playback with hash chaining for local-first compliance.
+
 Multi-Agent Collaboration
 
 Broadcast tasks to multiple agents

blux/adaptors/__init__.py

@@ -1,3 +1,9 @@
 # Adaptors package
 from .dummy_local import DummyLocalAdaptor
-from .http_api_adaptor import HTTPAPIAdaptor
+
+try:  # Optional HTTP adaptor dependency
+    from .http_api_adaptor import HTTPAPIAdaptor
+except ModuleNotFoundError:  # pragma: no cover - optional flask integration
+    HTTPAPIAdaptor = None  # type: ignore
+
+__all__ = ["DummyLocalAdaptor", "HTTPAPIAdaptor"]

blux/agent/audit.py

@@ -1,16 +1,24 @@
+from __future__ import annotations
+
+"""Simple in-memory audit trail."""
+
+from typing import Dict, List
+
+
 class AuditTrail:
-    """
-    Records agent decisions and actions.
-    """
-    def __init__(self):
-        self.logs = []
+    """Records agent decisions and actions in memory."""
+
+    def __init__(self) -> None:
+        self.logs: List[Dict[str, str]] = []
 
-    def log(self, user_input, user_type, decision):
-        self.logs.append({
-            "input": user_input,
-            "user_type": user_type,
-            "decision": decision
-        })
+    def log(self, user_input: str, user_type: str, decision: str) -> None:
+        self.logs.append(
+            {
+                "input": user_input,
+                "user_type": user_type,
+                "decision": decision,
+            }
+        )
 
-    def get_logs(self):
-        return self.logs
+    def get_logs(self) -> List[Dict[str, str]]:
+        return list(self.logs)

blux/agent/constitution.py

@@ -1,17 +1,21 @@
+from __future__ import annotations
+
+"""Minimal constitution facade for the basic BLUX agent."""
+
+
 class Constitution:
-    """
-    Implements core rules and spine of BLUX-cA.
-    """
-    def __init__(self):
+    """Applies constitution rules based on the classified user type."""
+
+    def __init__(self) -> None:
         self.rules = [
             "truth_over_comfort",
             "integrity_over_approval",
-            "audit_required_for_decisions"
+            "audit_required_for_decisions",
         ]
 
-    def apply_rules(self, user_input, user_type):
+    def apply_rules(self, user_input: str, user_type: str) -> str:
+        """Return the decision aligned with the active constitution."""
+
         if user_type == "struggler":
             return "validate and provide guidance"
-        return "set boundaries / off-ramp"provide guidance"
-        else:
-            return "set boundaries / off-ramp"
+        return "set boundaries / off-ramp"

blux/agent/core_agent.py

@@ -1,32 +1,28 @@
-from .memory import Memory
-from .discernment import DiscernmentCompass
-from .constitution import Constitution
+from __future__ import annotations
+
+"""Legacy BLUX agent wiring used by unit tests."""
+
 from .audit import AuditTrail
+from .constitution import Constitution
+from .discernment import DiscernmentCompass
+from .memory import Memory
+
 
 class BLUXAgent:
-    """
-    Core BLUX-cA agent implementing:
-    - Constitutional rules enforcement
-    - Discernment compass
-    - Memory handling
-    - Auditing
-    """
-    def __init__(self, name="BLUX-cA"):
+    """Minimal agent coordinating constitution, discernment, memory, and audit."""
+
+    def __init__(self, name: str = "BLUX-cA") -> None:
         self.name = name
         self.memory = Memory()
         self.constitution = Constitution()
         self.discernment = DiscernmentCompass()
         self.audit = AuditTrail()
 
-    def process_input(self, user_input):
+    def process_input(self, user_input: str) -> str:
+        """Run the full decision flow for ``user_input``."""
+
         user_type = self.discernment.classify(user_input)
         decision = self.constitution.apply_rules(user_input, user_type)
         self.memory.store(user_input, user_type, decision)
         self.audit.log(user_input, user_type, decision)
-        return f"[{user_type}] Decision: {decision}"pe, decision)
-
-        # 4. Record audit
-        self.audit.log(user_input, user_type, decision)
-
-        # 5. Generate output (placeholder)
-        return f"[{user_type}] Decision: {decision}"
+        return f"[{user_type}] Decision: {decision}"

blux/agent/discernment.py

@@ -1,12 +1,17 @@
+from __future__ import annotations
+
+"""Simple discernment compass for legacy BLUX agent tests."""
+
+
 class DiscernmentCompass:
-    """
-    Determines user type:
-    - Struggler: validate and offer tools
-    - Indulgent: boundary & off-ramp
-    """
-    def classify(self, user_input):
-        if any(word in user_input.lower() for word in ["help", "struggle", "problem"]):
+    """Classifies user intent into coarse support buckets."""
+
+    _SUPPORT_KEYWORDS = {"help", "struggle", "problem", "support", "lost"}
+
+    def classify(self, user_input: str) -> str:
+        """Return ``"struggler"`` if supportive keywords are present."""
+
+        lowered = user_input.lower()
+        if any(keyword in lowered for keyword in self._SUPPORT_KEYWORDS):
             return "struggler"
-        return "indulgent"       return "struggler"
-        else:
-            return "indulgent"
+        return "indulgent"

blux/agent/memory.py

@@ -1,22 +1,28 @@
+from __future__ import annotations
+
+"""In-memory storage used by legacy BLUX agent tests."""
+
+from typing import Dict, List
+
+
 class Memory:
-    """
-    Handles long-term and session memory storage.
-    """
-    def __init__(self):
-        self.session_memory = []
-        self.long_term_memory = []
-
-    def store(self, user_input, user_type, decision):
+    """Stores session and long-term memory entries."""
+
+    def __init__(self) -> None:
+        self.session_memory: List[Dict[str, str]] = []
+        self.long_term_memory: List[Dict[str, str]] = []
+
+    def store(self, user_input: str, user_type: str, decision: str) -> None:
         entry = {
             "input": user_input,
             "user_type": user_type,
-            "decision": decision
+            "decision": decision,
         }
         self.session_memory.append(entry)
-        self.long_term_memory.append(entry)  # placeholder for consented storage
+        self.long_term_memory.append(entry)
 
-    def recall_session(self):
-        return self.session_memory
+    def recall_session(self) -> List[Dict[str, str]]:
+        return list(self.session_memory)
 
-    def recall_long_term(self):
-        return self.long_term_memorylong_term_memory
+    def recall_long_term(self) -> List[Dict[str, str]]:
+        return list(self.long_term_memory)

blux/orchestrator/controller.py

@@ -1,49 +1,47 @@
-"""Controller: fan-out -> gather -> evaluate -> merge.
+"""Minimal orchestration controller used in legacy tests."""
 
-This is intentionally simple and synchronous for clarity.
-"""
-from typing import List, Dict, Any
-from .registry import ModelRegistry
-from .router import Router
-from .logs import AuditLogger
-from importlib import import_module
+from __future__ import annotations
 
+from dataclasses import dataclass, field
+from typing import Any, Dict
 
-class Controller:
-    def __init__(self, router: Router, logger: AuditLogger):
-        self.router = router
-        self.logger = logger
-
-    def handle_request(self, prompt: str) -> Dict[str, Any]:
-        selected = self.router.select(prompt)
-        responses = []
-        for name in selected:
-            adapter = self.router.registry.get_adapter(name)
-            if not adapter:
-                continue
-            out = adapter.predict(prompt)
-            responses.append({"model": name, "output": out})
-
-        # Evaluate responses
-        from .evaluator.python import PythonEvaluator
-
-        evaluator = PythonEvaluator()
-        scored = []
-        for r in responses:
-            score, metadata = evaluator.score(r["output"], prompt)
-            scored.append({"model": r["model"], "output": r["output"], "score": score, "meta": metadata})
-
-        # Merge policy: choose top score
-        best = sorted(scored, key=lambda x: x["score"], reverse=True)[0] if scored else None
-
-        audit_entry = {
-            "prompt": prompt,
-            "candidates": scored,
-            "selected": best,
+
+@dataclass
+class _RegistryState:
+    agents: Dict[str, Any] = field(default_factory=dict)
+    adaptors: Dict[str, Any] = field(default_factory=dict)
+    evaluators: Dict[str, Any] = field(default_factory=dict)
+
+    def list_all(self) -> Dict[str, list[str]]:
+        return {
+            "agents": list(self.agents.keys()),
+            "adaptors": list(self.adaptors.keys()),
+            "evaluators": list(self.evaluators.keys()),
         }
-        self.logger.log(audit_entry)
 
-        return {"selected": best, "candidates": scored}
+
+class Controller:
+    """Coordinates agents, adaptors, and evaluators for simple task processing."""
+
+    def __init__(self) -> None:
+        self.registry = _RegistryState()
+
+    def register_agent(self, name: str, agent: Any) -> None:
+        self.registry.agents[name] = agent
+
+    def register_adaptor(self, name: str, adaptor: Any) -> None:
+        self.registry.adaptors[name] = adaptor
+
+    def register_evaluator(self, name: str, evaluator: Any) -> None:
+        self.registry.evaluators[name] = evaluator
+
+    def process_task(self, prompt: str, *, agent_name: str) -> str:
+        agent = self.registry.agents[agent_name]
+        result = agent.process_input(prompt)
+        for evaluator in self.registry.evaluators.values():
+            if hasattr(evaluator, "evaluate"):
+                evaluator.evaluate(result)
+        return result
 
 
 __all__ = ["Controller"]

blux/orchestrator/logs.py

@@ -2,6 +2,7 @@
 import json
 from pathlib import Path
 from datetime import datetime
+import json
 from typing import Any
 
 

blux/orchestrator/registry.py

@@ -1,15 +1,19 @@
-"""Model registry and adapter registration.
+"""Simple in-memory registry for orchestrator adapters."""
 
-This registry is in-memory and discovers adapters that implement the simple
-Adapter interface defined by `predict(prompt: str) -> dict`.
-"""
+from __future__ import annotations
+
+import json
 from pathlib import Path
-import yaml
 from typing import Dict, List, Optional
 
+try:  # Optional dependency for YAML parsing
+    import yaml
+except ModuleNotFoundError:  # pragma: no cover - fall back to JSON
+    yaml = None
+
 
 class ModelAdapter:
-    """Adapter interface. Implement `predict`."""
+    """Adapter interface. Implement ``predict``."""
 
     def __init__(self, name: str):
         self.name = name
@@ -36,10 +40,12 @@ class ModelRegistry:
         cfg_path = Path(config_path)
         if not cfg_path.exists():
             raise FileNotFoundError(cfg_path)
-        cfg = yaml.safe_load(cfg_path.read_text())
+        raw = cfg_path.read_text(encoding="utf-8")
+        if yaml is not None:
+            cfg = yaml.safe_load(raw)
+        else:
+            cfg = json.loads(raw)
         registry = cls()
-        # we don't auto-create adapters here — caller will register adapters
-        # but return the registry and the model list in case it's useful
         registry._model_list = cfg.get("models", [])
         return registry
 

blux_ca/adapters/__init__.py

@@ -3,9 +3,13 @@
 from .doctrine import DoctrineAdapter
 from .guard import GuardAdapter
 from .lite import LiteAdapter
-from .quantum import QuantumAdapter
 from .reg import RegistryValidator, RegistrationResult
 
+try:  # Optional dependency guard
+    from .quantum import QuantumAdapter
+except ModuleNotFoundError:  # pragma: no cover - optional import
+    QuantumAdapter = None  # type: ignore
+
 __all__ = [
     "DoctrineAdapter",
     "GuardAdapter",

blux_ca/adapters/bq_cli.py

@@ -0,0 +1,55 @@
+"""Integration helpers for orchestrating reflection via ``bq-cli``."""
+
+from __future__ import annotations
+
+import shlex
+import shutil
+import subprocess
+from dataclasses import dataclass
+from typing import Callable, Iterable, List, Sequence
+
+
+@dataclass
+class BQTask:
+    command: List[str]
+    executed: bool
+    output: str
+
+
+class BQCliAdapter:
+    """Lightweight wrapper around ``bq-cli`` commands."""
+
+    def __init__(
+        self,
+        executable: str | None = None,
+        runner: Callable[[List[str]], subprocess.CompletedProcess[str]] | None = None,
+    ) -> None:
+        self.executable = executable or shutil.which("bq") or "bq"
+        self.runner = runner or (lambda cmd: subprocess.run(cmd, capture_output=True, text=True))
+
+    def available(self) -> bool:
+        return shutil.which(self.executable) is not None
+
+    def plan_reflection(self, prompt: str, *, koans: Sequence[str]) -> List[str]:
+        command = [self.executable, "reflect", "--prompt", prompt]
+        for koan in koans:
+            command.extend(["--koan", koan])
+        return command
+
+    def run_reflection(
+        self,
+        prompt: str,
+        *,
+        koans: Iterable[str] = (),
+        dry_run: bool = True,
+    ) -> BQTask:
+        command = self.plan_reflection(prompt, koans=list(koans))
+        if dry_run or not self.available():
+            quoted = " ".join(shlex.quote(part) for part in command)
+            return BQTask(command=command, executed=False, output=f"dry-run: {quoted}")
+        result = self.runner(command)
+        output = (result.stdout or "") + (result.stderr or "")
+        return BQTask(command=command, executed=result.returncode == 0, output=output)
+
+
+__all__ = ["BQCliAdapter", "BQTask"]

blux_ca/core/__init__.py

@@ -1,24 +1,38 @@
 """Core modules for BLUX-cA."""
 
 from .audit import AuditLog, AuditRecord
+from .compass import CompassAxis, IntentCompass, IntentProfile, IntentSignal
 from .constitution import ConstitutionEngine, DoctrineVerdict
 from .discernment import DiscernmentCompass, DiscernmentDecision, IntentType
+from .heart import ConsciousHeart, ConsciousOutput
 from .intervention import compassionate_edge, layered_truth, light_shift, mirror
+from .koan import Koan, KoanProbe
+from .memory import ConsentMemory, MemoryEntry
 from .perception import PerceptionInput, PerceptionLayer
 from .reflection import ReflectionEngine, ReflectionInsight
 
 __all__ = [
     "AuditLog",
     "AuditRecord",
+    "CompassAxis",
+    "IntentCompass",
+    "IntentProfile",
+    "IntentSignal",
     "ConstitutionEngine",
     "DoctrineVerdict",
     "DiscernmentCompass",
     "DiscernmentDecision",
     "IntentType",
+    "ConsciousHeart",
+    "ConsciousOutput",
     "compassionate_edge",
     "layered_truth",
     "light_shift",
     "mirror",
+    "Koan",
+    "KoanProbe",
+    "ConsentMemory",
+    "MemoryEntry",
     "PerceptionInput",
     "PerceptionLayer",
     "ReflectionEngine",

blux_ca/core/audit.py

@@ -2,11 +2,12 @@
 
 from __future__ import annotations
 
+import hashlib
 import json
 from dataclasses import dataclass, asdict
 from datetime import datetime, timezone
 from pathlib import Path
-from typing import Iterable, List
+from typing import Iterable, Iterator, List, Optional
 
 
 @dataclass
@@ -16,21 +17,67 @@ class AuditRecord:
     verdict: str
     doctrine_refs: List[str]
     rationale: str
+    prev_hash: Optional[str] = None
+    chain_hash: Optional[str] = None
+
+    def to_dict(self) -> dict:
+        return asdict(self)
+
+    @classmethod
+    def from_dict(cls, payload: dict) -> "AuditRecord":
+        return cls(
+            timestamp=payload["timestamp"],
+            input_hash=payload["input_hash"],
+            verdict=payload["verdict"],
+            doctrine_refs=list(payload.get("doctrine_refs", [])),
+            rationale=payload["rationale"],
+            prev_hash=payload.get("prev_hash"),
+            chain_hash=payload.get("chain_hash"),
+        )
 
 
 class AuditLog:
-    """Append-only JSONL audit log."""
+    """Append-only JSONL audit log with hash chaining."""
 
-    def __init__(self, path: Path | None = None) -> None:
+    def __init__(self, path: Path | None = None, *, hash_alg: str = "sha256") -> None:
         self.path = path or Path.home() / ".config" / "blux-ca" / "audit" / "decisions.jsonl"
         self.path.parent.mkdir(parents=True, exist_ok=True)
+        self._hash_alg = hash_alg
+        self._last_hash = self._load_tail_hash()
+
+    def _hash(self, payload: dict) -> str:
+        canonical = json.dumps(payload, sort_keys=True, ensure_ascii=False).encode("utf-8")
+        return hashlib.new(self._hash_alg, canonical).hexdigest()
+
+    def _load_tail_hash(self) -> Optional[str]:
+        if not self.path.exists():
+            return None
+        last_line = ""
+        with self.path.open("r", encoding="utf-8") as handle:
+            for line in handle:
+                if line.strip():
+                    last_line = line
+        if not last_line:
+            return None
+        payload = json.loads(last_line)
+        return payload.get("chain_hash")
 
-    def append(self, record: AuditRecord) -> None:
+    def append(self, record: AuditRecord) -> AuditRecord:
+        payload = record.to_dict()
+        payload["prev_hash"] = self._last_hash
+        payload["chain_hash"] = self._hash({key: payload[key] for key in payload if key != "chain_hash"})
+        self._last_hash = payload["chain_hash"]
         with self.path.open("a", encoding="utf-8") as handle:
-            handle.write(json.dumps(asdict(record), ensure_ascii=False) + "\n")
+            handle.write(json.dumps(payload, ensure_ascii=False) + "\n")
+        return AuditRecord.from_dict(payload)
 
     def create_record(
-        self, *, input_hash: str, verdict: str, doctrine_refs: Iterable[str], rationale: str
+        self,
+        *,
+        input_hash: str,
+        verdict: str,
+        doctrine_refs: Iterable[str],
+        rationale: str,
     ) -> AuditRecord:
         return AuditRecord(
             timestamp=datetime.now(timezone.utc).isoformat(),
@@ -38,7 +85,50 @@ class AuditLog:
             verdict=verdict,
             doctrine_refs=list(doctrine_refs),
             rationale=rationale,
+            prev_hash=self._last_hash,
         )
 
+    def playback(self, *, limit: int | None = None) -> List[AuditRecord]:
+        records = []
+        for index, record in enumerate(self._iter_records()):
+            records.append(record)
+            if limit is not None and index + 1 >= limit:
+                break
+        return records
+
+    def verify_chain(self) -> bool:
+        previous: Optional[str] = None
+        for record in self._iter_records():
+            expected = self._hash(
+                {
+                    key: getattr(record, key)
+                    for key in (
+                        "timestamp",
+                        "input_hash",
+                        "verdict",
+                        "doctrine_refs",
+                        "rationale",
+                        "prev_hash",
+                    )
+                }
+            )
+            if record.prev_hash != previous or record.chain_hash != expected:
+                return False
+            previous = record.chain_hash
+        return True
+
+    def _iter_records(self) -> Iterator[AuditRecord]:
+        if not self.path.exists():
+            return iter(())
+
+        def generator() -> Iterator[AuditRecord]:
+            with self.path.open("r", encoding="utf-8") as handle:
+                for line in handle:
+                    if not line.strip():
+                        continue
+                    yield AuditRecord.from_dict(json.loads(line))
+
+        return generator()
+
 
 __all__ = ["AuditLog", "AuditRecord"]

blux_ca/core/compass/__init__.py

@@ -0,0 +1,10 @@
+"""Intent compass for BLUX-cA."""
+
+from .intent import CompassAxis, IntentCompass, IntentProfile, IntentSignal
+
+__all__ = [
+    "CompassAxis",
+    "IntentCompass",
+    "IntentProfile",
+    "IntentSignal",
+]

blux_ca/core/compass/intent.py

@@ -0,0 +1,83 @@
+"""Intent compass scoring four doctrinal axes."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from enum import Enum
+from typing import Dict, Iterable, List
+
+
+class CompassAxis(str, Enum):
+    """Intent axes emphasised by BLUX doctrine."""
+
+    TRUTH = "truth"
+    INTEGRITY = "integrity"
+    COMPASSION = "compassion"
+    AWARENESS = "awareness"
+
+
+@dataclass
+class IntentSignal:
+    """Keyword evidence that contributed to an axis score."""
+
+    axis: CompassAxis
+    keyword: str
+
+
+@dataclass
+class IntentProfile:
+    """Result of intent classification along doctrine axes."""
+
+    dominant: CompassAxis
+    scores: Dict[CompassAxis, float]
+    signals: List[IntentSignal]
+
+    def narrative(self) -> str:
+        dominant_score = self.scores[self.dominant]
+        return (
+            f"Dominant intent {self.dominant.value} with confidence {dominant_score:.2f}."
+        )
+
+
+class IntentCompass:
+    """Classifies natural language against doctrine axes."""
+
+    _KEYWORDS: Dict[CompassAxis, tuple[str, ...]] = {
+        CompassAxis.TRUTH: ("truth", "honest", "fact", "evidence", "reality"),
+        CompassAxis.INTEGRITY: ("boundary", "integrity", "duty", "ethic", "principle"),
+        CompassAxis.COMPASSION: ("help", "care", "support", "compassion", "kind"),
+        CompassAxis.AWARENESS: (
+            "reflect",
+            "aware",
+            "notice",
+            "mindful",
+            "observe",
+        ),
+    }
+
+    def __init__(self, *, baseline: float = 0.1) -> None:
+        self._baseline = max(0.0, baseline)
+
+    def classify(self, text: str, *, hints: Iterable[str] | None = None) -> IntentProfile:
+        lowered = text.lower()
+        scores: Dict[CompassAxis, float] = {axis: self._baseline for axis in CompassAxis}
+        signals: List[IntentSignal] = []
+
+        for axis, keywords in self._KEYWORDS.items():
+            hits = [kw for kw in keywords if kw in lowered]
+            if hints:
+                hits.extend(kw for kw in hints if kw in keywords)
+            if hits:
+                signals.extend(IntentSignal(axis=axis, keyword=kw) for kw in hits)
+                scores[axis] += 0.2 * len(hits)
+
+        # Encourage compassion as legacy voice default if nothing detected
+        if not signals:
+            signals.append(IntentSignal(axis=CompassAxis.COMPASSION, keyword="legacy"))
+            scores[CompassAxis.COMPASSION] += 0.05
+
+        dominant = max(scores, key=scores.get)
+        return IntentProfile(dominant=dominant, scores=scores, signals=signals)
+
+
+__all__ = ["CompassAxis", "IntentCompass", "IntentProfile", "IntentSignal"]

blux_ca/core/discernment.py

@@ -4,7 +4,9 @@ from __future__ import annotations
 
 from dataclasses import dataclass
 from enum import Enum
-from typing import Dict
+from typing import Optional
+
+from .compass import IntentCompass, IntentProfile
 
 
 class IntentType(str, Enum):
@@ -17,18 +19,23 @@ class IntentType(str, Enum):
 class DiscernmentDecision:
     intent: IntentType
     rationale: str
+    profile: Optional[IntentProfile] = None
 
 
 class DiscernmentCompass:
-    """Classifies intent using simple heuristics."""
+    """Classifies intent using heuristics and the doctrine compass."""
+
+    def __init__(self, *, compass: IntentCompass | None = None) -> None:
+        self._compass = compass or IntentCompass()
 
     def classify(self, text: str) -> DiscernmentDecision:
+        profile = self._compass.classify(text)
         lowered = text.lower()
         if any(word in lowered for word in ("hurt", "harm", "kill")):
-            return DiscernmentDecision(IntentType.HARM, "Detected explicit harm intent.")
+            return DiscernmentDecision(IntentType.HARM, "Detected explicit harm intent.", profile)
         if any(word in lowered for word in ("enjoy", "love", "want", "indulge")):
-            return DiscernmentDecision(IntentType.INDULGER, "Language emphasises indulgence.")
-        return DiscernmentDecision(IntentType.STRUGGLER, "Defaulting to supportive framing.")
+            return DiscernmentDecision(IntentType.INDULGER, "Language emphasises indulgence.", profile)
+        return DiscernmentDecision(IntentType.STRUGGLER, "Defaulting to supportive framing.", profile)
 
 
 __all__ = ["DiscernmentCompass", "DiscernmentDecision", "IntentType"]

blux_ca/core/heart.py

@@ -0,0 +1,126 @@
+"""Conscious heart orchestrating discernment, reflection, and policy."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Iterable, List, Optional
+
+from .audit import AuditLog, AuditRecord
+from .compass import IntentCompass, IntentProfile
+from .constitution import ConstitutionEngine
+from .discernment import DiscernmentCompass, DiscernmentDecision
+from .koan import KoanProbe
+from .memory import ConsentMemory, MemoryEntry
+from .perception import PerceptionLayer, PerceptionInput
+from .reflection import ReflectionEngine, ReflectionInsight
+
+try:  # Optional adapter import
+    from blux_ca.adapters.bq_cli import BQCliAdapter, BQTask
+except Exception:  # pragma: no cover - fallback when adapter deps missing
+    BQCliAdapter = None  # type: ignore
+    BQTask = None  # type: ignore
+
+
+@dataclass
+class ConsciousOutput:
+    perception: PerceptionInput
+    decision: DiscernmentDecision
+    intent_profile: IntentProfile
+    verdict: str
+    rationale: str
+    reflection: ReflectionInsight
+    koans: List[str]
+    audit_record: AuditRecord
+    memory_entry: MemoryEntry
+    bq_task: Optional["BQTask"]
+    voice: str
+
+
+class ConsciousHeart:
+    """Central coordination engine for BLUX-cA."""
+
+    def __init__(
+        self,
+        *,
+        perception: PerceptionLayer | None = None,
+        discernment: DiscernmentCompass | None = None,
+        constitution: ConstitutionEngine | None = None,
+        reflection: ReflectionEngine | None = None,
+        koans: KoanProbe | None = None,
+        memory: ConsentMemory | None = None,
+        audit: AuditLog | None = None,
+        intent_compass: IntentCompass | None = None,
+        bq_adapter: Optional["BQCliAdapter"] = None,
+        legacy_voice: str = "BLUX-cA legacy-guidance",
+    ) -> None:
+        self.perception = perception or PerceptionLayer(default_tags=["local", "phase1"])
+        self.intent_compass = intent_compass or IntentCompass()
+        self.discernment = discernment or DiscernmentCompass(compass=self.intent_compass)
+        self.constitution = constitution or ConstitutionEngine()
+        self.reflection = reflection or ReflectionEngine()
+        self.koans = koans or KoanProbe()
+        self.memory = memory or ConsentMemory()
+        self.audit = audit or AuditLog()
+        self.bq_adapter = bq_adapter
+        self.legacy_voice = legacy_voice
+
+    def process(
+        self,
+        text: str,
+        *,
+        tags: Iterable[str] | None = None,
+        consent: bool = False,
+        dry_run_reflection: bool = True,
+    ) -> ConsciousOutput:
+        perception = self.perception.observe(text, tags=tags)
+        decision = self.discernment.classify(perception.text)
+        profile = decision.profile or self.intent_compass.classify(perception.text)
+        reflection = self.reflection.reflect(perception.text, seeds=[profile.narrative()])
+        koan_prompts = [koan.prompt for koan in self.koans.probe(profile, intent=decision.intent)]
+        verdict = self.constitution.evaluate(
+            insights=reflection.chain,
+            intent=decision.intent.value,
+        )
+        audit_record = self.audit.append(
+            self.audit.create_record(
+                input_hash=perception.fingerprint,
+                verdict=verdict.decision,
+                doctrine_refs=verdict.doctrine_refs,
+                rationale=verdict.reason,
+            )
+        )
+        memory_entry = self.memory.store(
+            perception,
+            consent=consent,
+            summary=reflection.summary,
+            metadata={
+                "verdict": verdict.decision,
+                "intent": decision.intent.value,
+                "dominant_axis": profile.dominant.value,
+            },
+        )
+
+        bq_task = None
+        if self.bq_adapter is not None:
+            koans_payload = koan_prompts or [reflection.summary]
+            bq_task = self.bq_adapter.run_reflection(
+                reflection.summary,
+                koans=koans_payload,
+                dry_run=dry_run_reflection,
+            )
+
+        voice = f"{self.legacy_voice}: {verdict.reason}"
+
+        return ConsciousOutput(
+            perception=perception,
+            decision=decision,
+            intent_profile=profile,
+            verdict=verdict.decision,
+            rationale=verdict.reason,
+            reflection=reflection,
+            koans=koan_prompts,
+            audit_record=audit_record,
+            memory_entry=memory_entry,
+            bq_task=bq_task,
+            voice=voice,
+        )

blux_ca/core/koan.py

@@ -0,0 +1,54 @@
+"""Koan probes guiding reflective inquiry."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Dict, Iterable, List, Mapping
+
+from .compass import CompassAxis, IntentProfile
+from .discernment import IntentType
+
+
+@dataclass
+class Koan:
+    """A reflective probe anchored in doctrine."""
+
+    axis: CompassAxis
+    intent: IntentType
+    prompt: str
+
+
+class KoanProbe:
+    """Selects koan prompts based on intent profile."""
+
+    def __init__(self, library: Mapping[CompassAxis, Iterable[str]] | None = None) -> None:
+        self._library: Dict[CompassAxis, List[str]] = {
+            axis: list(prompts)
+            for axis, prompts in (
+                library.items()
+                if library
+                else {
+                    CompassAxis.TRUTH: [
+                        "What story are you telling that might be incomplete?",
+                        "Where does evidence ask for a clearer lantern?",
+                    ],
+                    CompassAxis.INTEGRITY: [
+                        "Which boundary, if honoured, keeps you aligned?",
+                        "What duty do you owe to the person you are becoming?",
+                    ],
+                    CompassAxis.COMPASSION: [
+                        "How can care be offered without losing yourself?",
+                        "What tenderness is waiting to be voiced?",
+                    ],
+                    CompassAxis.AWARENESS: [
+                        "What are you noticing beneath the first thought?",
+                        "Where is the silence inviting you to listen deeper?",
+                    ],
+                }.items()
+            )
+        }
+
+    def probe(self, profile: IntentProfile, *, intent: IntentType, limit: int = 2) -> List[Koan]:
+        prompts = self._library.get(profile.dominant, []) or self._library[CompassAxis.AWARENESS]
+        selected = prompts[: max(1, limit)]
+        return [Koan(axis=profile.dominant, intent=intent, prompt=prompt) for prompt in selected]

blux_ca/core/memory.py

@@ -0,0 +1,85 @@
+"""Consent-based memory with local-first persistence."""
+
+from __future__ import annotations
+
+import json
+from dataclasses import dataclass, asdict
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Dict, List, Optional
+
+from .perception import PerceptionInput
+
+
+@dataclass
+class MemoryEntry:
+    fingerprint: str
+    consent: bool
+    summary: str
+    tags: List[str]
+    timestamp: str
+    metadata: Dict[str, str]
+
+    @classmethod
+    def from_dict(cls, payload: Dict[str, str]) -> "MemoryEntry":
+        return cls(
+            fingerprint=payload["fingerprint"],
+            consent=payload["consent"],
+            summary=payload["summary"],
+            tags=list(payload.get("tags", [])),
+            timestamp=payload["timestamp"],
+            metadata=dict(payload.get("metadata", {})),
+        )
+
+
+class ConsentMemory:
+    """Stores entries locally only when consent is provided."""
+
+    def __init__(self, path: Path | None = None) -> None:
+        self._path = path or Path.home() / ".config" / "blux-ca" / "memory" / "consent.jsonl"
+        self._path.parent.mkdir(parents=True, exist_ok=True)
+        self._session: Dict[str, MemoryEntry] = {}
+
+    def store(
+        self,
+        perception: PerceptionInput,
+        *,
+        consent: bool,
+        summary: str,
+        metadata: Optional[Dict[str, str]] = None,
+    ) -> MemoryEntry:
+        entry = MemoryEntry(
+            fingerprint=perception.fingerprint,
+            consent=consent,
+            summary=summary,
+            tags=list(perception.tags),
+            timestamp=datetime.now(timezone.utc).isoformat(),
+            metadata=dict(metadata or {}),
+        )
+        self._session[entry.fingerprint] = entry
+        if consent:
+            self._append(entry)
+        return entry
+
+    def _append(self, entry: MemoryEntry) -> None:
+        with self._path.open("a", encoding="utf-8") as handle:
+            handle.write(json.dumps(asdict(entry), ensure_ascii=False) + "\n")
+
+    def session_entries(self) -> List[MemoryEntry]:
+        return list(self._session.values())
+
+    def persistent_entries(self, *, limit: int | None = None) -> List[MemoryEntry]:
+        if not self._path.exists():
+            return []
+        entries: List[MemoryEntry] = []
+        with self._path.open("r", encoding="utf-8") as handle:
+            for line in handle:
+                if not line.strip():
+                    continue
+                entries.append(MemoryEntry.from_dict(json.loads(line)))
+                if limit and len(entries) >= limit:
+                    break
+        return entries
+
+
+__all__ = ["ConsentMemory", "MemoryEntry"]

tests/ca/test_audit.py

@@ -6,6 +6,24 @@ from blux_ca.core.audit import AuditLog
 def test_audit_appends(tmp_path: Path):
     audit = AuditLog(path=tmp_path / "audit.jsonl")
     record = audit.create_record(input_hash="abc", verdict="allow", doctrine_refs=["law"], rationale="ok")
-    audit.append(record)
+    stored = audit.append(record)
     contents = (tmp_path / "audit.jsonl").read_text(encoding="utf-8").strip()
     assert "allow" in contents
+    assert stored.chain_hash is not None
+    assert audit.verify_chain()
+
+
+def test_audit_playback(tmp_path: Path):
+    audit = AuditLog(path=tmp_path / "audit.jsonl")
+    for idx in range(3):
+        audit.append(
+            audit.create_record(
+                input_hash=str(idx),
+                verdict="allow",
+                doctrine_refs=["law"],
+                rationale=f"ok-{idx}",
+            )
+        )
+    records = audit.playback(limit=2)
+    assert len(records) == 2
+    assert records[0].chain_hash is not None

tests/ca/test_bq_cli.py

@@ -0,0 +1,9 @@
+from blux_ca.adapters.bq_cli import BQCliAdapter
+
+
+def test_bq_cli_dry_run():
+    adapter = BQCliAdapter(executable="bq")
+    task = adapter.run_reflection("Reflect with me", koans=["What is true?"], dry_run=True)
+    assert task.executed is False
+    assert "reflect" in " ".join(task.command)
+    assert "dry-run" in task.output

tests/ca/test_compass.py

@@ -0,0 +1,14 @@
+from blux_ca.core.compass import CompassAxis, IntentCompass
+
+
+def test_intent_compass_detects_truth_axis():
+    compass = IntentCompass()
+    profile = compass.classify("I seek truth and honest evidence")
+    assert profile.dominant is CompassAxis.TRUTH
+    assert any(signal.axis is CompassAxis.TRUTH for signal in profile.signals)
+
+
+def test_intent_compass_defaults_to_compassion():
+    compass = IntentCompass()
+    profile = compass.classify("Just words")
+    assert profile.dominant in {CompassAxis.COMPASSION, CompassAxis.AWARENESS}

tests/ca/test_heart.py

@@ -0,0 +1,17 @@
+from blux_ca.adapters.bq_cli import BQCliAdapter
+from blux_ca.core import AuditLog, ConsciousHeart, ConsentMemory
+
+
+def test_conscious_heart_flow(tmp_path):
+    heart = ConsciousHeart(
+        memory=ConsentMemory(path=tmp_path / "memory.jsonl"),
+        audit=AuditLog(path=tmp_path / "audit.jsonl"),
+        bq_adapter=BQCliAdapter(executable="bq"),
+    )
+    output = heart.process("I seek truth and help others", consent=True)
+    assert output.verdict in {"allow", "reflect", "deny"}
+    assert output.audit_record.chain_hash is not None
+    assert output.memory_entry.consent is True
+    assert output.bq_task is not None
+    assert output.voice.startswith("BLUX-cA")
+    assert output.koans

tests/ca/test_memory.py

@@ -0,0 +1,19 @@
+from blux_ca.core.memory import ConsentMemory
+from blux_ca.core.perception import PerceptionLayer
+
+
+def test_consent_memory_respects_consent(tmp_path):
+    memory = ConsentMemory(path=tmp_path / "memory.jsonl")
+    perception = PerceptionLayer().observe("Hold this gently")
+    entry = memory.store(perception, consent=False, summary="gentle", metadata={"note": "test"})
+    assert entry.consent is False
+    assert memory.persistent_entries() == []
+
+
+def test_consent_memory_persists_with_consent(tmp_path):
+    memory = ConsentMemory(path=tmp_path / "memory.jsonl")
+    perception = PerceptionLayer().observe("Support me with care")
+    memory.store(perception, consent=True, summary="care")
+    persisted = memory.persistent_entries()
+    assert len(persisted) == 1
+    assert persisted[0].summary == "care"