-
Run (Text / Decoded Intent)
+
Run (Declared Text Command)
@@ -33,7 +34,7 @@
Run (Text / Decoded Intent)
Run (Signal / Simulated Label)
-
+
diff --git a/synapdrive_ai/interface/web_dashboard.py b/synapdrive_ai/interface/web_dashboard.py
index 41b867c..1d468e4 100644
--- a/synapdrive_ai/interface/web_dashboard.py
+++ b/synapdrive_ai/interface/web_dashboard.py
@@ -1,58 +1,188 @@
from __future__ import annotations
-from flask import Flask, jsonify, render_template, request
+import io
+import json
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any, Callable, Iterable
+from wsgiref.simple_server import make_server
+from wsgiref.types import StartResponse, WSGIEnvironment
+from wsgiref.util import setup_testing_defaults
from synapdrive_ai.pipeline import SynapDrivePipeline
-app = Flask(__name__)
-
-# IMPORTANT: use no-delay mode so:
-# - local dev feels snappy
-# - tests don’t sleep for simulated actuation
-pipeline = SynapDrivePipeline(simulate_delay=False)
-
-
-@app.get("/")
-def home():
- return render_template("index.html")
-
-
-@app.get("/api/log")
-def get_log():
- log = pipeline.get_action_log()
- return jsonify({"count": len(log), "log": log[-50:]})
-
-
-@app.get("/api/assurance")
-def get_assurance():
- return jsonify(
- {
- "report": pipeline.get_assurance_report(),
- "receipts": pipeline.get_assurance_log()[-50:],
- }
- )
-
-
-@app.post("/api/run/text")
-def run_text():
- data = request.get_json(force=True) or {}
- cmd = (data.get("text") or "").strip()
- image = data.get("image")
- out = pipeline.run_text_command(cmd, image_label=image)
- return jsonify(out)
-
-
-@app.post("/api/run/signal")
-def run_signal():
- data = request.get_json(force=True) or {}
- label = data.get("label") # None or string
- image = data.get("image")
- out = pipeline.run_signal_event(label=label, image_label=image)
- return jsonify(out)
-
-
-def main():
- app.run(host="127.0.0.1", port=5055, debug=True)
+_TEMPLATE = Path(__file__).with_name("templates") / "index.html"
+
+
+@dataclass(frozen=True)
+class DashboardResponse:
+ status_code: int
+ data: bytes
+ content_type: str
+
+ def get_json(self) -> Any:
+ if "application/json" not in self.content_type:
+ return None
+ return json.loads(self.data.decode("utf-8"))
+
+
+class DashboardTestClient:
+ """In-process WSGI client used by the repository's route tests."""
+
+ def __init__(self, application: "DashboardApplication") -> None:
+ self.application = application
+
+ def get(self, path: str) -> DashboardResponse:
+ return self._request("GET", path, None)
+
+ def post(self, path: str, json: Any = None) -> DashboardResponse:
+ return self._request("POST", path, json)
+
+ def _request(self, method: str, path: str, payload: Any) -> DashboardResponse:
+ body = b"" if payload is None else json.dumps(payload).encode("utf-8")
+ environ: dict[str, Any] = {}
+ setup_testing_defaults(environ)
+ environ.update(
+ {
+ "REQUEST_METHOD": method,
+ "PATH_INFO": path,
+ "CONTENT_TYPE": "application/json" if body else "",
+ "CONTENT_LENGTH": str(len(body)),
+ "wsgi.input": io.BytesIO(body),
+ }
+ )
+ captured: dict[str, Any] = {}
+
+ def start_response(
+ status: str,
+ headers: list[tuple[str, str]],
+ exc_info: object = None,
+ ) -> Callable[[bytes], object]:
+ del exc_info
+ captured["status"] = status
+ captured["headers"] = dict(headers)
+ return lambda data: len(data)
+
+ chunks = self.application(environ, start_response)
+ data = b"".join(chunks)
+ status_code = int(str(captured["status"]).split()[0])
+ headers = captured.get("headers", {})
+ return DashboardResponse(
+ status_code=status_code,
+ data=data,
+ content_type=str(headers.get("Content-Type", "application/octet-stream")),
+ )
+
+
+class DashboardApplication:
+ """Small localhost-only WSGI dashboard over the canonical governed pipeline."""
+
+ def __init__(self, pipeline: SynapDrivePipeline | None = None) -> None:
+ self.pipeline = pipeline or SynapDrivePipeline(simulate_delay=False)
+
+ def test_client(self) -> DashboardTestClient:
+ return DashboardTestClient(self)
+
+ def __call__(
+ self,
+ environ: WSGIEnvironment,
+ start_response: StartResponse,
+ ) -> Iterable[bytes]:
+ method = str(environ.get("REQUEST_METHOD", "GET")).upper()
+ path = str(environ.get("PATH_INFO", "/"))
+ try:
+ status, content_type, payload = self._dispatch(method, path, environ)
+ except (json.JSONDecodeError, UnicodeDecodeError, ValueError) as exc:
+ status, content_type, payload = (
+ "400 Bad Request",
+ "application/json; charset=utf-8",
+ self._json_bytes({"error": "invalid-request", "detail": str(exc)}),
+ )
+ except Exception:
+ status, content_type, payload = (
+ "500 Internal Server Error",
+ "application/json; charset=utf-8",
+ self._json_bytes({"error": "internal-server-error"}),
+ )
+ start_response(
+ status,
+ [
+ ("Content-Type", content_type),
+ ("Content-Length", str(len(payload))),
+ ("Cache-Control", "no-store"),
+ ("X-Content-Type-Options", "nosniff"),
+ ],
+ )
+ return [payload]
+
+ def _dispatch(
+ self,
+ method: str,
+ path: str,
+ environ: WSGIEnvironment
+ ) -> tuple[str, str, bytes]:
+ if method == "GET" and path == "/":
+ return "200 OK", "text/html; charset=utf-8", _TEMPLATE.read_bytes()
+ if method == "GET" and path == "/api/log":
+ log = self.pipeline.get_action_log()
+ return self._json_response({"count": len(log), "log": log[-50:]})
+ if method == "GET" and path == "/api/assurance":
+ return self._json_response(
+ {
+ "report": self.pipeline.get_assurance_report(),
+ "receipts": self.pipeline.get_assurance_log()[-50:],
+ }
+ )
+ if method == "POST" and path == "/api/run/text":
+ data = self._read_json(environ)
+ command = str(data.get("text") or "").strip()
+ image = data.get("image")
+ return self._json_response(
+ self.pipeline.run_text_command(command, image_label=image)
+ )
+ if method == "POST" and path == "/api/run/signal":
+ data = self._read_json(environ)
+ label = data.get("label")
+ image = data.get("image")
+ return self._json_response(
+ self.pipeline.run_signal_event(label=label, image_label=image)
+ )
+ if path.startswith("/api/"):
+ return self._json_response({"error": "not-found"}, status="404 Not Found")
+ return "404 Not Found", "text/plain; charset=utf-8", b"Not Found\n"
+
+ @staticmethod
+ def _read_json(environ: dict[str, Any]) -> dict[str, Any]:
+ raw_length = str(environ.get("CONTENT_LENGTH") or "0")
+ length = int(raw_length) if raw_length.isdigit() else 0
+ raw = environ["wsgi.input"].read(length) if length else b"{}"
+ payload = json.loads(raw.decode("utf-8"))
+ if not isinstance(payload, dict):
+ raise ValueError("JSON body must be an object")
+ return payload
+
+ @staticmethod
+ def _json_bytes(payload: Any) -> bytes:
+ return json.dumps(payload, default=str, separators=(",", ":")).encode("utf-8")
+
+ def _json_response(
+ self,
+ payload: Any,
+ *,
+ status: str = "200 OK",
+ ) -> tuple[str, str, bytes]:
+ return status, "application/json; charset=utf-8", self._json_bytes(payload)
+
+
+app = DashboardApplication()
+
+
+def main() -> None:
+ host = "127.0.0.1"
+ port = 5055
+ print(f"SynapDrive-AI dashboard listening on http://{host}:{port}")
+ with make_server(host, port, app) as server:
+ server.serve_forever()
if __name__ == "__main__":
diff --git a/synapdrive_ai/main/__init__.py b/synapdrive_ai/main/__init__.py
new file mode 100644
index 0000000..3f1c04f
--- /dev/null
+++ b/synapdrive_ai/main/__init__.py
@@ -0,0 +1 @@
+"""Command-line and integration entry points."""
diff --git a/synapdrive_ai/main/cli.py b/synapdrive_ai/main/cli.py
index 16397c6..6802f7e 100644
--- a/synapdrive_ai/main/cli.py
+++ b/synapdrive_ai/main/cli.py
@@ -19,8 +19,14 @@ def _print_summary(out: dict) -> None:
print(json.dumps(out.get("result", {}), indent=2, default=str))
print("\nevaluation:")
print(json.dumps(out.get("evaluation", {}), indent=2, default=str))
+ print("\nruntime:")
+ print(json.dumps(out.get("runtime", {}), indent=2, default=str))
+ print("\nreality:")
+ print(json.dumps(out.get("reality", {}), indent=2, default=str))
print("\nassurance:")
print(json.dumps(out.get("assurance", {}), indent=2, default=str))
+ print("\nevidence:")
+ print(json.dumps(out.get("evidence", {}), indent=2, default=str))
def build_parser() -> argparse.ArgumentParser:
@@ -31,26 +37,34 @@ def build_parser() -> argparse.ArgumentParser:
mode = p.add_mutually_exclusive_group(required=True)
mode.add_argument(
- "--text", help='Run one cycle using decoded intent text (e.g. "move left", "stop").'
+ "--text",
+ help='Run one cycle using caller-declared command text (e.g. "move left", "stop").',
)
mode.add_argument(
"--signal",
nargs="?",
const="RANDOM",
- help="Run one cycle using a signal label (or RANDOM).",
+ help="Run one cycle using an explicit synthetic fixture label (or seeded RANDOM).",
)
mode.add_argument(
- "--brainflow", action="store_true", help="Run one cycle using BrainFlow input (optional)."
+ "--brainflow",
+ action="store_true",
+ help="Acquire one BrainFlow window. CLI has no implicit decoder, so this path abstains.",
)
mode.add_argument(
- "--lsl", action="store_true", help="Run one cycle using LSL input (optional)."
+ "--lsl",
+ action="store_true",
+ help="Acquire one LSL window. CLI has no implicit decoder, so this path abstains.",
)
mode.add_argument("--replay", help="Replay JSONL records from a prior --record run.")
p.add_argument(
"--image",
default=None,
- help="Optional simulated vision label (road, hazard, person, vehicle).",
+ help=(
+ "Optional caller-declared context label (road, hazard, person, vehicle); "
+ "no image model runs."
+ ),
)
p.add_argument("--count", type=int, default=1, help="How many cycles to run (default: 1).")
p.add_argument(
@@ -61,6 +75,22 @@ def build_parser() -> argparse.ArgumentParser:
p.add_argument(
"--record", default=None, help="Write each cycle to a JSONL file (reproducible runs)."
)
+ p.add_argument(
+ "--evidence-out",
+ default=None,
+ help="Export the tamper-evident cycle ledger to JSONL after the run.",
+ )
+ p.add_argument(
+ "--errp-probability",
+ type=float,
+ default=None,
+ help="Optional externally-derived error-potential probability for feedback testing.",
+ )
+ p.add_argument(
+ "--reject-feedback",
+ action="store_true",
+ help="Explicitly mark the action feedback as rejected.",
+ )
p.add_argument(
"--no-delay",
action="store_true",
@@ -68,7 +98,7 @@ def build_parser() -> argparse.ArgumentParser:
)
# BrainFlow options
- p.add_argument("--bf-board-id", type=int, default=0)
+ p.add_argument("--bf-board-id", type=int, default=-1)
p.add_argument("--bf-serial-port", default=None)
p.add_argument("--bf-seconds", type=float, default=2.0)
@@ -88,6 +118,12 @@ def main(argv=None) -> int:
pipe = SynapDrivePipeline(simulate_delay=simulate_delay)
recorder = JsonlRecorder(args.record) if args.record else None
+ feedback = {}
+ if args.errp_probability is not None:
+ feedback["errp_probability"] = args.errp_probability
+ if args.reject_feedback:
+ feedback["accepted"] = False
+ feedback = feedback or None
if args.replay:
# Replay should be deterministic + fast
@@ -95,8 +131,12 @@ def main(argv=None) -> int:
for rec in iter_jsonl(args.replay):
intent_packet = rec["intent_packet"]
image_label = rec.get("image_label")
- out = pipe.run_intent_packet(intent_packet, image_label=image_label)
+ out = pipe.run_intent_packet(
+ intent_packet, image_label=image_label, feedback=feedback
+ )
_print_summary(out)
+ if args.evidence_out:
+ pipe.evidence.export_jsonl(args.evidence_out)
return 0
for i in range(args.count):
@@ -107,7 +147,9 @@ def main(argv=None) -> int:
if args.text is not None:
mode = "text"
raw_input = {"text": args.text}
- out = pipe.run_text_command(args.text, image_label=args.image)
+ out = pipe.run_text_command(
+ args.text, image_label=args.image, feedback=feedback
+ )
elif args.brainflow:
mode = "brainflow"
@@ -120,7 +162,9 @@ def main(argv=None) -> int:
)
intent_packet = src.next_intent_packet()
raw_input = {"brainflow": {"board_id": args.bf_board_id, "seconds": args.bf_seconds}}
- out = pipe.run_intent_packet(intent_packet, image_label=args.image)
+ out = pipe.run_intent_packet(
+ intent_packet, image_label=args.image, feedback=feedback
+ )
elif args.lsl:
mode = "lsl"
@@ -136,13 +180,17 @@ def main(argv=None) -> int:
raw_input = {
"lsl": {"name": args.lsl_name, "type": args.lsl_type, "seconds": args.lsl_seconds}
}
- out = pipe.run_intent_packet(intent_packet, image_label=args.image)
+ out = pipe.run_intent_packet(
+ intent_packet, image_label=args.image, feedback=feedback
+ )
else:
mode = "signal"
label = None if args.signal == "RANDOM" else args.signal
raw_input = {"label": label or "RANDOM"}
- out = pipe.run_signal_event(label=label, image_label=args.image)
+ out = pipe.run_signal_event(
+ label=label, image_label=args.image, feedback=feedback
+ )
_print_summary(out)
@@ -163,6 +211,9 @@ def main(argv=None) -> int:
if i < args.count - 1 and args.interval > 0:
time.sleep(args.interval)
+ if args.evidence_out:
+ pipe.evidence.export_jsonl(args.evidence_out)
+
return 0
diff --git a/synapdrive_ai/main/test_simulation.py b/synapdrive_ai/main/test_simulation.py
index d8238d9..e3b0b3f 100644
--- a/synapdrive_ai/main/test_simulation.py
+++ b/synapdrive_ai/main/test_simulation.py
@@ -1,36 +1,33 @@
from synapdrive_ai.main.integration_runner import SynapDriveExecutor
-def run_simulation():
- engine = SynapDriveExecutor()
+def run_simulation() -> None:
+ engine = SynapDriveExecutor(simulate_delay=False)
- print("\n🔬 SynapDrive-AI AGI Simulation\n")
- simulated_bci_input = input(
- "🧠 Enter simulated intent (e.g. 'navigate forward', 'initiate docking'): "
- )
- simulated_image_input = input(
- "👁️ Enter simulated visual label (e.g. 'road', 'hazard', or leave blank): "
+ print("\nSynapDrive-AI governed simulation\n")
+ command = input("Enter a declared text intent (for example 'move left' or 'stop'): ")
+ visual = input(
+ "Optional declared visual context label (road, hazard, person, vehicle; blank for none): "
)
- result = engine.run_once(simulated_bci_input, simulated_image_input or None)
+ result = engine.run_once(command, visual or None)
- print("\n🧠 AGI Decision Summary:")
- print("------------------------")
- for k, v in result["intent"].items():
- print(f"{k}: {v}")
+ print("\nGoverned decision summary")
+ print("-------------------------")
+ for key, value in result["intent"].items():
+ print(f"{key}: {value}")
- print("\n✅ Execution Result:")
- print("--------------------")
- for k, v in result["result"].items():
- print(f"{k}: {v}")
+ print("\nSimulation result")
+ print("-----------------")
+ for key, value in result["result"].items():
+ print(f"{key}: {value}")
- print("\n📊 Evaluation Metrics:")
- print("----------------------")
- for k, v in result["evaluation"].items():
- print(f"{k}: {v}")
+ print("\nAssurance")
+ print("---------")
+ print(result.get("assurance", {}))
if result["status"] == "blocked":
- print(f"\n🚫 Safety Blocked Execution:\nReason: {result['reason']}")
+ print(f"\nBlocked: {result['reason']}")
if __name__ == "__main__":
diff --git a/synapdrive_ai/memory/__init__.py b/synapdrive_ai/memory/__init__.py
new file mode 100644
index 0000000..add1d98
--- /dev/null
+++ b/synapdrive_ai/memory/__init__.py
@@ -0,0 +1 @@
+"""Episodic and evidence-gated memory stores."""
diff --git a/synapdrive_ai/memory/episodic_memory.py b/synapdrive_ai/memory/episodic_memory.py
index dd92cdc..25527b3 100644
--- a/synapdrive_ai/memory/episodic_memory.py
+++ b/synapdrive_ai/memory/episodic_memory.py
@@ -2,18 +2,12 @@
class EpisodicMemory:
- """
- Stores sequences of past AGI decisions and outcomes.
- Supports context-aware recall and adaptation.
- """
+ """Backward-compatible episodic store with outcome-aware recall."""
def __init__(self):
self.episodes = []
def record_episode(self, intent_packet, result_packet):
- """
- Stores a full cognitive interaction and result as a retrievable memory.
- """
episode = {
"timestamp": time.time(),
"intent": intent_packet["intent"],
@@ -30,5 +24,12 @@ def retrieve_recent(self, count=5):
def find_by_intent(self, intent_keyword):
return [ep for ep in self.episodes if intent_keyword in ep["intent"]]
+ def find_successful_by_intent(self, intent_keyword):
+ return [
+ ep
+ for ep in self.episodes
+ if intent_keyword in ep["intent"] and ep.get("result") in {"success", "executed"}
+ ]
+
def clear_memory(self):
self.episodes.clear()
diff --git a/synapdrive_ai/memory/evidence_memory.py b/synapdrive_ai/memory/evidence_memory.py
new file mode 100644
index 0000000..cb7e813
--- /dev/null
+++ b/synapdrive_ai/memory/evidence_memory.py
@@ -0,0 +1,59 @@
+from __future__ import annotations
+
+import time
+from dataclasses import asdict, dataclass
+from typing import Any, Dict, List, Literal
+
+MemoryState = Literal["validated", "quarantined"]
+
+
+@dataclass(frozen=True)
+class EvidenceMemoryRecord:
+ timestamp: float
+ cycle_id: str
+ intent: str
+ confidence: float
+ outcome: str
+ state: MemoryState
+ reason: str
+
+ def to_dict(self) -> Dict[str, Any]:
+ return asdict(self)
+
+
+class EvidenceMemory:
+ """Outcome-bound memory that refuses to reinforce contradicted cycles."""
+
+ def __init__(self) -> None:
+ self._records: List[EvidenceMemoryRecord] = []
+
+ def record(
+ self,
+ *,
+ cycle_id: str,
+ intent: str,
+ confidence: float,
+ outcome: str,
+ aligned: bool,
+ reason: str,
+ ) -> EvidenceMemoryRecord:
+ record = EvidenceMemoryRecord(
+ timestamp=time.time(),
+ cycle_id=cycle_id,
+ intent=intent,
+ confidence=max(0.0, min(1.0, float(confidence))),
+ outcome=outcome,
+ state="validated" if aligned else "quarantined",
+ reason=reason,
+ )
+ self._records.append(record)
+ return record
+
+ def validated_for_intent(self, intent: str) -> List[EvidenceMemoryRecord]:
+ return [r for r in self._records if r.intent == intent and r.state == "validated"]
+
+ def quarantined(self) -> List[EvidenceMemoryRecord]:
+ return [r for r in self._records if r.state == "quarantined"]
+
+ def all(self) -> List[EvidenceMemoryRecord]:
+ return list(self._records)
diff --git a/synapdrive_ai/neuro/__init__.py b/synapdrive_ai/neuro/__init__.py
index 3861dbc..2a7757d 100644
--- a/synapdrive_ai/neuro/__init__.py
+++ b/synapdrive_ai/neuro/__init__.py
@@ -1,13 +1,9 @@
from synapdrive_ai.neuro.band_analyzer import BANDS, BandPowerAnalyzer, BandPowerResult
from synapdrive_ai.neuro.eeg_loader import EEGLoader, EEGRecording
-from synapdrive_ai.neuro.session_analyzer import EpochResult, SessionAnalyzer, SessionReport
-from synapdrive_ai.neuro.task_planner import (
- ExecutorBridge,
- PlanTrace,
- StepTrace,
- TaskPlan,
- TaskStep,
-)
+from synapdrive_ai.neuro.errp import ErrPFeatureExtractor, ErrPFeatures, ErrPLDAClassifier
+from synapdrive_ai.neuro.fusion import FusionResult, ModalityEvidence, WeightedEvidenceFusion
+from synapdrive_ai.neuro.signal_quality import SignalQualityAnalyzer, SignalQualityReport
+from synapdrive_ai.neuro.uncertainty import ReliabilityCalibrator, UncertaintyEstimate
__all__ = [
"BANDS",
@@ -15,12 +11,14 @@
"BandPowerResult",
"EEGLoader",
"EEGRecording",
- "EpochResult",
- "SessionAnalyzer",
- "SessionReport",
- "ExecutorBridge",
- "PlanTrace",
- "StepTrace",
- "TaskPlan",
- "TaskStep",
+ "ErrPFeatureExtractor",
+ "ErrPFeatures",
+ "ErrPLDAClassifier",
+ "FusionResult",
+ "ModalityEvidence",
+ "ReliabilityCalibrator",
+ "SignalQualityAnalyzer",
+ "SignalQualityReport",
+ "UncertaintyEstimate",
+ "WeightedEvidenceFusion",
]
diff --git a/synapdrive_ai/neuro/band_analyzer.py b/synapdrive_ai/neuro/band_analyzer.py
index 086cda2..1d18e6d 100644
--- a/synapdrive_ai/neuro/band_analyzer.py
+++ b/synapdrive_ai/neuro/band_analyzer.py
@@ -16,7 +16,11 @@
@dataclass
class BandPowerResult:
- """Per-band absolute and relative power, plus derived confidence."""
+ """Per-band spectral features plus an uncalibrated heuristic separation score.
+
+ The historical ``confidence`` field name is retained for API compatibility. It is not
+ a calibrated probability and must not be used as neural-command authority.
+ """
absolute: Dict[str, float]
relative: Dict[str, float]
diff --git a/synapdrive_ai/neuro/cli.py b/synapdrive_ai/neuro/cli.py
index 2d2fcb4..6c33a70 100644
--- a/synapdrive_ai/neuro/cli.py
+++ b/synapdrive_ai/neuro/cli.py
@@ -33,7 +33,7 @@ def cmd_demo(args) -> int:
+ 1.2 * np.sin(2 * np.pi * 20 * t[burst])
+ 0.8 * np.sin(2 * np.pi * 40 * t[burst])
)
- signal += np.random.normal(0, 0.05, n_samples)
+ signal += np.random.default_rng(0).normal(0, 0.05, n_samples)
loader = EEGLoader(sampling_rate=sr)
recording = loader.load_array(
@@ -41,7 +41,7 @@ def cmd_demo(args) -> int:
)
print(f"Synthetic recording: {recording.summary()}")
- print("Motor intent burst injected at t=4s–7s\n")
+ print("Synthetic beta/gamma spectral burst injected at t=4s-7s; no neural intent is claimed.\n")
analyzer = BandPowerAnalyzer(sampling_rate=sr)
result = analyzer.analyze(signal)
@@ -50,15 +50,15 @@ def cmd_demo(args) -> int:
bar = "█" * int(power * 40)
print(f" {band:6s} {power:.3f} {bar}")
print(f" engagement ratio: {result.engagement_ratio:.3f}")
- print(f" intent class: {result.intent_class}")
- print(f" confidence: {result.confidence:.3f}\n")
+ print(f" heuristic class: {result.intent_class}")
+ print(f" heuristic score: {result.confidence:.3f} (uncalibrated)\n")
session_analyzer = SessionAnalyzer(channel="C3", window_s=1.0, step_s=0.5)
report = session_analyzer.run(recording)
print(report.summary())
print("\nEpoch detail:")
- print(f" {'t_start':>7} {'class':>10} {'conf':>6} {'status':>8}")
+ print(f" {'t_start':>7} {'class':>10} {'score':>6} {'status':>12}")
for ep in report.epochs:
print(
f" {ep.time_start_s:>7.1f}s {ep.intent_class:>10} "
@@ -126,17 +126,21 @@ def cmd_threshold(args) -> int:
channel = args.channel or recording.channels[0]
thresholds = [float(t) for t in args.min_conf]
- print(f"Threshold sweep — {path.name} channel={channel}\n")
- print(f" {'threshold':>10} {'blocked%':>10} {'mean_conf':>10} {'n_success':>10}")
+ analyzer = SessionAnalyzer(channel=channel, window_s=args.window, step_s=args.step)
+ report = analyzer.run(recording)
+ scores = np.asarray([epoch.signal_confidence for epoch in report.epochs], dtype=float)
+ print(f"Heuristic spectral-score sweep: {path.name} channel={channel}\n")
+ print(
+ "This is descriptive analysis only. The score is uncalibrated and is not decoder "
+ "confidence, accuracy, or actuation authority.\n"
+ )
+ print(f" {'threshold':>10} {'epochs>=thr':>12} {'fraction':>10} {'mean_score':>10}")
for thresh in thresholds:
- analyzer = SessionAnalyzer(channel=channel, window_s=args.window, step_s=args.step)
- analyzer._pipe.guard.min_confidence_threshold = thresh
- report = analyzer.run(recording)
- print(
- f" {thresh:>10.3f} {report.block_rate * 100:>9.1f}% "
- f"{report.mean_confidence:>10.3f} {report.n_success:>10}"
- )
+ count = int(np.sum(scores >= thresh))
+ fraction = count / len(scores) if len(scores) else 0.0
+ mean_score = float(np.mean(scores)) if len(scores) else 0.0
+ print(f" {thresh:>10.3f} {count:>12} {fraction:>10.3f} {mean_score:>10.3f}")
return 0
@@ -175,7 +179,7 @@ def cmd_plan(args) -> int:
if args.list:
print("Available plans:")
for name, plan in plans.items():
- print(f" {name}: {len(plan)} steps — {plan.name}")
+ print(f" {name}: {len(plan)} steps: {plan.name}")
return 0
task_name = args.task or "reach_grasp"
@@ -208,7 +212,7 @@ def build_parser() -> argparse.ArgumentParser:
a.add_argument("--sr", type=float, default=256.0, help="Sampling rate Hz for CSV/NPY")
a.add_argument("--out", default=None, help="Output directory for JSONL + CSV results")
- t = sub.add_parser("threshold", help="Test multiple confidence thresholds")
+ t = sub.add_parser("threshold", help="Sweep uncalibrated spectral-score thresholds")
t.add_argument("file")
t.add_argument("--min-conf", nargs="+", default=["0.4", "0.5", "0.6", "0.7"])
t.add_argument("--channel", default=None)
diff --git a/synapdrive_ai/neuro/drift.py b/synapdrive_ai/neuro/drift.py
new file mode 100644
index 0000000..278f520
--- /dev/null
+++ b/synapdrive_ai/neuro/drift.py
@@ -0,0 +1,51 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Iterable
+
+import numpy as np
+
+
+@dataclass(frozen=True)
+class DriftResult:
+ score: float
+ drifted: bool
+ threshold: float
+
+
+class FeatureDriftMonitor:
+ """Mahalanobis-distance drift monitor for decoder feature vectors."""
+
+ def __init__(self, quantile: float = 0.99, regularization: float = 1e-6) -> None:
+ if not 0.5 < quantile < 1.0:
+ raise ValueError("quantile must be between 0.5 and 1.0")
+ self.quantile = quantile
+ self.regularization = regularization
+ self.mean_: np.ndarray | None = None
+ self.inv_cov_: np.ndarray | None = None
+ self.threshold_: float | None = None
+
+ def fit(self, features: Iterable[Iterable[float]]) -> "FeatureDriftMonitor":
+ x = np.asarray(list(features), dtype=float)
+ if x.ndim != 2 or len(x) < 3:
+ raise ValueError("need at least three feature vectors")
+ self.mean_ = np.mean(x, axis=0)
+ cov = np.cov(x, rowvar=False)
+ if np.ndim(cov) == 0:
+ cov = np.array([[float(cov)]])
+ cov = np.atleast_2d(cov) + np.eye(x.shape[1]) * self.regularization
+ self.inv_cov_ = np.linalg.pinv(cov)
+ scores = np.array([self._score(row) for row in x])
+ self.threshold_ = max(float(np.quantile(scores, self.quantile)), 1e-9)
+ return self
+
+ def evaluate(self, feature: Iterable[float]) -> DriftResult:
+ if self.mean_ is None or self.inv_cov_ is None or self.threshold_ is None:
+ raise RuntimeError("drift monitor has not been fit")
+ score = self._score(np.asarray(list(feature), dtype=float))
+ return DriftResult(round(score, 6), score > self.threshold_, round(self.threshold_, 6))
+
+ def _score(self, row: np.ndarray) -> float:
+ assert self.mean_ is not None and self.inv_cov_ is not None
+ delta = row - self.mean_
+ return float(np.sqrt(max(0.0, delta @ self.inv_cov_ @ delta)))
diff --git a/synapdrive_ai/neuro/eeg_loader.py b/synapdrive_ai/neuro/eeg_loader.py
index b99fee2..9954f73 100644
--- a/synapdrive_ai/neuro/eeg_loader.py
+++ b/synapdrive_ai/neuro/eeg_loader.py
@@ -51,7 +51,7 @@ def n_samples(self) -> int:
def summary(self) -> str:
return (
f"EEGRecording: {self.n_channels} channels × {self.n_samples} samples "
- f"@ {self.sampling_rate} Hz ({self.duration_s:.1f}s) — {self.source_file}"
+ f"@ {self.sampling_rate} Hz ({self.duration_s:.1f}s): {self.source_file}"
)
@@ -143,7 +143,8 @@ def _field(offset: int, width: int, sig_idx: int) -> str:
data_raw = raw[data_start:]
total_per_record = sum(n_samples_per_record)
- records_available = len(data_raw) // (total_per_record * 2)
+ sample_width = 3 if path.suffix.lower() == ".bdf" else 2
+ records_available = len(data_raw) // (total_per_record * sample_width)
if n_records < 0:
n_records = records_available
n_records = min(n_records, records_available)
@@ -153,10 +154,21 @@ def _field(offset: int, width: int, sig_idx: int) -> str:
for _ in range(n_records):
for sig_idx in range(n_signals):
ns = n_samples_per_record[sig_idx]
- chunk = data_raw[offset_bytes : offset_bytes + ns * 2]
- samples = struct.unpack(f"<{ns}h", chunk)
+ width = ns * sample_width
+ chunk = data_raw[offset_bytes : offset_bytes + width]
+ if sample_width == 2:
+ samples = struct.unpack(f"<{ns}h", chunk)
+ else:
+ raw24 = np.frombuffer(chunk, dtype=np.uint8).reshape(-1, 3)
+ values = (
+ raw24[:, 0].astype(np.int32)
+ | (raw24[:, 1].astype(np.int32) << 8)
+ | (raw24[:, 2].astype(np.int32) << 16)
+ )
+ values[values >= 0x800000] -= 0x1000000
+ samples = values.tolist()
channels_raw[sig_idx].extend(samples)
- offset_bytes += ns * 2
+ offset_bytes += width
data = np.array(
[
diff --git a/synapdrive_ai/neuro/errp.py b/synapdrive_ai/neuro/errp.py
new file mode 100644
index 0000000..7b98799
--- /dev/null
+++ b/synapdrive_ai/neuro/errp.py
@@ -0,0 +1,109 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Iterable, Sequence
+
+import numpy as np
+
+
+@dataclass(frozen=True)
+class ErrPFeatures:
+ early_mean: float
+ negative_peak: float
+ late_mean: float
+ positive_peak: float
+
+ def as_array(self) -> np.ndarray:
+ return np.array(
+ [
+ self.early_mean,
+ self.negative_peak,
+ self.late_mean,
+ self.positive_peak,
+ ],
+ dtype=float,
+ )
+
+
+class ErrPFeatureExtractor:
+ """Extracts event-related potential features from an event-locked epoch.
+
+ Windows are relative to event onset. This is a generic feature extractor, not a
+ validated ErrP detector. Classification requires user-supplied labeled calibration.
+ """
+
+ def __init__(self, sampling_rate: float, event_index: int = 0) -> None:
+ self.sampling_rate = float(sampling_rate)
+ self.event_index = int(event_index)
+
+ def extract(self, epoch: Sequence[float] | np.ndarray) -> ErrPFeatures:
+ x = np.asarray(epoch, dtype=float)
+ if x.ndim != 1:
+ raise ValueError("epoch must be one-dimensional")
+ if len(x) < max(4, self.event_index + int(0.60 * self.sampling_rate)):
+ raise ValueError("epoch is too short for configured ErrP windows")
+ early = self._window(x, 0.10, 0.25)
+ negative = self._window(x, 0.20, 0.35)
+ late = self._window(x, 0.35, 0.60)
+ return ErrPFeatures(
+ early_mean=float(np.mean(early)),
+ negative_peak=float(np.min(negative)),
+ late_mean=float(np.mean(late)),
+ positive_peak=float(np.max(late)),
+ )
+
+ def _window(self, x: np.ndarray, start_s: float, end_s: float) -> np.ndarray:
+ start = self.event_index + int(start_s * self.sampling_rate)
+ end = self.event_index + int(end_s * self.sampling_rate)
+ return x[start:max(start + 1, end)]
+
+
+class ErrPLDAClassifier:
+ """Small, inspectable LDA classifier trained only on supplied labeled epochs."""
+
+ def __init__(self, regularization: float = 1e-5) -> None:
+ self.regularization = float(regularization)
+ self.w_: np.ndarray | None = None
+ self.b_: float | None = None
+
+ @property
+ def fitted(self) -> bool:
+ return self.w_ is not None and self.b_ is not None
+
+ def fit(
+ self,
+ features: Iterable[Iterable[float]],
+ labels: Iterable[int],
+ ) -> "ErrPLDAClassifier":
+ x = np.asarray(list(features), dtype=float)
+ y = np.asarray(list(labels), dtype=int)
+ if x.ndim != 2 or len(x) != len(y) or len(x) < 4:
+ raise ValueError("features and labels must contain at least four aligned samples")
+ if set(np.unique(y)) != {0, 1}:
+ raise ValueError("labels must contain both classes 0 and 1")
+ x0, x1 = x[y == 0], x[y == 1]
+ m0, m1 = np.mean(x0, axis=0), np.mean(x1, axis=0)
+ c0 = np.cov(x0, rowvar=False) if len(x0) > 1 else np.eye(x.shape[1])
+ c1 = np.cov(x1, rowvar=False) if len(x1) > 1 else np.eye(x.shape[1])
+ numerator = (
+ (len(x0) - 1) * np.atleast_2d(c0)
+ + (len(x1) - 1) * np.atleast_2d(c1)
+ )
+ cov = numerator / max(len(x) - 2, 1)
+ cov = cov + np.eye(x.shape[1]) * self.regularization
+ inv = np.linalg.pinv(cov)
+ self.w_ = inv @ (m1 - m0)
+ p0, p1 = len(x0) / len(x), len(x1) / len(x)
+ self.b_ = float(-0.5 * (m1 + m0) @ self.w_ + np.log(max(p1, 1e-12) / max(p0, 1e-12)))
+ return self
+
+ def predict_proba(self, feature: Iterable[float]) -> float:
+ if not self.fitted:
+ raise RuntimeError("classifier has not been fit")
+ x = np.asarray(list(feature), dtype=float)
+ assert self.w_ is not None and self.b_ is not None
+ z = float(np.clip(x @ self.w_ + self.b_, -60, 60))
+ return float(1.0 / (1.0 + np.exp(-z)))
+
+ def predict(self, feature: Iterable[float], threshold: float = 0.5) -> int:
+ return int(self.predict_proba(feature) >= threshold)
diff --git a/synapdrive_ai/neuro/fusion.py b/synapdrive_ai/neuro/fusion.py
new file mode 100644
index 0000000..3c799aa
--- /dev/null
+++ b/synapdrive_ai/neuro/fusion.py
@@ -0,0 +1,55 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Dict, Iterable, Mapping
+
+from synapdrive_ai.neuro.uncertainty import estimate_uncertainty, normalize_distribution
+
+
+@dataclass(frozen=True)
+class ModalityEvidence:
+ modality: str
+ probabilities: Mapping[str, float]
+ reliability: float
+
+
+@dataclass(frozen=True)
+class FusionResult:
+ probabilities: Dict[str, float]
+ intent: str
+ confidence: float
+ uncertainty: float
+ weights: Dict[str, float]
+
+
+class WeightedEvidenceFusion:
+ """Reliability-weighted late fusion for independent modality decoders."""
+
+ def fuse(self, evidence: Iterable[ModalityEvidence]) -> FusionResult:
+ items = list(evidence)
+ if not items:
+ raise ValueError("at least one modality is required")
+ labels = sorted({label for item in items for label in item.probabilities})
+ accum = {label: 0.0 for label in labels}
+ weights: Dict[str, float] = {}
+ total_weight = 0.0
+ members = []
+ for item in items:
+ probs = normalize_distribution(item.probabilities)
+ members.append(probs)
+ w = max(0.0, min(1.0, float(item.reliability)))
+ weights[item.modality] = w
+ total_weight += w
+ for label in labels:
+ accum[label] += w * probs.get(label, 0.0)
+ if total_weight <= 0:
+ raise ValueError("at least one modality must have positive reliability")
+ fused = normalize_distribution({k: v / total_weight for k, v in accum.items()})
+ uncertainty = estimate_uncertainty(fused, members)
+ return FusionResult(
+ probabilities=fused,
+ intent=uncertainty.top_label,
+ confidence=uncertainty.top_probability,
+ uncertainty=uncertainty.combined,
+ weights=weights,
+ )
diff --git a/synapdrive_ai/neuro/session_analyzer.py b/synapdrive_ai/neuro/session_analyzer.py
index dfbd0a0..2338294 100644
--- a/synapdrive_ai/neuro/session_analyzer.py
+++ b/synapdrive_ai/neuro/session_analyzer.py
@@ -4,15 +4,17 @@
import time
from dataclasses import asdict, dataclass, field
from pathlib import Path
-from typing import Any, Dict, List, Optional
+from typing import Any, Callable, Dict, List, Mapping, Optional
import numpy as np
-from synapdrive_ai.bci.intent_generator import generate_intent
from synapdrive_ai.neuro.band_analyzer import BandPowerAnalyzer, BandPowerResult
from synapdrive_ai.neuro.eeg_loader import EEGRecording
+from synapdrive_ai.neuro.signal_quality import SignalQualityAnalyzer
from synapdrive_ai.pipeline import SynapDrivePipeline
+EpochDecoder = Callable[[np.ndarray, Mapping[str, Any]], Mapping[str, Any]]
+
@dataclass
class EpochResult:
@@ -50,7 +52,7 @@ class SessionReport:
def summary(self) -> str:
lines = [
- f"Session Analysis — {self.source_file}",
+ f"Session Analysis: {self.source_file}",
f" Channel: {self.channel}",
f" Window: {self.window_s}s step: {self.step_s}s",
f" Epochs: {self.n_epochs}",
@@ -103,7 +105,11 @@ def save_csv(self, path: str | Path) -> None:
class SessionAnalyzer:
"""
- Runs a sliding-window analysis of an EEGRecording through the pipeline.
+ Sliding-window EEG feature analysis with an explicit decoder boundary.
+
+ Without ``decoder``, spectral features are analysis-only and are deliberately blocked
+ from actuation. A supplied decoder must return an intent packet containing at least
+ ``intent`` and ``confidence``; its outputs still pass through runtime governance.
"""
def __init__(
@@ -114,12 +120,14 @@ def __init__(
image_label: Optional[str] = None,
simulate_delay: bool = False,
pipeline: Optional[SynapDrivePipeline] = None,
+ decoder: Optional[EpochDecoder] = None,
) -> None:
self.channel = channel
self.window_s = float(window_s)
self.step_s = float(step_s)
self.image_label = image_label
self._pipe = pipeline or SynapDrivePipeline(simulate_delay=simulate_delay)
+ self.decoder = decoder
self._band_analyzer: Optional[BandPowerAnalyzer] = None
def run(self, recording: EEGRecording) -> SessionReport:
@@ -146,8 +154,11 @@ def run(self, recording: EEGRecording) -> SessionReport:
t_start = start / sr
t_end = (start + window_samples) / sr
+ quality_result = SignalQualityAnalyzer(sr).analyze(epoch_signal)
band_result = self._band_analyzer.analyze(epoch_signal)
- pipeline_out = self._run_epoch_through_pipeline(band_result)
+ pipeline_out = self._run_epoch_through_pipeline(
+ epoch_signal, sr, band_result, quality_result.score, quality_result.state
+ )
intent_out = pipeline_out.get("intent", {}) or {}
eval_out = pipeline_out.get("evaluation", {}) or {}
@@ -175,26 +186,54 @@ def run(self, recording: EEGRecording) -> SessionReport:
return self._build_report(recording, ch_name, epochs)
- def _run_epoch_through_pipeline(self, band_result: BandPowerResult) -> Dict[str, Any]:
- if band_result.intent_class == "motor":
- if band_result.engagement_ratio > 2.0:
- text = "move forward"
- elif band_result.engagement_ratio > 1.2:
- text = "move left"
- else:
- text = "move right"
- elif band_result.intent_class == "cognitive":
- text = "calculate" if band_result.cognitive_ratio > 2.0 else "recall"
+ def _run_epoch_through_pipeline(
+ self,
+ epoch_signal: np.ndarray,
+ sampling_rate: float,
+ band_result: BandPowerResult,
+ signal_quality: float = 1.0,
+ quality_state: str = "good",
+ ) -> Dict[str, Any]:
+ if self.decoder is None:
+ packet: Dict[str, Any] = {
+ "intent": "unknown",
+ "confidence": 0.0,
+ "source": f"eeg_band_features/{band_result.intent_class}",
+ "memory_context": [],
+ "band_power": band_result.relative,
+ "engagement_ratio": band_result.engagement_ratio,
+ "cognitive_ratio": band_result.cognitive_ratio,
+ "heuristic_score": band_result.confidence,
+ "confidence_semantics": "no-decoder",
+ "inference_authority": "analysis-only-band-features",
+ "analysis_only": True,
+ "neural_decode_performed": False,
+ "signal_quality": float(signal_quality),
+ "signal_quality_state": quality_state,
+ }
else:
- text = "unknown"
-
- base_packet = dict(generate_intent(text))
- base_packet["confidence"] = band_result.confidence
- base_packet["source"] = f"eeg_band/{band_result.intent_class}"
- base_packet["band_power"] = band_result.relative
- base_packet["engagement_ratio"] = band_result.engagement_ratio
+ epoch_data = np.asarray(epoch_signal, dtype=float).reshape(1, -1)
+ decoder_metadata = {
+ "sampling_rate": float(sampling_rate),
+ "n_channels": 1,
+ "n_samples": int(epoch_data.shape[1]),
+ "signal_quality": float(signal_quality),
+ "signal_quality_state": quality_state,
+ }
+ packet = dict(self.decoder(epoch_data.copy(), decoder_metadata))
+ packet.setdefault("intent", "unknown")
+ packet.setdefault("confidence", 0.0)
+ packet.setdefault("source", "external-epoch-decoder")
+ packet.setdefault("memory_context", [])
+ packet.setdefault("inference_authority", "external-decoder")
+ packet.setdefault("neural_decode_performed", True)
+ packet["signal_quality"] = float(signal_quality)
+ packet["signal_quality_state"] = quality_state
+ packet["band_power"] = band_result.relative
+ packet["spectral_feature_class"] = band_result.intent_class
+ packet["spectral_heuristic_score"] = band_result.confidence
- return self._pipe.run_intent_packet(base_packet, image_label=self.image_label)
+ return self._pipe.run_intent_packet(packet, image_label=self.image_label)
def _build_report(
self,
diff --git a/synapdrive_ai/neuro/signal_quality.py b/synapdrive_ai/neuro/signal_quality.py
new file mode 100644
index 0000000..f2365a0
--- /dev/null
+++ b/synapdrive_ai/neuro/signal_quality.py
@@ -0,0 +1,138 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Dict, Tuple
+
+import numpy as np
+
+
+@dataclass(frozen=True)
+class SignalQualityReport:
+ score: float
+ state: str
+ finite_fraction: float
+ flatline_fraction: float
+ clipping_fraction: float
+ dropout_fraction: float
+ line_noise_ratio: float
+ issues: Tuple[str, ...] = field(default_factory=tuple)
+
+ def to_dict(self) -> Dict[str, float | str | list[str]]:
+ return {
+ "score": self.score,
+ "state": self.state,
+ "finite_fraction": self.finite_fraction,
+ "flatline_fraction": self.flatline_fraction,
+ "clipping_fraction": self.clipping_fraction,
+ "dropout_fraction": self.dropout_fraction,
+ "line_noise_ratio": self.line_noise_ratio,
+ "issues": list(self.issues),
+ }
+
+
+class SignalQualityAnalyzer:
+ """Deterministic quality checks for one or more EEG-style channels.
+
+ The score is an engineering heuristic for gating analysis. It is not a clinical
+ signal-quality index and is deliberately exposed as separate component metrics.
+ """
+
+ def __init__(self, sampling_rate: float, line_frequency: float = 60.0) -> None:
+ if sampling_rate <= 0:
+ raise ValueError("sampling_rate must be positive")
+ self.sampling_rate = float(sampling_rate)
+ self.line_frequency = float(line_frequency)
+
+ def analyze(self, signal: np.ndarray) -> SignalQualityReport:
+ x = np.asarray(signal, dtype=float)
+ if x.ndim == 1:
+ x = x[None, :]
+ if x.size == 0 or x.shape[1] < 4:
+ return SignalQualityReport(
+ 0.0,
+ "invalid",
+ 0.0,
+ 1.0,
+ 0.0,
+ 1.0,
+ 0.0,
+ ("insufficient-samples",),
+ )
+
+ finite = np.isfinite(x)
+ finite_fraction = float(np.mean(finite))
+ safe = np.where(finite, x, 0.0)
+ centered = safe - np.mean(safe, axis=1, keepdims=True)
+ scale = np.std(centered, axis=1, keepdims=True)
+
+ # Flatline is assessed from near-zero first differences relative to channel scale.
+ diffs = np.abs(np.diff(safe, axis=1))
+ eps = np.maximum(scale[:, :1] * 1e-5, 1e-12)
+ flatline_fraction = float(np.mean(diffs <= eps))
+
+ # Dropout catches exact-zero runs, a common stream failure representation.
+ dropout_fraction = float(np.mean(np.isclose(safe, 0.0, atol=1e-12)))
+
+ # Clipping catches excessive samples pinned at the observed extrema.
+ clipping_parts = []
+ for ch in safe:
+ lo, hi = float(np.min(ch)), float(np.max(ch))
+ if np.isclose(lo, hi):
+ clipping_parts.append(1.0)
+ else:
+ clipping_parts.append(float(np.mean(np.isclose(ch, lo) | np.isclose(ch, hi))))
+ clipping_fraction = float(np.mean(clipping_parts))
+
+ line_noise_ratio = self._line_noise_ratio(centered)
+
+ issues: list[str] = []
+ if finite_fraction < 0.999:
+ issues.append("non-finite-samples")
+ if flatline_fraction > 0.20:
+ issues.append("flatline")
+ if dropout_fraction > 0.20:
+ issues.append("dropout")
+ if clipping_fraction > 0.08:
+ issues.append("clipping")
+ if line_noise_ratio > 0.35:
+ issues.append("line-noise")
+
+ penalty = (
+ 0.45 * (1.0 - finite_fraction)
+ + 0.25 * min(1.0, flatline_fraction / 0.30)
+ + 0.15 * min(1.0, dropout_fraction / 0.30)
+ + 0.10 * min(1.0, clipping_fraction / 0.12)
+ + 0.05 * min(1.0, line_noise_ratio / 0.50)
+ )
+ score = float(np.clip(1.0 - penalty, 0.0, 1.0))
+ if score >= 0.80:
+ state = "good"
+ elif score >= 0.55:
+ state = "degraded"
+ elif score >= 0.35:
+ state = "unreliable"
+ else:
+ state = "invalid"
+ return SignalQualityReport(
+ round(score, 6),
+ state,
+ round(finite_fraction, 6),
+ round(flatline_fraction, 6),
+ round(clipping_fraction, 6),
+ round(dropout_fraction, 6),
+ round(line_noise_ratio, 6),
+ tuple(issues),
+ )
+
+ def _line_noise_ratio(self, centered: np.ndarray) -> float:
+ n = centered.shape[1]
+ if n < 8:
+ return 0.0
+ freqs = np.fft.rfftfreq(n, d=1.0 / self.sampling_rate)
+ power = np.mean(np.abs(np.fft.rfft(centered, axis=1)) ** 2, axis=0)
+ total = float(np.sum(power[(freqs >= 1.0) & (freqs <= min(100.0, self.sampling_rate / 2))]))
+ if total <= 1e-18:
+ return 0.0
+ width = max(0.75, self.sampling_rate / n * 2.0)
+ mask = np.abs(freqs - self.line_frequency) <= width
+ return float(np.clip(np.sum(power[mask]) / total, 0.0, 1.0))
diff --git a/synapdrive_ai/neuro/task_planner.py b/synapdrive_ai/neuro/task_planner.py
index 5d14b2c..98dc9d0 100644
--- a/synapdrive_ai/neuro/task_planner.py
+++ b/synapdrive_ai/neuro/task_planner.py
@@ -140,6 +140,9 @@ def execute(self, plan: TaskPlan) -> PlanTrace:
def _execute_step(self, idx: int, step: TaskStep) -> StepTrace:
base_packet = generate_intent(step.intent_text)
+ # A task-specific threshold is a stricter pre-action requirement. The canonical
+ # runtime consumes this field before the router is reachable.
+ base_packet["required_confidence"] = step.min_confidence
out = self._pipe.run_intent_packet(base_packet, image_label=step.image_label)
intent_out = out.get("intent", {}) or {}
@@ -150,7 +153,7 @@ def _execute_step(self, idx: int, step: TaskStep) -> StepTrace:
eval_score = float(eval_out.get("score", 0.0))
fallback_applied: Optional[FallbackPolicy] = None
- if pipeline_status == "success" and confidence < step.min_confidence:
+ if pipeline_status == "blocked" and "confidence-below-policy" in str(block_reason):
fallback_applied = step.fallback
if step.fallback == "freeze":
pipeline_status = "deferred"
diff --git a/synapdrive_ai/neuro/uncertainty.py b/synapdrive_ai/neuro/uncertainty.py
new file mode 100644
index 0000000..2968a18
--- /dev/null
+++ b/synapdrive_ai/neuro/uncertainty.py
@@ -0,0 +1,235 @@
+from __future__ import annotations
+
+import math
+from dataclasses import dataclass
+from typing import Dict, Iterable, Mapping, Sequence
+
+import numpy as np
+
+
+@dataclass(frozen=True)
+class UncertaintyEstimate:
+ entropy: float
+ margin_uncertainty: float
+ disagreement: float
+ combined: float
+ top_label: str
+ top_probability: float
+
+ def to_dict(self) -> Dict[str, float | str]:
+ return {
+ "entropy": self.entropy,
+ "margin_uncertainty": self.margin_uncertainty,
+ "disagreement": self.disagreement,
+ "combined": self.combined,
+ "top_label": self.top_label,
+ "top_probability": self.top_probability,
+ }
+
+
+def _numeric_value(value: object, *, field: str) -> float:
+ if isinstance(value, (int, float, str)):
+ return float(value)
+ raise TypeError(f"{field} must be numeric")
+
+
+def normalize_distribution(values: Mapping[str, float]) -> Dict[str, float]:
+ clean = {str(k): max(0.0, float(v)) for k, v in values.items()}
+ total = sum(clean.values())
+ if total <= 0:
+ if not clean:
+ return {"unknown": 1.0}
+ uniform = 1.0 / len(clean)
+ return {k: uniform for k in clean}
+ return {k: v / total for k, v in clean.items()}
+
+
+def packet_distribution(packet: Mapping[str, object]) -> Dict[str, float]:
+ provided = packet.get("probabilities")
+ if isinstance(provided, Mapping) and provided:
+ parsed = {
+ str(k): _numeric_value(v, field=f"probabilities[{k!r}]")
+ for k, v in provided.items()
+ }
+ return normalize_distribution(parsed)
+
+ intent = str(packet.get("intent") or "unknown")
+ confidence = max(
+ 0.0,
+ min(
+ 1.0,
+ _numeric_value(
+ packet.get("confidence", 0.0),
+ field="confidence",
+ ),
+ ),
+ )
+
+ if intent == "unknown":
+ return {
+ "unknown": max(confidence, 0.5),
+ "other": 1.0 - max(confidence, 0.5),
+ }
+
+ return normalize_distribution(
+ {
+ intent: confidence,
+ "unknown": 1.0 - confidence,
+ }
+ )
+
+
+def estimate_uncertainty(
+ probabilities: Mapping[str, float],
+ ensemble_probabilities: Sequence[Mapping[str, float]] | None = None,
+) -> UncertaintyEstimate:
+ probs = normalize_distribution(probabilities)
+ labels = sorted(probs)
+ arr = np.array([probs[k] for k in labels], dtype=float)
+
+ if len(arr) <= 1:
+ entropy = 0.0
+ margin_u = 0.0
+ else:
+ raw_entropy = -float(
+ np.sum(arr * np.log(np.clip(arr, 1e-12, 1.0)))
+ )
+ entropy = raw_entropy / math.log(len(arr))
+ sorted_probs = np.sort(arr)[::-1]
+ margin_u = 1.0 - float(
+ sorted_probs[0] - sorted_probs[1]
+ )
+
+ disagreement = 0.0
+ if ensemble_probabilities and len(ensemble_probabilities) > 1:
+ union = sorted(
+ {
+ k
+ for member in ensemble_probabilities
+ for k in member
+ }
+ )
+ matrix = np.array(
+ [
+ [
+ normalize_distribution(member).get(k, 0.0)
+ for k in union
+ ]
+ for member in ensemble_probabilities
+ ],
+ dtype=float,
+ )
+ disagreement = float(
+ np.clip(
+ np.mean(np.std(matrix, axis=0)) * 2.0,
+ 0.0,
+ 1.0,
+ )
+ )
+
+ combined = float(
+ np.clip(
+ 0.55 * entropy
+ + 0.30 * margin_u
+ + 0.15 * disagreement,
+ 0.0,
+ 1.0,
+ )
+ )
+
+ top_label = max(
+ probs.items(),
+ key=lambda item: item[1],
+ )[0]
+
+ return UncertaintyEstimate(
+ entropy=round(entropy, 6),
+ margin_uncertainty=round(margin_u, 6),
+ disagreement=round(disagreement, 6),
+ combined=round(combined, 6),
+ top_label=top_label,
+ top_probability=round(probs[top_label], 6),
+ )
+
+
+class ReliabilityCalibrator:
+ """Histogram reliability calibrator with deterministic binning.
+
+ Fit using model confidence and binary correctness. Transform maps raw
+ confidence to observed empirical accuracy in its bin. No model claims
+ are made before fit().
+ """
+
+ def __init__(self, n_bins: int = 10) -> None:
+ if n_bins < 2:
+ raise ValueError("n_bins must be >= 2")
+ self.n_bins = int(n_bins)
+ self._rates: np.ndarray | None = None
+ self._counts: np.ndarray | None = None
+
+ @property
+ def fitted(self) -> bool:
+ return self._rates is not None
+
+ def fit(
+ self,
+ confidence: Iterable[float],
+ correct: Iterable[bool],
+ ) -> "ReliabilityCalibrator":
+ conf = np.clip(
+ np.asarray(list(confidence), dtype=float),
+ 0.0,
+ 1.0,
+ )
+ ok = np.asarray(list(correct), dtype=float)
+
+ if conf.size == 0 or conf.size != ok.size:
+ raise ValueError(
+ "confidence and correct must be non-empty and equal length"
+ )
+
+ indices = np.minimum(
+ (conf * self.n_bins).astype(int),
+ self.n_bins - 1,
+ )
+
+ rates = np.zeros(self.n_bins, dtype=float)
+ counts = np.zeros(self.n_bins, dtype=int)
+ global_rate = float(np.mean(ok))
+
+ for idx in range(self.n_bins):
+ mask = indices == idx
+ counts[idx] = int(np.sum(mask))
+ rates[idx] = (
+ float(np.mean(ok[mask]))
+ if counts[idx]
+ else global_rate
+ )
+
+ self._rates = rates
+ self._counts = counts
+ return self
+
+ def transform(self, confidence: float) -> float:
+ if self._rates is None:
+ raise RuntimeError("calibrator has not been fit")
+
+ value = float(
+ np.clip(
+ confidence,
+ 0.0,
+ 1.0,
+ )
+ )
+ idx = min(
+ int(value * self.n_bins),
+ self.n_bins - 1,
+ )
+
+ return float(
+ np.clip(
+ self._rates[idx],
+ 0.0,
+ 1.0,
+ )
+ )
\ No newline at end of file
diff --git a/synapdrive_ai/pipeline.py b/synapdrive_ai/pipeline.py
index 233a1c7..1e6efc6 100644
--- a/synapdrive_ai/pipeline.py
+++ b/synapdrive_ai/pipeline.py
@@ -1,78 +1,101 @@
from __future__ import annotations
-import random
-from typing import Any, Dict, Optional
+from typing import Any, Dict, Mapping, Optional
from synapdrive_ai.action.decision_router import DecisionRouter
from synapdrive_ai.agi.cognitive_optimizer import CognitiveOptimizer
-from synapdrive_ai.agi.core_reasoning import AGICoreReasoner
+from synapdrive_ai.agi.core_reasoning import LabeledSignalMapper
from synapdrive_ai.agi.meta_evaluator import MetaEvaluator
from synapdrive_ai.assurance import AssuranceMonitor
+from synapdrive_ai.assurance.evidence_ledger import SessionEvidenceLedger
from synapdrive_ai.bci.intent_generator import generate_intent
from synapdrive_ai.bci.signal_simulator import BrainSignalSimulator
+from synapdrive_ai.cognition.planner import CandidateAction, CounterfactualPlanner
+from synapdrive_ai.cognition.shared_autonomy import SharedAutonomyArbiter
from synapdrive_ai.memory.episodic_memory import EpisodicMemory
+from synapdrive_ai.memory.evidence_memory import EvidenceMemory
+from synapdrive_ai.neuro.uncertainty import estimate_uncertainty, packet_distribution
+from synapdrive_ai.runtime.governed_runtime import GovernedRuntime
+from synapdrive_ai.runtime.reality import RealityReconciler
from synapdrive_ai.safety.safety_guard import SafetyGuard
from synapdrive_ai.vision.visual_inference import VisualInferenceEngine
class SynapDrivePipeline:
- """
- Canonical end-to-end simulation pipeline.
+ """Canonical governed simulation pipeline.
- Wiring: input → intent → optimizer (memory + vision) → safety → actuation → evaluation
+ Flow:
+ input -> intent -> context -> uncertainty -> policy/permission/world assessment
+ -> legacy safety checks -> simulated actuation or fail-closed abstention
+ -> reality reconciliation -> evidence-gated memory -> assurance + hash-chain evidence
- Args:
- simulate_delay: If True, actuation sleeps to emulate real timing.
- Set False for tests, replay, and the Flask dashboard.
+ No element of this class constitutes medical-device or physical-actuation validation.
"""
- def __init__(self, simulate_delay: bool = True) -> None:
+ def __init__(self, simulate_delay: bool = True, runtime: GovernedRuntime | None = None) -> None:
self.simulator = BrainSignalSimulator()
- self.reasoner = AGICoreReasoner()
-
+ self.reasoner = LabeledSignalMapper()
self.memory = EpisodicMemory()
+ self.evidence_memory = EvidenceMemory()
self.visual = VisualInferenceEngine()
-
self.optimizer = CognitiveOptimizer()
- # Share the same memory/vision instances so the optimizer sees live state
self.optimizer.memory = self.memory
self.optimizer.visual = self.visual
-
self.guard = SafetyGuard()
self.router = DecisionRouter(simulate_delay=simulate_delay)
self.evaluator = MetaEvaluator()
self.assurance = AssuranceMonitor()
-
- # ------------------------------------------------------------------
- # Public entrypoints
- # ------------------------------------------------------------------
+ self.runtime = runtime or GovernedRuntime()
+ self.reconciler = RealityReconciler()
+ self.evidence = SessionEvidenceLedger()
def run_text_command(
- self, command_text: str, image_label: Optional[str] = None
+ self,
+ command_text: str,
+ image_label: Optional[str] = None,
+ feedback: Mapping[str, Any] | None = None,
) -> Dict[str, Any]:
- """Text path: decoded intent or operator input (e.g. 'move left', 'stop')."""
intent_packet = generate_intent(command_text)
- return self.run_intent_packet(intent_packet, image_label=image_label)
+ return self.run_intent_packet(intent_packet, image_label=image_label, feedback=feedback)
def run_signal_event(
- self, label: Optional[str] = None, image_label: Optional[str] = None
+ self,
+ label: Optional[str] = None,
+ image_label: Optional[str] = None,
+ feedback: Mapping[str, Any] | None = None,
) -> Dict[str, Any]:
- """Signal path: simulated EEG-like waveform event."""
- label = label or random.choice(
- ["left_arm", "right_arm", "walk", "stop", "calculate", "recall", "explore"]
- )
+ label = label or self.simulator.random_label()
signal = self._generate_signal_for_label(label)
intent_packet = self.reasoner.receive_signal(label, signal)
- return self.run_intent_packet(intent_packet, image_label=image_label)
+ return self.run_intent_packet(intent_packet, image_label=image_label, feedback=feedback)
def run_intent_packet(
- self, intent_packet: Dict[str, Any], image_label: Optional[str] = None
+ self,
+ intent_packet: Dict[str, Any],
+ image_label: Optional[str] = None,
+ feedback: Mapping[str, Any] | None = None,
) -> Dict[str, Any]:
+ return self._run_common(intent_packet, image_label=image_label, feedback=feedback)
+
+ def plan_shared_action(
+ self,
+ *,
+ user_intent: str,
+ user_confidence: float,
+ machine_candidates: list[CandidateAction],
+ goal,
+ ):
+ """Rank bounded machine alternatives against the runtime world model.
+
+ This method proposes only. It does not actuate the selected action.
"""
- Primary entrypoint for integrations and replay.
- Accepts a pre-built intent packet and runs it through the full pipeline.
- """
- return self._run_common(intent_packet, image_label=image_label)
+ arbiter = SharedAutonomyArbiter(CounterfactualPlanner(self.runtime.world_model))
+ return arbiter.propose(
+ user_intent=user_intent,
+ user_confidence=user_confidence,
+ machine_candidates=machine_candidates,
+ goal=goal,
+ )
def get_action_log(self):
return self.router.get_action_log()
@@ -81,11 +104,14 @@ def get_assurance_log(self):
return [receipt.to_dict() for receipt in self.assurance.history()]
def get_assurance_report(self) -> Dict[str, Any]:
- return self.assurance.health_report()
-
- # ------------------------------------------------------------------
- # Internals
- # ------------------------------------------------------------------
+ report = dict(self.assurance.health_report())
+ report["evidence_chain_entries"] = len(self.evidence.chain.entries())
+ report["evidence_chain_valid"] = self.evidence.verify(self.evidence.chain.entries())
+ report["validated_memories"] = sum(
+ 1 for record in self.evidence_memory.all() if record.state == "validated"
+ )
+ report["quarantined_memories"] = len(self.evidence_memory.quarantined())
+ return report
def _generate_signal_for_label(self, label: str):
patterns = {
@@ -102,63 +128,155 @@ def _generate_signal_for_label(self, label: str):
return self.simulator.generate_waveform(patterns[label])
def _run_common(
- self, intent_packet: Dict[str, Any], image_label: Optional[str]
+ self,
+ intent_packet: Dict[str, Any],
+ image_label: Optional[str],
+ feedback: Mapping[str, Any] | None,
) -> Dict[str, Any]:
optimized = self.optimizer.optimize(intent_packet, image_label=image_label)
+ distribution = packet_distribution(optimized)
+ uncertainty = estimate_uncertainty(distribution)
+ optimized["probabilities"] = distribution
+ optimized["uncertainty"] = float(optimized.get("uncertainty", uncertainty.combined))
+ optimized["uncertainty_detail"] = uncertainty.to_dict()
+
+ runtime_decision = self.runtime.assess(optimized)
+ legacy_safe, legacy_reason = self.guard.evaluate_safety(optimized)
+ allowed = runtime_decision.allowed and legacy_safe
+ block_reason = runtime_decision.reason if not runtime_decision.allowed else legacy_reason
- is_safe, reason = self.guard.evaluate_safety(optimized)
- if not is_safe:
+ if not allowed:
result = {
"status": "blocked",
"intent": optimized.get("intent", "unknown"),
"confidence": optimized.get("confidence", 0.0),
"duration": 0.0,
+ "fallback_action": runtime_decision.fallback_action,
}
evaluation = {
"score": 0.0,
- "total_actions": 0,
+ "total_actions": len(self.evaluator.scores),
"avg_score": 0.0,
}
+ reality = self.reconciler.reconcile(
+ predicted_success=0.0,
+ result_packet=result,
+ feedback=feedback,
+ errp_threshold=self.runtime.policy.errp_contradiction_threshold,
+ )
receipt = self.assurance.record_cycle(
intent_packet=optimized,
image_label=image_label,
safety_allowed=False,
- safety_reason=reason,
+ safety_reason=block_reason,
result_packet=result,
evaluation=evaluation,
executed=False,
)
+ evidence_entry = self._record_evidence(
+ optimized, runtime_decision.to_dict(), result, reality.to_dict(), receipt.to_dict()
+ )
return {
"status": "blocked",
- "reason": reason,
+ "reason": block_reason,
"intent": optimized,
"result": result,
"evaluation": evaluation,
+ "runtime": runtime_decision.to_dict(),
+ "reality": reality.to_dict(),
+ "memory": {"state": "not-recorded", "reason": "blocked-before-actuation"},
"assurance": receipt.to_dict(),
+ "evidence": evidence_entry.to_dict(),
}
result = self.router.route(optimized)
+ world_prediction = self.runtime.world_model.predict(
+ str(optimized.get("intent", "unknown"))
+ )
+ if result.get("status") == "success" and world_prediction.feasible:
+ self.runtime.world_model.commit(world_prediction)
+ evaluation = self.evaluator.evaluate(optimized, result)
+ predicted_success = max(
+ 0.0,
+ min(1.0, 1.0 - runtime_decision.assessment.predicted_risk),
+ )
+ reality = self.reconciler.reconcile(
+ predicted_success=predicted_success,
+ result_packet=result,
+ feedback=feedback,
+ errp_threshold=self.runtime.policy.errp_contradiction_threshold,
+ )
- try:
+ # Legacy memory remains for compatibility but only aligned successful outcomes reinforce it.
+ if reality.aligned and result.get("status") == "success":
self.memory.record_episode(optimized, result)
- except Exception:
- pass
- evaluation = self.evaluator.evaluate(optimized, result)
receipt = self.assurance.record_cycle(
intent_packet=optimized,
image_label=image_label,
safety_allowed=True,
- safety_reason=reason,
+ safety_reason=legacy_reason,
result_packet=result,
evaluation=evaluation,
executed=True,
)
-
+ cycle_id = receipt.receipt_id
+ memory_record = self.evidence_memory.record(
+ cycle_id=cycle_id,
+ intent=str(optimized.get("intent", "unknown")),
+ confidence=float(optimized.get("confidence", 0.0)),
+ outcome=str(result.get("status", "unknown")),
+ aligned=reality.aligned and result.get("status") == "success",
+ reason=reality.reason,
+ )
+ evidence_entry = self._record_evidence(
+ optimized, runtime_decision.to_dict(), result, reality.to_dict(), receipt.to_dict()
+ )
return {
"status": result["status"],
"intent": optimized,
"result": result,
"evaluation": evaluation,
+ "runtime": runtime_decision.to_dict(),
+ "reality": reality.to_dict(),
+ "memory": memory_record.to_dict(),
"assurance": receipt.to_dict(),
+ "evidence": evidence_entry.to_dict(),
+ "world_state": dict(self.runtime.world_model.state),
+ }
+
+ def _record_evidence(
+ self,
+ intent: Dict[str, Any],
+ runtime: Dict[str, Any],
+ result: Dict[str, Any],
+ reality: Dict[str, Any],
+ assurance: Dict[str, Any],
+ ):
+ # Deliberately exclude wall-clock timestamps so cycle content remains comparable.
+ payload = {
+ "schema": "synapdrive.cycle-evidence.v2",
+ "intent": {
+ "intent": intent.get("intent"),
+ "confidence": intent.get("confidence"),
+ "source": intent.get("source"),
+ "inference_authority": intent.get("inference_authority", "unspecified"),
+ "confidence_semantics": intent.get("confidence_semantics", "unspecified"),
+ "neural_decode_performed": bool(intent.get("neural_decode_performed", False)),
+ "analysis_only": bool(intent.get("analysis_only", False)),
+ "signal_quality": intent.get("signal_quality", 1.0),
+ "uncertainty": intent.get("uncertainty"),
+ "required_confidence": intent.get("required_confidence"),
+ },
+ "runtime": runtime,
+ "result": {
+ "status": result.get("status"),
+ "intent": result.get("intent"),
+ "confidence": result.get("confidence"),
+ "fallback_action": result.get("fallback_action"),
+ },
+ "reality": reality,
+ "assurance_receipt_id": assurance.get("receipt_id"),
+ "assurance_passed": assurance.get("passed"),
}
+ return self.evidence.record(payload)
diff --git a/synapdrive_ai/runtime/__init__.py b/synapdrive_ai/runtime/__init__.py
new file mode 100644
index 0000000..a099b09
--- /dev/null
+++ b/synapdrive_ai/runtime/__init__.py
@@ -0,0 +1,4 @@
+from .contracts import PreActionDecision, RealityVerdict, RuntimeAssessment
+from .governed_runtime import GovernedRuntime
+
+__all__ = ["GovernedRuntime", "PreActionDecision", "RealityVerdict", "RuntimeAssessment"]
diff --git a/synapdrive_ai/runtime/contracts.py b/synapdrive_ai/runtime/contracts.py
new file mode 100644
index 0000000..effa691
--- /dev/null
+++ b/synapdrive_ai/runtime/contracts.py
@@ -0,0 +1,69 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any, Dict, Tuple
+
+
+@dataclass(frozen=True)
+class RuntimeAssessment:
+ confidence: float
+ required_confidence: float
+ uncertainty: float
+ signal_quality: float
+ predicted_risk: float
+ permission_allowed: bool
+ drift_score: float = 0.0
+ issues: Tuple[str, ...] = field(default_factory=tuple)
+
+ @property
+ def admissible(self) -> bool:
+ return not self.issues
+
+ def to_dict(self) -> Dict[str, Any]:
+ return {
+ "confidence": self.confidence,
+ "required_confidence": self.required_confidence,
+ "uncertainty": self.uncertainty,
+ "signal_quality": self.signal_quality,
+ "predicted_risk": self.predicted_risk,
+ "permission_allowed": self.permission_allowed,
+ "drift_score": self.drift_score,
+ "admissible": self.admissible,
+ "issues": list(self.issues),
+ }
+
+
+@dataclass(frozen=True)
+class PreActionDecision:
+ action: str
+ allowed: bool
+ reason: str
+ fallback_action: str
+ assessment: RuntimeAssessment
+
+ def to_dict(self) -> Dict[str, Any]:
+ return {
+ "action": self.action,
+ "allowed": self.allowed,
+ "reason": self.reason,
+ "fallback_action": self.fallback_action,
+ "assessment": self.assessment.to_dict(),
+ }
+
+
+@dataclass(frozen=True)
+class RealityVerdict:
+ aligned: bool
+ outcome: str
+ reason: str
+ prediction_error: float
+ user_error_probability: float
+
+ def to_dict(self) -> Dict[str, Any]:
+ return {
+ "aligned": self.aligned,
+ "outcome": self.outcome,
+ "reason": self.reason,
+ "prediction_error": self.prediction_error,
+ "user_error_probability": self.user_error_probability,
+ }
diff --git a/synapdrive_ai/runtime/governed_runtime.py b/synapdrive_ai/runtime/governed_runtime.py
new file mode 100644
index 0000000..eec0715
--- /dev/null
+++ b/synapdrive_ai/runtime/governed_runtime.py
@@ -0,0 +1,82 @@
+from __future__ import annotations
+
+from typing import Any, Mapping
+
+from synapdrive_ai.cognition.world_model import WorldModel, default_simulation_world_model
+from synapdrive_ai.governance.permission import PermissionGate
+from synapdrive_ai.governance.policy import RuntimePolicy
+from synapdrive_ai.neuro.uncertainty import estimate_uncertainty, packet_distribution
+from synapdrive_ai.runtime.contracts import PreActionDecision, RuntimeAssessment
+
+
+class GovernedRuntime:
+ """Fail-closed pre-action arbiter for provenance, uncertainty, quality and risk."""
+
+ def __init__(
+ self,
+ policy: RuntimePolicy | None = None,
+ permission_gate: PermissionGate | None = None,
+ world_model: WorldModel | None = None,
+ ) -> None:
+ self.policy = policy or RuntimePolicy()
+ self.permissions = permission_gate or PermissionGate()
+ self.world_model = world_model or default_simulation_world_model()
+
+ def assess(self, packet: Mapping[str, Any]) -> PreActionDecision:
+ action = str(packet.get("intent") or "unknown")
+ confidence = max(0.0, min(1.0, float(packet.get("confidence", 0.0))))
+ requested_min_confidence = max(
+ self.policy.min_confidence,
+ max(0.0, float(packet.get("required_confidence", 0.0))),
+ )
+ uncertainty = packet.get("uncertainty")
+ if uncertainty is None:
+ uncertainty = estimate_uncertainty(packet_distribution(packet)).combined
+ uncertainty = max(0.0, min(1.0, float(uncertainty)))
+ quality = max(0.0, min(1.0, float(packet.get("signal_quality", 1.0))))
+ drift_score = max(0.0, float(packet.get("drift_score", 0.0)))
+ permission = self.permissions.evaluate(action)
+ prediction = self.world_model.predict(action)
+ predicted_risk = max(float(packet.get("predicted_risk", 0.0)), prediction.predicted_risk)
+
+ issues: list[str] = []
+ if bool(packet.get("analysis_only", False)):
+ issues.append("analysis-only-inference")
+ if action == "unknown":
+ issues.append("unknown-intent")
+ if confidence < requested_min_confidence:
+ issues.append("confidence-below-policy")
+ if uncertainty > self.policy.max_uncertainty:
+ issues.append("uncertainty-above-policy")
+ if quality < self.policy.min_signal_quality:
+ issues.append("signal-quality-below-policy")
+ if predicted_risk > self.policy.max_predicted_risk:
+ issues.append("predicted-risk-above-policy")
+ if drift_score > self.policy.max_drift_score:
+ issues.append("decoder-drift-above-policy")
+ if not permission.allowed:
+ issues.append("permission-denied")
+ if not prediction.feasible:
+ issues.append("world-model-precondition-failed")
+
+ # Stable ordering with no duplicate issue labels keeps replay evidence deterministic.
+ issues = list(dict.fromkeys(issues))
+ assessment = RuntimeAssessment(
+ round(confidence, 6),
+ round(requested_min_confidence, 6),
+ round(uncertainty, 6),
+ round(quality, 6),
+ round(predicted_risk, 6),
+ permission.allowed,
+ round(drift_score, 6),
+ tuple(issues),
+ )
+ allowed = not issues
+ reason = "runtime policy passed" if allowed else "; ".join(issues)
+ return PreActionDecision(
+ action,
+ allowed,
+ reason,
+ self.policy.safe_fallback_action,
+ assessment,
+ )
diff --git a/synapdrive_ai/runtime/reality.py b/synapdrive_ai/runtime/reality.py
new file mode 100644
index 0000000..59c9deb
--- /dev/null
+++ b/synapdrive_ai/runtime/reality.py
@@ -0,0 +1,54 @@
+from __future__ import annotations
+
+from typing import Any, Mapping
+
+from synapdrive_ai.runtime.contracts import RealityVerdict
+
+
+class RealityReconciler:
+ """Compare predicted software outcome with result and user-error feedback."""
+
+ def reconcile(
+ self,
+ *,
+ predicted_success: float,
+ result_packet: Mapping[str, Any],
+ feedback: Mapping[str, Any] | None = None,
+ errp_threshold: float = 0.70,
+ ) -> RealityVerdict:
+ observed_success = str(result_packet.get("status", "")) in {
+ "success",
+ "executed",
+ }
+ predicted_success = max(0.0, min(1.0, float(predicted_success)))
+ error_probability = 0.0
+ if feedback:
+ raw_probability = float(feedback.get("errp_probability", 0.0))
+ error_probability = max(0.0, min(1.0, raw_probability))
+ explicit = feedback.get("accepted")
+ if explicit is False:
+ error_probability = 1.0
+ prediction_error = abs(predicted_success - float(observed_success))
+ if error_probability >= errp_threshold:
+ return RealityVerdict(
+ False,
+ "contradicted",
+ "user/error-potential feedback contradicts action",
+ round(prediction_error, 6),
+ error_probability,
+ )
+ if predicted_success >= 0.5 and not observed_success:
+ return RealityVerdict(
+ False,
+ "failed",
+ "predicted success but execution did not succeed",
+ round(prediction_error, 6),
+ error_probability,
+ )
+ return RealityVerdict(
+ True,
+ "aligned",
+ "observed outcome is consistent with current prediction",
+ round(prediction_error, 6),
+ error_probability,
+ )
diff --git a/synapdrive_ai/runtime/shadow.py b/synapdrive_ai/runtime/shadow.py
new file mode 100644
index 0000000..40a6e2f
--- /dev/null
+++ b/synapdrive_ai/runtime/shadow.py
@@ -0,0 +1,34 @@
+from __future__ import annotations
+
+from dataclasses import asdict, dataclass
+from typing import Any, Callable, Dict, List
+
+
+@dataclass(frozen=True)
+class ShadowComparison:
+ trusted_action: str
+ shadow_action: str
+ agreement: bool
+ shadow_metadata: Dict[str, Any]
+
+ def to_dict(self) -> Dict[str, Any]:
+ return asdict(self)
+
+
+class ShadowController:
+ """Evaluates an experimental policy without granting it actuation authority."""
+
+ def __init__(self, policy: Callable[[Dict[str, Any]], tuple[str, Dict[str, Any]]]) -> None:
+ self.policy = policy
+ self.history: List[ShadowComparison] = []
+
+ def evaluate(self, trusted_action: str, context: Dict[str, Any]) -> ShadowComparison:
+ action, metadata = self.policy(dict(context))
+ comparison = ShadowComparison(
+ trusted_action=trusted_action,
+ shadow_action=str(action),
+ agreement=str(action) == str(trusted_action),
+ shadow_metadata=dict(metadata),
+ )
+ self.history.append(comparison)
+ return comparison
diff --git a/synapdrive_ai/runtime/simplex.py b/synapdrive_ai/runtime/simplex.py
new file mode 100644
index 0000000..0f8ebda
--- /dev/null
+++ b/synapdrive_ai/runtime/simplex.py
@@ -0,0 +1,29 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any, Callable, Dict
+
+from synapdrive_ai.runtime.contracts import PreActionDecision
+
+
+@dataclass(frozen=True)
+class SimplexExecution:
+ controller: str
+ result: Dict[str, Any]
+
+
+class SimplexController:
+ """Routes to advanced control only after the runtime monitor admits the action."""
+
+ def __init__(
+ self,
+ advanced_controller: Callable[[Dict[str, Any]], Dict[str, Any]],
+ safe_controller: Callable[[str], Dict[str, Any]],
+ ) -> None:
+ self.advanced_controller = advanced_controller
+ self.safe_controller = safe_controller
+
+ def execute(self, packet: Dict[str, Any], decision: PreActionDecision) -> SimplexExecution:
+ if decision.allowed:
+ return SimplexExecution("advanced", self.advanced_controller(packet))
+ return SimplexExecution("reversionary", self.safe_controller(decision.fallback_action))
diff --git a/synapdrive_ai/safety/__init__.py b/synapdrive_ai/safety/__init__.py
new file mode 100644
index 0000000..503252a
--- /dev/null
+++ b/synapdrive_ai/safety/__init__.py
@@ -0,0 +1 @@
+"""Compatibility safety helpers; governed policy lives in the runtime layer."""
diff --git a/synapdrive_ai/stress/__init__.py b/synapdrive_ai/stress/__init__.py
new file mode 100644
index 0000000..9496584
--- /dev/null
+++ b/synapdrive_ai/stress/__init__.py
@@ -0,0 +1,4 @@
+from .campaign import CampaignReport, StressCampaign
+from .faults import FaultInjector
+
+__all__ = ["CampaignReport", "FaultInjector", "StressCampaign"]
diff --git a/synapdrive_ai/stress/campaign.py b/synapdrive_ai/stress/campaign.py
new file mode 100644
index 0000000..b4f7e42
--- /dev/null
+++ b/synapdrive_ai/stress/campaign.py
@@ -0,0 +1,72 @@
+from __future__ import annotations
+
+from dataclasses import asdict, dataclass
+from typing import Callable, Iterable, List
+
+import numpy as np
+
+from synapdrive_ai.neuro.signal_quality import SignalQualityAnalyzer
+
+
+@dataclass(frozen=True)
+class CampaignTrial:
+ trial: int
+ fault: str
+ quality_score: float
+ quality_state: str
+ expected_block: bool
+ blocked: bool
+ invariant_passed: bool
+
+
+@dataclass(frozen=True)
+class CampaignReport:
+ total_trials: int
+ invariant_passes: int
+ invariant_failures: int
+ trials: List[CampaignTrial]
+
+ def to_dict(self):
+ return {
+ "total_trials": self.total_trials,
+ "invariant_passes": self.invariant_passes,
+ "invariant_failures": self.invariant_failures,
+ "trials": [asdict(t) for t in self.trials],
+ }
+
+
+class StressCampaign:
+ """Runs faulted signals through a supplied analysis callback and checks fail-closed behavior.
+
+ callback(signal, quality_score) must return True when the action path is blocked.
+ """
+
+ def __init__(self, sampling_rate: float, min_quality: float = 0.35) -> None:
+ self.sampling_rate = float(sampling_rate)
+ self.min_quality = float(min_quality)
+ self.quality = SignalQualityAnalyzer(sampling_rate)
+
+ def run(
+ self,
+ trials: Iterable[tuple[str, np.ndarray]],
+ callback: Callable[[np.ndarray, float], bool],
+ ) -> CampaignReport:
+ results: List[CampaignTrial] = []
+ for idx, (fault_name, signal) in enumerate(trials, 1):
+ quality = self.quality.analyze(signal)
+ expected_block = quality.score < self.min_quality
+ blocked = bool(callback(signal, quality.score))
+ invariant = (not expected_block) or blocked
+ results.append(
+ CampaignTrial(
+ idx,
+ fault_name,
+ quality.score,
+ quality.state,
+ expected_block,
+ blocked,
+ invariant,
+ )
+ )
+ passes = sum(1 for t in results if t.invariant_passed)
+ return CampaignReport(len(results), passes, len(results) - passes, results)
diff --git a/synapdrive_ai/stress/cli.py b/synapdrive_ai/stress/cli.py
new file mode 100644
index 0000000..7e08b18
--- /dev/null
+++ b/synapdrive_ai/stress/cli.py
@@ -0,0 +1,58 @@
+from __future__ import annotations
+
+import argparse
+import json
+
+import numpy as np
+
+from synapdrive_ai.bci.intent_generator import generate_intent
+from synapdrive_ai.pipeline import SynapDrivePipeline
+from synapdrive_ai.stress.campaign import StressCampaign
+from synapdrive_ai.stress.faults import FaultInjector
+
+
+def main(argv=None) -> int:
+ parser = argparse.ArgumentParser(
+ description="Run deterministic SynapDrive signal fault-injection campaign."
+ )
+ parser.add_argument("--runs", type=int, default=25)
+ parser.add_argument("--seed", type=int, default=7)
+ parser.add_argument("--sampling-rate", type=float, default=256.0)
+ args = parser.parse_args(argv)
+ rng = np.random.default_rng(args.seed)
+ n = int(args.sampling_rate * 2)
+ t = np.arange(n) / args.sampling_rate
+ base = np.sin(2 * np.pi * 20 * t) + 0.05 * rng.normal(size=n)
+ injector = FaultInjector(args.seed)
+ trials = []
+ for i in range(args.runs):
+ mode = i % 4
+ if mode == 0:
+ res = injector.dropout(base, min(0.95, 0.25 + 0.03 * i))
+ elif mode == 1:
+ res = injector.line_noise(base, args.sampling_rate, amplitude=1.5)
+ elif mode == 2:
+ res = injector.clip(base, quantile=0.6)
+ else:
+ res = injector.gaussian_noise(base, sigma=1.0)
+ trials.append((str(res.metadata["fault"]), res.signal))
+
+ pipe = SynapDrivePipeline(simulate_delay=False)
+ campaign = StressCampaign(
+ args.sampling_rate,
+ min_quality=pipe.runtime.policy.min_signal_quality,
+ )
+
+ def callback(_signal, quality_score: float) -> bool:
+ packet = generate_intent("move left")
+ packet["signal_quality"] = quality_score
+ packet["source"] = "stress-campaign"
+ return pipe.run_intent_packet(packet)["status"] == "blocked"
+
+ report = campaign.run(trials, callback)
+ print(json.dumps(report.to_dict(), indent=2))
+ return 1 if report.invariant_failures else 0
+
+
+if __name__ == "__main__":
+ raise SystemExit(main())
diff --git a/synapdrive_ai/stress/faults.py b/synapdrive_ai/stress/faults.py
new file mode 100644
index 0000000..379a71f
--- /dev/null
+++ b/synapdrive_ai/stress/faults.py
@@ -0,0 +1,81 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Dict
+
+import numpy as np
+
+
+@dataclass(frozen=True)
+class FaultResult:
+ signal: np.ndarray
+ metadata: Dict[str, float | int | str]
+
+
+class FaultInjector:
+ """Deterministic scientific fault injection for EEG-style arrays."""
+
+ def __init__(self, seed: int = 0) -> None:
+ self.rng = np.random.default_rng(seed)
+
+ def gaussian_noise(self, signal: np.ndarray, sigma: float) -> FaultResult:
+ x = np.asarray(signal, dtype=float)
+ out = x + self.rng.normal(0.0, float(sigma), size=x.shape)
+ return FaultResult(out, {"fault": "gaussian_noise", "sigma": float(sigma)})
+
+ def dropout(self, signal: np.ndarray, fraction: float) -> FaultResult:
+ x = np.asarray(signal, dtype=float).copy()
+ fraction = max(0.0, min(1.0, float(fraction)))
+ n = x.shape[-1]
+ width = int(round(n * fraction))
+ if width > 0:
+ start = int(self.rng.integers(0, max(1, n - width + 1)))
+ x[..., start : start + width] = 0.0
+ else:
+ start = 0
+ return FaultResult(
+ x,
+ {
+ "fault": "dropout",
+ "fraction": fraction,
+ "start": start,
+ "width": width,
+ },
+ )
+
+ def clip(self, signal: np.ndarray, quantile: float = 0.90) -> FaultResult:
+ x = np.asarray(signal, dtype=float).copy()
+ q = max(0.5, min(0.999, float(quantile)))
+ limit = float(np.quantile(np.abs(x), q))
+ if limit > 0:
+ x = np.clip(x, -limit, limit)
+ return FaultResult(x, {"fault": "clip", "quantile": q, "limit": limit})
+
+ def line_noise(
+ self,
+ signal: np.ndarray,
+ sampling_rate: float,
+ frequency: float = 60.0,
+ amplitude: float = 1.0,
+ ) -> FaultResult:
+ x = np.asarray(signal, dtype=float)
+ t = np.arange(x.shape[-1], dtype=float) / float(sampling_rate)
+ noise = float(amplitude) * np.sin(2 * np.pi * float(frequency) * t)
+ out = x + noise
+ return FaultResult(
+ out,
+ {
+ "fault": "line_noise",
+ "frequency": float(frequency),
+ "amplitude": float(amplitude),
+ },
+ )
+
+ def channel_swap(self, signal: np.ndarray, a: int, b: int) -> FaultResult:
+ x = np.asarray(signal, dtype=float).copy()
+ if x.ndim != 2:
+ raise ValueError("channel_swap requires shape (channels, samples)")
+ if a == b or not (0 <= a < x.shape[0]) or not (0 <= b < x.shape[0]):
+ raise ValueError("invalid channel indices")
+ x[[a, b]] = x[[b, a]]
+ return FaultResult(x, {"fault": "channel_swap", "a": a, "b": b})
diff --git a/synapdrive_ai/tests/test_final_hardening.py b/synapdrive_ai/tests/test_final_hardening.py
new file mode 100644
index 0000000..456b60f
--- /dev/null
+++ b/synapdrive_ai/tests/test_final_hardening.py
@@ -0,0 +1,260 @@
+from __future__ import annotations
+
+import sys
+import types
+from pathlib import Path
+
+import numpy as np
+import pytest
+
+from synapdrive_ai.agi.cognitive_optimizer import CognitiveOptimizer
+from synapdrive_ai.bci.intent_generator import generate_intent
+from synapdrive_ai.benchmarking.dataset import EpochDataset
+from synapdrive_ai.benchmarking.decoders import SpectralCentroidDecoder
+from synapdrive_ai.benchmarking.runtime_adapter import QualifiedDecoderAdapter
+from synapdrive_ai.cognition.world_model import WorldModel
+from synapdrive_ai.neuro.eeg_loader import EEGLoader, EEGRecording
+from synapdrive_ai.neuro.session_analyzer import SessionAnalyzer
+from synapdrive_ai.pipeline import SynapDrivePipeline
+from synapdrive_ai.vision.visual_inference import VisualInferenceEngine
+
+
+def test_declared_visual_context_has_no_random_certainty() -> None:
+ engine = VisualInferenceEngine()
+ first = engine.infer("road")
+ second = engine.infer("road")
+ assert first == second
+ assert first["certainty"] == 1.0
+ assert first["evidence_kind"] == "declared-label"
+ assert first["model_used"] is False
+
+
+def test_visual_context_does_not_boost_command_confidence() -> None:
+ optimizer = CognitiveOptimizer()
+ packet = generate_intent("move left")
+ without_visual = optimizer.optimize(packet)
+ with_visual = optimizer.optimize(packet, image_label="road")
+ assert with_visual["confidence"] == without_visual["confidence"]
+
+
+def test_unmodeled_world_action_fails_closed() -> None:
+ prediction = WorldModel().predict("invented_action")
+ assert prediction.feasible is False
+ assert prediction.predicted_risk == 1.0
+ assert prediction.after == prediction.before
+
+
+def test_analysis_only_packet_cannot_reach_actuation() -> None:
+ pipe = SynapDrivePipeline(simulate_delay=False)
+ packet = {
+ "intent": "move_left",
+ "confidence": 1.0,
+ "source": "analysis",
+ "analysis_only": True,
+ "signal_quality": 1.0,
+ }
+ out = pipe.run_intent_packet(packet)
+ assert out["status"] == "blocked"
+ assert "analysis-only-inference" in out["reason"]
+ assert pipe.get_action_log() == []
+
+
+def test_session_analyzer_without_decoder_is_analysis_only() -> None:
+ sr = 128.0
+ t = np.arange(256) / sr
+ signal = np.sin(2 * np.pi * 10 * t)
+ recording = EEGLoader(sr).load_array(signal, channel_names=["C3"])
+ report = SessionAnalyzer(channel="C3", window_s=1.0, step_s=1.0).run(recording)
+ assert report.n_epochs == 2
+ assert all(epoch.pipeline_status == "blocked" for epoch in report.epochs)
+ assert all(epoch.pipeline_confidence == 0.0 for epoch in report.epochs)
+
+
+def _install_fake_brainflow(monkeypatch: pytest.MonkeyPatch) -> None:
+ board_mod = types.ModuleType("brainflow.board_shim")
+
+ class Params:
+ serial_port = ""
+
+ class Board:
+ def __init__(self, board_id, params):
+ self.board_id = board_id
+ self.params = params
+
+ def prepare_session(self):
+ return None
+
+ def start_stream(self):
+ return None
+
+ def get_board_data(self):
+ t = np.arange(256) / 256.0
+ return np.stack([np.sin(2 * np.pi * 10 * t), np.sin(2 * np.pi * 12 * t)])
+
+ def stop_stream(self):
+ return None
+
+ def release_session(self):
+ return None
+
+ @staticmethod
+ def get_eeg_channels(_board_id):
+ return [0, 1]
+
+ @staticmethod
+ def get_sampling_rate(_board_id):
+ return 256
+
+ setattr(board_mod, "BoardShim", Board)
+ setattr(board_mod, "BrainFlowInputParams", Params)
+ pkg = types.ModuleType("brainflow")
+ monkeypatch.setitem(sys.modules, "brainflow", pkg)
+ monkeypatch.setitem(sys.modules, "brainflow.board_shim", board_mod)
+
+
+def test_brainflow_without_decoder_abstains(monkeypatch: pytest.MonkeyPatch) -> None:
+ _install_fake_brainflow(monkeypatch)
+ import synapdrive_ai.integrations.brainflow_adapter as module
+
+ monkeypatch.setattr(module.time, "sleep", lambda _seconds: None)
+ source = module.BrainFlowIntentSource(stream_seconds=0.001)
+ packet = source.next_intent_packet()
+ assert packet["intent"] == "unknown"
+ assert packet["confidence"] == 0.0
+ assert packet["decoder_status"] == "not-configured"
+ assert packet["analysis_only"] is True
+
+
+def test_brainflow_explicit_decoder_is_only_decoder_authority(
+ monkeypatch: pytest.MonkeyPatch,
+) -> None:
+ _install_fake_brainflow(monkeypatch)
+ import synapdrive_ai.integrations.brainflow_adapter as module
+
+ monkeypatch.setattr(module.time, "sleep", lambda _seconds: None)
+
+ def decoder(data, metadata):
+ assert data.shape == (2, 256)
+ assert metadata["sampling_rate"] == 256.0
+ return {
+ "intent": "move_left",
+ "confidence": 0.8,
+ "probabilities": {"move_left": 0.8, "unknown": 0.2},
+ "source": "test-decoder",
+ }
+
+ source = module.BrainFlowIntentSource(stream_seconds=0.001, decoder=decoder)
+ packet = source.next_intent_packet()
+ assert packet["intent"] == "move_left"
+ assert packet["confidence"] == 0.8
+ assert packet["decoder_status"] == "configured"
+ assert packet["source"] == "test-decoder"
+
+
+def _install_fake_lsl(monkeypatch: pytest.MonkeyPatch, sampling_rate: float = 256.0) -> None:
+ mod = types.ModuleType("pylsl")
+
+ class Info:
+ def name(self):
+ return "unit-test-stream"
+
+ def type(self):
+ return "EEG"
+
+ def nominal_srate(self):
+ return sampling_rate
+
+ class Inlet:
+ def __init__(self, info, max_chunklen=512):
+ self.info = info
+ self.max_chunklen = max_chunklen
+ self.calls = 0
+
+ def pull_chunk(self, timeout=0.2, max_samples=512):
+ self.calls += 1
+ if self.calls == 1:
+ t = np.arange(64) / 256.0
+ samples = np.stack(
+ [np.sin(2 * np.pi * 10 * t), np.sin(2 * np.pi * 12 * t)], axis=1
+ )
+ return samples.tolist(), t.tolist()
+ return [], []
+
+ def resolve_stream(_prop, _value, timeout=5.0):
+ del timeout
+ return [Info()]
+
+ setattr(mod, "StreamInlet", Inlet)
+ setattr(mod, "resolve_stream", resolve_stream)
+ monkeypatch.setitem(sys.modules, "pylsl", mod)
+
+
+def test_lsl_without_decoder_abstains(monkeypatch: pytest.MonkeyPatch) -> None:
+ _install_fake_lsl(monkeypatch)
+ import synapdrive_ai.integrations.lsl_adapter as module
+
+ ticks = iter([0.0, 0.0, 1.0])
+ monkeypatch.setattr(module.time, "monotonic", lambda: next(ticks, 1.0))
+ source = module.LSLIntentSource(snapshot_seconds=0.1)
+ packet = source.next_intent_packet()
+ assert packet["intent"] == "unknown"
+ assert packet["confidence"] == 0.0
+ assert packet["decoder_status"] == "not-configured"
+ assert packet["analysis_only"] is True
+
+
+def test_repository_contains_no_retired_fabricated_routes() -> None:
+ root = Path(__file__).resolve().parents[2]
+ retired = ("Tesla" + "AutonomySystem", "SpaceX" + "CommandCore", "Hyperloop" + "Ops")
+ for path in root.rglob("*.py"):
+ if "__pycache__" in path.parts:
+ continue
+ text = path.read_text(encoding="utf-8")
+ for marker in retired:
+ assert marker not in text, f"retired fabricated route {marker!r} remains in {path}"
+
+
+def test_qualified_decoder_adapter_connects_to_session_analyzer() -> None:
+ sr = 128.0
+ n = 128
+ times = np.arange(n, dtype=float) / sr
+ rng = np.random.default_rng(123)
+ epochs = []
+ labels = []
+ for label, frequency in (("left", 10.0), ("right", 20.0)):
+ for _ in range(12):
+ signal = np.sin(2.0 * np.pi * frequency * times)
+ signal += rng.normal(0.0, 0.01, n)
+ epochs.append(signal[None, :])
+ labels.append(label)
+ dataset = EpochDataset(
+ epochs=np.asarray(epochs, dtype=float),
+ labels=np.asarray(labels),
+ sampling_rate=sr,
+ channel_names=["C3"],
+ source="deterministic-synthetic-verification",
+ )
+ adapter = QualifiedDecoderAdapter(
+ SpectralCentroidDecoder(),
+ dataset,
+ {"left": "move_left", "right": "move_right"},
+ )
+ assert adapter.qualification.qualified is True
+
+ recording = EEGRecording(
+ channels=["C3"],
+ data=np.sin(2.0 * np.pi * 10.0 * times)[None, :],
+ sampling_rate=sr,
+ duration_s=1.0,
+ source_file="synthetic-verification",
+ )
+ report = SessionAnalyzer(
+ channel="C3",
+ window_s=1.0,
+ step_s=1.0,
+ decoder=adapter,
+ ).run(recording)
+ assert report.n_epochs == 1
+ assert report.n_success == 1
+ assert report.epochs[0].pipeline_status == "success"
+ assert report.epochs[0].pipeline_confidence >= adapter.policy.abstain_threshold
diff --git a/synapdrive_ai/tests/test_legacy_core.py b/synapdrive_ai/tests/test_legacy_core.py
index 19a7957..731c182 100644
--- a/synapdrive_ai/tests/test_legacy_core.py
+++ b/synapdrive_ai/tests/test_legacy_core.py
@@ -13,3 +13,11 @@ def test_legacy_core_router_example_remains_importable() -> None:
result = router.route(parsed)
assert result == "Executed simulated move: forward"
+
+
+def test_legacy_core_package_exports_router() -> None:
+ from core.intent_router import IntentParser as ExportedParser
+ from core.intent_router import IntentRouter as ExportedRouter
+
+ assert ExportedParser is IntentParser
+ assert ExportedRouter is IntentRouter
diff --git a/synapdrive_ai/tests/test_neuro.py b/synapdrive_ai/tests/test_neuro.py
index baadab1..ab6d82a 100644
--- a/synapdrive_ai/tests/test_neuro.py
+++ b/synapdrive_ai/tests/test_neuro.py
@@ -226,6 +226,7 @@ def test_impossible_confidence_defers(self):
bridge = ExecutorBridge(simulate_delay=False)
trace = bridge.execute(plan)
assert any(s.pipeline_status == "deferred" for s in trace.steps)
+ assert bridge._pipe.get_action_log() == []
def test_abort_fallback_stops_plan(self):
plan = TaskPlan(
@@ -239,6 +240,7 @@ def test_abort_fallback_stops_plan(self):
trace = bridge.execute(plan)
assert trace.outcome == "aborted"
assert len(trace.steps) == 1
+ assert bridge._pipe.get_action_log() == []
def test_complete_fallback_proceeds(self):
plan = TaskPlan(
@@ -261,3 +263,46 @@ def test_empty_plan(self):
trace = bridge.execute(plan)
assert trace.n_steps == 0
assert trace.outcome == "completed"
+
+
+def test_bdf_24bit_single_channel_loading(tmp_path):
+ def field(value, width):
+ return str(value).ljust(width)[:width].encode("ascii")
+
+ n_signals = 1
+ ns = 4
+ header_bytes = 256 + 256 * n_signals
+ fixed = bytearray(b" " * 256)
+ fixed[0:8] = field("255", 8)
+ fixed[184:192] = field(header_bytes, 8)
+ fixed[236:244] = field(1, 8)
+ fixed[244:252] = field(1, 8)
+ fixed[252:256] = field(n_signals, 4)
+
+ signal_header = bytearray()
+ signal_header += field("C3", 16)
+ signal_header += field("", 80)
+ signal_header += field("uV", 8)
+ signal_header += field(-8388608, 8)
+ signal_header += field(8388607, 8)
+ signal_header += field(-8388608, 8)
+ signal_header += field(8388607, 8)
+ signal_header += field("", 80)
+ signal_header += field(ns, 8)
+ signal_header += field("", 32)
+
+ samples = [-8388608, -1, 0, 8388607]
+ encoded = bytearray()
+ for sample in samples:
+ unsigned = sample if sample >= 0 else sample + (1 << 24)
+ encoded.extend(
+ [unsigned & 0xFF, (unsigned >> 8) & 0xFF, (unsigned >> 16) & 0xFF]
+ )
+
+ path = tmp_path / "tiny.bdf"
+ path.write_bytes(bytes(fixed) + bytes(signal_header) + bytes(encoded))
+ recording = EEGLoader().load(path)
+ assert recording.n_samples == ns
+ assert recording.sampling_rate == 4.0
+ assert recording.channel("C3")[0] < 0
+ assert recording.channel("C3")[-1] > 0
diff --git a/synapdrive_ai/tests/test_v1_benchmark_stress.py b/synapdrive_ai/tests/test_v1_benchmark_stress.py
new file mode 100644
index 0000000..741b4e2
--- /dev/null
+++ b/synapdrive_ai/tests/test_v1_benchmark_stress.py
@@ -0,0 +1,199 @@
+from __future__ import annotations
+
+import numpy as np
+
+from synapdrive_ai.adaptation.guarded import GuardedThresholdAdapter
+from synapdrive_ai.benchmarking.dataset import EpochDataset, load_npz_dataset
+from synapdrive_ai.benchmarking.decoders import (
+ EnsembleDecoder,
+ RiemannianCentroidDecoder,
+ SpectralCentroidDecoder,
+)
+from synapdrive_ai.benchmarking.evaluation import evaluate_decoder
+from synapdrive_ai.stress.campaign import StressCampaign
+from synapdrive_ai.stress.faults import FaultInjector
+
+
+def _dataset(n_per_class=24, sr=128.0):
+ rng = np.random.default_rng(11)
+ n = int(sr)
+ t = np.arange(n) / sr
+ epochs, labels = [], []
+ for label, freq in [("left", 10.0), ("right", 20.0)]:
+ for _ in range(n_per_class):
+ ch1 = np.sin(2 * np.pi * freq * t) + rng.normal(0, 0.12, n)
+ ch2 = 0.7 * np.sin(2 * np.pi * freq * t + 0.3) + rng.normal(0, 0.12, n)
+ epochs.append(np.stack([ch1, ch2]))
+ labels.append(label)
+ return EpochDataset(np.asarray(epochs), np.asarray(labels), sr, ["C3", "C4"], "test-generated")
+
+
+def test_npz_dataset_contract_roundtrip(tmp_path):
+ ds = _dataset(4)
+ p = tmp_path / "d.npz"
+ np.savez(
+ p,
+ X=ds.epochs,
+ y=ds.labels,
+ sampling_rate=np.array([ds.sampling_rate]),
+ channel_names=np.array(ds.channel_names),
+ )
+ loaded = load_npz_dataset(p)
+ assert loaded.epochs.shape == ds.epochs.shape
+ assert loaded.channel_names == ["C3", "C4"]
+
+
+def test_spectral_decoder_learns_frequency_classes():
+ ds = _dataset()
+ report = evaluate_decoder(SpectralCentroidDecoder(), ds, seed=3)
+ assert report.accuracy > 0.9
+ assert report.n_test > 0
+
+
+def test_riemannian_decoder_outputs_probabilities():
+ ds = _dataset()
+ decoder = RiemannianCentroidDecoder().fit(ds.epochs, ds.labels, ds.sampling_rate)
+ p = decoder.predict_proba(ds.epochs[:3], ds.sampling_rate)
+ assert p.shape == (3, 2)
+ assert np.allclose(np.sum(p, axis=1), 1.0)
+
+
+def test_ensemble_benchmark_metrics_are_bounded():
+ ds = _dataset()
+ decoder = EnsembleDecoder([SpectralCentroidDecoder(), RiemannianCentroidDecoder()])
+ report = evaluate_decoder(decoder, ds)
+ assert 0 <= report.accuracy <= 1
+ assert 0 <= report.balanced_accuracy <= 1
+ assert 0 <= report.ece <= 1
+ assert 0 <= report.coverage <= 1
+
+
+def test_guarded_adapter_promotes_safer_better_threshold():
+ rows = [
+ {"confidence": 0.9, "safe": True},
+ {"confidence": 0.8, "safe": True},
+ {"confidence": 0.6, "safe": False},
+ {"confidence": 0.55, "safe": False},
+ ]
+ adapter = GuardedThresholdAdapter(0.45)
+ decision = adapter.evaluate_candidate(0.7, rows)
+ assert decision.promoted is True
+ assert adapter.threshold == 0.7
+
+
+def test_guarded_adapter_rejects_more_unsafe_accepts():
+ rows = [
+ {"confidence": 0.9, "safe": True},
+ {"confidence": 0.6, "safe": False},
+ ]
+ adapter = GuardedThresholdAdapter(0.7)
+ decision = adapter.evaluate_candidate(0.5, rows)
+ assert decision.promoted is False
+ assert adapter.threshold == 0.7
+
+
+def test_fault_injector_is_seed_deterministic():
+ x = np.ones(100)
+ a = FaultInjector(5).dropout(x, 0.2)
+ b = FaultInjector(5).dropout(x, 0.2)
+ assert np.array_equal(a.signal, b.signal)
+ assert a.metadata == b.metadata
+
+
+def test_fault_dropout_writes_zeros():
+ out = FaultInjector(1).dropout(np.ones(100), 0.5)
+ assert np.sum(out.signal == 0) >= 50
+
+
+def test_fault_channel_swap():
+ x = np.vstack([np.zeros(8), np.ones(8)])
+ out = FaultInjector().channel_swap(x, 0, 1).signal
+ assert np.all(out[0] == 1)
+ assert np.all(out[1] == 0)
+
+
+def test_stress_campaign_enforces_low_quality_block():
+ clean = np.sin(2 * np.pi * 10 * np.arange(512) / 256)
+ bad = np.zeros(512)
+ campaign = StressCampaign(256, min_quality=0.5)
+ report = campaign.run([("clean", clean), ("flatline", bad)], lambda _sig, q: q < 0.5)
+ assert report.invariant_failures == 0
+ assert report.total_trials == 2
+
+
+def test_decoder_arena_ranks_all_builtins():
+ from synapdrive_ai.benchmarking.evaluation import run_arena
+
+ reports = run_arena(_dataset())
+ assert len(reports) == 3
+ assert {report.decoder for report in reports} == {
+ "SpectralCentroidDecoder",
+ "RiemannianCentroidDecoder",
+ "EnsembleDecoder",
+ }
+
+
+def test_epoch_recording_uses_explicit_events():
+ from synapdrive_ai.benchmarking.epoching import LabeledEvent, epoch_recording
+ from synapdrive_ai.neuro.eeg_loader import EEGLoader
+
+ sr = 100.0
+ data = np.vstack([np.arange(500), np.arange(500) * 2.0])
+ recording = EEGLoader(sr).load_array(data, channel_names=["C3", "C4"])
+ dataset = epoch_recording(
+ recording,
+ [LabeledEvent(1.0, "left"), LabeledEvent(3.0, "right")],
+ tmin_s=0.0,
+ tmax_s=0.5,
+ channels=["C3"],
+ )
+ assert dataset.epochs.shape == (2, 1, 50)
+ assert dataset.labels.tolist() == ["left", "right"]
+
+
+def test_qualified_decoder_adapter_bridges_labeled_data_to_action_packet():
+ from synapdrive_ai.benchmarking.runtime_adapter import QualifiedDecoderAdapter
+
+ ds = _dataset()
+ adapter = QualifiedDecoderAdapter(
+ SpectralCentroidDecoder(),
+ ds,
+ {"left": "move_left", "right": "move_right"},
+ seed=3,
+ )
+ assert adapter.qualification.qualified is True
+ packet = adapter(ds.epochs[0], {"sampling_rate": ds.sampling_rate})
+ assert packet["intent"] == "move_left"
+ assert packet["analysis_only"] is False
+ assert packet["neural_decode_performed"] is True
+ assert 0.0 <= packet["confidence"] <= 1.0
+
+
+def test_qualified_decoder_adapter_abstains_on_sampling_rate_mismatch():
+ from synapdrive_ai.benchmarking.runtime_adapter import QualifiedDecoderAdapter
+
+ ds = _dataset()
+ adapter = QualifiedDecoderAdapter(
+ SpectralCentroidDecoder(),
+ ds,
+ {"left": "move_left", "right": "move_right"},
+ seed=3,
+ )
+ packet = adapter(ds.epochs[0], {"sampling_rate": ds.sampling_rate + 1})
+ assert packet["intent"] == "unknown"
+ assert packet["analysis_only"] is True
+ assert "sampling-rate mismatch" in packet["abstention_reason"]
+
+
+def test_qualified_decoder_requires_action_mapping_for_all_classes():
+ from synapdrive_ai.benchmarking.runtime_adapter import QualifiedDecoderAdapter
+
+ ds = _dataset()
+ adapter = QualifiedDecoderAdapter(
+ SpectralCentroidDecoder(),
+ ds,
+ {"left": "move_left"},
+ seed=3,
+ )
+ assert adapter.qualification.qualified is False
+ assert "unmapped-decoder-classes:right" in adapter.qualification.reason
diff --git a/synapdrive_ai/tests/test_v1_cognition_evidence.py b/synapdrive_ai/tests/test_v1_cognition_evidence.py
new file mode 100644
index 0000000..d3ca10f
--- /dev/null
+++ b/synapdrive_ai/tests/test_v1_cognition_evidence.py
@@ -0,0 +1,108 @@
+from __future__ import annotations
+
+from typing import cast
+
+from synapdrive_ai.assurance.hashchain import EvidenceHashChain
+from synapdrive_ai.cognition.planner import CandidateAction, CounterfactualPlanner
+from synapdrive_ai.cognition.world_model import ActionModel, WorldModel
+from synapdrive_ai.memory.evidence_memory import EvidenceMemory
+from synapdrive_ai.runtime.reality import RealityReconciler
+
+
+def test_world_model_known_action_predicts_effects():
+ world = WorldModel({"x": 0})
+ world.register(ActionModel("right", effects={"x": 1}, predicted_risk=0.1))
+ p = world.predict("right")
+ assert p.after["x"] == 1
+ assert world.state["x"] == 0
+
+
+def test_world_model_commit_changes_state():
+ world = WorldModel({"x": 0})
+ world.register(ActionModel("right", effects={"x": 1}))
+ world.commit(world.predict("right"))
+ assert world.state["x"] == 1
+
+
+def test_world_model_unknown_does_not_invent_effects():
+ world = WorldModel({"x": 3})
+ p = world.predict("unmodeled")
+ assert p.after == {"x": 3}
+ assert "unmodeled-action" in p.unmet_requirements
+
+
+def test_counterfactual_planner_penalizes_risk():
+ world = WorldModel({"goal": 0})
+ world.register_many([
+ ActionModel("safe", effects={"goal": 1}, predicted_risk=0.1),
+ ActionModel("risky", effects={"goal": 1}, predicted_risk=0.9),
+ ])
+ planner = CounterfactualPlanner(world)
+ choice = planner.rank([
+ CandidateAction("safe", 1.0, 0.9),
+ CandidateAction("risky", 1.0, 0.9),
+ ], lambda s: float(cast(int | float, s.get("goal", 0))))
+ assert choice.selected is not None
+ assert choice.selected.action == "safe"
+
+
+def test_evidence_hash_chain_verifies():
+ chain = EvidenceHashChain()
+ chain.append({"a": 1})
+ chain.append({"b": 2})
+ assert EvidenceHashChain.verify(chain.entries()) is True
+
+
+def test_evidence_hash_chain_detects_tamper():
+ chain = EvidenceHashChain()
+ chain.append({"a": 1})
+ rows = [e.to_dict() for e in chain.entries()]
+ rows[0]["payload"]["a"] = 999
+ assert EvidenceHashChain.verify(rows) is False
+
+
+def test_evidence_memory_validates_aligned():
+ memory = EvidenceMemory()
+ row = memory.record(
+ cycle_id="1",
+ intent="left",
+ confidence=0.8,
+ outcome="success",
+ aligned=True,
+ reason="ok",
+ )
+ assert row.state == "validated"
+ assert len(memory.validated_for_intent("left")) == 1
+
+
+def test_evidence_memory_quarantines_contradiction():
+ memory = EvidenceMemory()
+ memory.record(
+ cycle_id="1",
+ intent="left",
+ confidence=0.8,
+ outcome="success",
+ aligned=False,
+ reason="ErrP",
+ )
+ assert len(memory.quarantined()) == 1
+ assert memory.validated_for_intent("left") == []
+
+
+def test_reality_reconciler_errp_contradicts_success():
+ verdict = RealityReconciler().reconcile(
+ predicted_success=0.9,
+ result_packet={"status": "success"},
+ feedback={"errp_probability": 0.95},
+ )
+ assert verdict.aligned is False
+ assert verdict.outcome == "contradicted"
+
+
+def test_reality_reconciler_explicit_rejection_wins():
+ verdict = RealityReconciler().reconcile(
+ predicted_success=0.9,
+ result_packet={"status": "success"},
+ feedback={"accepted": False},
+ )
+ assert verdict.aligned is False
\ No newline at end of file
diff --git a/synapdrive_ai/tests/test_v1_neuro.py b/synapdrive_ai/tests/test_v1_neuro.py
new file mode 100644
index 0000000..7f54def
--- /dev/null
+++ b/synapdrive_ai/tests/test_v1_neuro.py
@@ -0,0 +1,131 @@
+from __future__ import annotations
+
+import numpy as np
+import pytest
+
+from synapdrive_ai.neuro.drift import FeatureDriftMonitor
+from synapdrive_ai.neuro.errp import ErrPFeatureExtractor, ErrPLDAClassifier
+from synapdrive_ai.neuro.fusion import ModalityEvidence, WeightedEvidenceFusion
+from synapdrive_ai.neuro.signal_quality import SignalQualityAnalyzer
+from synapdrive_ai.neuro.uncertainty import (
+ ReliabilityCalibrator,
+ estimate_uncertainty,
+ normalize_distribution,
+ packet_distribution,
+)
+
+
+def test_distribution_normalizes():
+ p = normalize_distribution({"a": 2, "b": 1})
+ assert abs(sum(p.values()) - 1.0) < 1e-12
+
+
+def test_packet_distribution_has_unknown_mass():
+ p = packet_distribution({"intent": "left", "confidence": 0.8})
+ assert p["left"] == pytest.approx(0.8)
+ assert p["unknown"] == pytest.approx(0.2)
+
+
+def test_uncertainty_lower_for_decisive_distribution():
+ decisive = estimate_uncertainty({"a": 0.99, "b": 0.01}).combined
+ ambiguous = estimate_uncertainty({"a": 0.51, "b": 0.49}).combined
+ assert decisive < ambiguous
+
+
+def test_ensemble_disagreement_contributes_uncertainty():
+ same = estimate_uncertainty({"a": 0.8, "b": 0.2}, [{"a": 0.8, "b": 0.2}, {"a": 0.8, "b": 0.2}])
+ split = estimate_uncertainty(
+ {"a": 0.5, "b": 0.5},
+ [{"a": 0.99, "b": 0.01}, {"a": 0.01, "b": 0.99}],
+ )
+ assert split.disagreement > same.disagreement
+
+
+def test_calibrator_requires_fit():
+ with pytest.raises(RuntimeError):
+ ReliabilityCalibrator().transform(0.8)
+
+
+def test_calibrator_maps_to_empirical_accuracy():
+ cal = ReliabilityCalibrator(n_bins=2).fit([0.1, 0.2, 0.8, 0.9], [False, False, True, True])
+ assert cal.transform(0.85) == pytest.approx(1.0)
+ assert cal.transform(0.15) == pytest.approx(0.0)
+
+
+def test_signal_quality_good_clean_signal():
+ sr = 256
+ t = np.arange(sr * 2) / sr
+ x = np.sin(2 * np.pi * 10 * t) + 0.1 * np.sin(2 * np.pi * 20 * t)
+ report = SignalQualityAnalyzer(sr).analyze(x)
+ assert report.score > 0.8
+ assert report.state == "good"
+
+
+def test_signal_quality_detects_nan():
+ x = np.ones(512)
+ x[::2] = np.nan
+ report = SignalQualityAnalyzer(256).analyze(x)
+ assert "non-finite-samples" in report.issues
+ assert report.score < 0.8
+
+
+def test_signal_quality_detects_flatline():
+ report = SignalQualityAnalyzer(256).analyze(np.ones(512))
+ assert "flatline" in report.issues
+ assert report.state in {"unreliable", "invalid", "degraded"}
+
+
+def test_signal_quality_detects_dropout():
+ sr = 256
+ t = np.arange(512) / sr
+ x = np.sin(2 * np.pi * 10 * t)
+ x[:300] = 0
+ report = SignalQualityAnalyzer(sr).analyze(x)
+ assert "dropout" in report.issues
+
+
+def test_fusion_prefers_reliable_modality():
+ result = WeightedEvidenceFusion().fuse([
+ ModalityEvidence("eeg", {"left": 0.9, "right": 0.1}, 0.2),
+ ModalityEvidence("gaze", {"left": 0.1, "right": 0.9}, 1.0),
+ ])
+ assert result.intent == "right"
+ assert result.weights["gaze"] > result.weights["eeg"]
+
+
+def test_fusion_rejects_zero_reliability():
+ with pytest.raises(ValueError):
+ WeightedEvidenceFusion().fuse([ModalityEvidence("eeg", {"left": 1.0}, 0.0)])
+
+
+def test_drift_monitor_detects_far_outlier():
+ rng = np.random.default_rng(2)
+ baseline = rng.normal(0, 0.2, (100, 3))
+ monitor = FeatureDriftMonitor().fit(baseline)
+ assert monitor.evaluate([8, 8, 8]).drifted is True
+
+
+def test_errp_feature_extractor_windows():
+ sr = 100
+ x = np.zeros(sr)
+ x[20:35] = -2.0
+ x[35:60] = 1.0
+ f = ErrPFeatureExtractor(sr).extract(x)
+ assert f.negative_peak == pytest.approx(-2.0)
+ assert f.positive_peak == pytest.approx(1.0)
+
+
+def test_errp_lda_learns_supplied_calibration():
+ rng = np.random.default_rng(4)
+ x0 = rng.normal(0, 0.1, (20, 4))
+ x1 = rng.normal(2, 0.1, (20, 4))
+ x = np.vstack([x0, x1])
+ y = [0] * 20 + [1] * 20
+ clf = ErrPLDAClassifier().fit(x, y)
+ assert clf.predict([2, 2, 2, 2]) == 1
+ assert clf.predict([0, 0, 0, 0]) == 0
+
+
+def test_errp_classifier_requires_both_classes():
+ with pytest.raises(ValueError):
+ ErrPLDAClassifier().fit(np.zeros((4, 4)), [0, 0, 0, 0])
diff --git a/synapdrive_ai/tests/test_v1_pipeline_integration.py b/synapdrive_ai/tests/test_v1_pipeline_integration.py
new file mode 100644
index 0000000..01e75ef
--- /dev/null
+++ b/synapdrive_ai/tests/test_v1_pipeline_integration.py
@@ -0,0 +1,82 @@
+from __future__ import annotations
+
+from synapdrive_ai.bci.intent_generator import generate_intent
+from synapdrive_ai.pipeline import SynapDrivePipeline
+
+
+def test_pipeline_exposes_runtime_reality_memory_and_evidence():
+ pipe = SynapDrivePipeline(simulate_delay=False)
+ out = pipe.run_text_command("move left")
+ assert out["status"] == "success"
+ assert out["runtime"]["allowed"] is True
+ assert out["reality"]["aligned"] is True
+ assert out["memory"]["state"] == "validated"
+ assert len(out["evidence"]["event_hash"]) == 64
+
+
+def test_pipeline_quarantines_errp_contradicted_success():
+ pipe = SynapDrivePipeline(simulate_delay=False)
+ out = pipe.run_text_command("move left", feedback={"errp_probability": 0.99})
+ assert out["status"] == "success"
+ assert out["reality"]["aligned"] is False
+ assert out["memory"]["state"] == "quarantined"
+
+
+def test_quarantined_memory_does_not_boost_legacy_optimizer():
+ pipe = SynapDrivePipeline(simulate_delay=False)
+ first = pipe.run_text_command("move left", feedback={"errp_probability": 0.99})
+ second = pipe.run_text_command("move left")
+ assert first["intent"]["confidence"] == second["intent"]["confidence"]
+
+
+def test_validated_memory_is_context_not_confidence_inflation():
+ pipe = SynapDrivePipeline(simulate_delay=False)
+ first = pipe.run_text_command("move left")
+ second = pipe.run_text_command("move left")
+ assert second["intent"]["confidence"] == first["intent"]["confidence"]
+ assert first["intent"]["history_support_count"] == 0
+ assert second["intent"]["history_support_count"] == 1
+
+
+def test_bad_signal_quality_blocks_before_action():
+ pipe = SynapDrivePipeline(simulate_delay=False)
+ packet = generate_intent("move left")
+ packet["signal_quality"] = 0.0
+ out = pipe.run_intent_packet(packet)
+ assert out["status"] == "blocked"
+ assert "signal-quality-below-policy" in out["reason"]
+ assert pipe.get_action_log() == []
+
+
+def test_permission_revoke_blocks_but_stop_stays_available():
+ pipe = SynapDrivePipeline(simulate_delay=False)
+ pipe.runtime.permissions.revoke_all()
+ blocked = pipe.run_text_command("move left")
+ stop = pipe.run_text_command("stop")
+ assert blocked["status"] == "blocked"
+ assert stop["status"] == "success"
+
+
+def test_successful_action_commits_explicit_world_model_state():
+ pipe = SynapDrivePipeline(simulate_delay=False)
+ out = pipe.run_text_command("switch mode to assistive")
+ assert out["status"] == "success"
+ assert out["world_state"]["mode"] == "assistive"
+ assert pipe.runtime.world_model.state["mode"] == "assistive"
+
+
+def test_evidence_chain_detects_no_tamper_after_multiple_cycles():
+ pipe = SynapDrivePipeline(simulate_delay=False)
+ pipe.run_text_command("move left")
+ pipe.run_text_command("stop")
+ report = pipe.get_assurance_report()
+ assert report["evidence_chain_entries"] == 2
+ assert report["evidence_chain_valid"] is True
+
+
+def test_unknown_text_abstains_with_runtime_evidence():
+ pipe = SynapDrivePipeline(simulate_delay=False)
+ out = pipe.run_text_command("not a known command")
+ assert out["status"] == "blocked"
+ assert out["runtime"]["allowed"] is False
+ assert out["memory"]["state"] == "not-recorded"
diff --git a/synapdrive_ai/tests/test_v1_runtime.py b/synapdrive_ai/tests/test_v1_runtime.py
new file mode 100644
index 0000000..80bcaa1
--- /dev/null
+++ b/synapdrive_ai/tests/test_v1_runtime.py
@@ -0,0 +1,80 @@
+from __future__ import annotations
+
+import json
+
+import pytest
+
+from synapdrive_ai.cognition.world_model import ActionModel, WorldModel
+from synapdrive_ai.governance.permission import PermissionGate
+from synapdrive_ai.governance.policy import RuntimePolicy
+from synapdrive_ai.runtime.governed_runtime import GovernedRuntime
+
+
+def test_runtime_policy_rejects_bad_probability():
+ with pytest.raises(ValueError):
+ RuntimePolicy(min_confidence=1.2)
+
+
+def test_runtime_policy_roundtrip_json(tmp_path):
+ path = tmp_path / "policy.json"
+ policy = RuntimePolicy(min_confidence=0.6)
+ path.write_text(json.dumps(policy.to_dict()))
+ assert RuntimePolicy.from_json(path).min_confidence == 0.6
+
+
+def test_permission_safe_state_survives_revoke():
+ gate = PermissionGate()
+ gate.revoke_all()
+ assert gate.evaluate("halt_all_motion").allowed is True
+ assert gate.evaluate("move_left").allowed is False
+
+
+def test_permission_deny_pattern():
+ gate = PermissionGate()
+ assert gate.evaluate("override_security_now").allowed is False
+
+
+def test_runtime_blocks_low_confidence():
+ rt = GovernedRuntime(policy=RuntimePolicy(min_confidence=0.7))
+ out = rt.assess({"intent": "move_left", "confidence": 0.6, "uncertainty": 0.1})
+ assert out.allowed is False
+ assert "confidence-below-policy" in out.assessment.issues
+
+
+def test_runtime_blocks_uncertainty():
+ rt = GovernedRuntime(policy=RuntimePolicy(max_uncertainty=0.2))
+ out = rt.assess({"intent": "move_left", "confidence": 0.9, "uncertainty": 0.5})
+ assert out.allowed is False
+ assert "uncertainty-above-policy" in out.assessment.issues
+
+
+def test_runtime_blocks_signal_quality():
+ rt = GovernedRuntime(policy=RuntimePolicy(min_signal_quality=0.8))
+ out = rt.assess(
+ {
+ "intent": "move_left",
+ "confidence": 0.9,
+ "uncertainty": 0.1,
+ "signal_quality": 0.2,
+ }
+ )
+ assert out.allowed is False
+ assert "signal-quality-below-policy" in out.assessment.issues
+
+
+def test_runtime_world_precondition_failure():
+ world = WorldModel({"armed": False})
+ world.register(ActionModel("move_left", required={"armed": True}))
+ rt = GovernedRuntime(world_model=world)
+ out = rt.assess({"intent": "move_left", "confidence": 0.9, "uncertainty": 0.1})
+ assert out.allowed is False
+ assert "world-model-precondition-failed" in out.assessment.issues
+
+
+def test_runtime_blocks_predicted_risk():
+ world = WorldModel()
+ world.register(ActionModel("move_left", predicted_risk=0.9))
+ rt = GovernedRuntime(world_model=world)
+ out = rt.assess({"intent": "move_left", "confidence": 0.9, "uncertainty": 0.1})
+ assert out.allowed is False
+ assert "predicted-risk-above-policy" in out.assessment.issues
diff --git a/synapdrive_ai/tests/test_v1_shared_autonomy.py b/synapdrive_ai/tests/test_v1_shared_autonomy.py
new file mode 100644
index 0000000..45ce7b4
--- /dev/null
+++ b/synapdrive_ai/tests/test_v1_shared_autonomy.py
@@ -0,0 +1,114 @@
+from typing import cast
+
+import pytest
+
+from synapdrive_ai.cognition.expectation import ExpectationReconciler
+from synapdrive_ai.cognition.planner import CandidateAction, CounterfactualPlanner
+from synapdrive_ai.cognition.shared_autonomy import SharedAutonomyArbiter
+from synapdrive_ai.cognition.world_model import ActionModel, WorldModel
+from synapdrive_ai.runtime.governed_runtime import GovernedRuntime
+from synapdrive_ai.runtime.shadow import ShadowController
+from synapdrive_ai.runtime.simplex import SimplexController
+
+
+def test_shared_autonomy_can_choose_lower_risk_machine_action():
+ world = WorldModel({"goal": 0})
+ world.register(ActionModel("direct", effects={"goal": 1}, predicted_risk=0.8))
+ world.register(ActionModel("detour", effects={"goal": 1}, predicted_risk=0.1))
+ arbiter = SharedAutonomyArbiter(CounterfactualPlanner(world))
+
+ proposal = arbiter.propose(
+ user_intent="direct",
+ user_confidence=0.8,
+ machine_candidates=[CandidateAction("detour", 0.95, 0.9)],
+ goal=lambda state: float(
+ cast(int | float, state.get("goal", 0))
+ ),
+ )
+
+ assert proposal.selected_action == "detour"
+ assert proposal.preserved_user_goal is True
+
+
+def test_expectation_reconciler_requires_explicit_human_expectation():
+ delta = ExpectationReconciler().compare(0.8, True)
+
+ assert delta.machine_reality == pytest.approx(0.2)
+ assert delta.human_reality is None
+
+
+def test_expectation_three_way_delta():
+ delta = ExpectationReconciler().compare(
+ 0.2,
+ True,
+ human_expected_success=0.9,
+ )
+
+ assert delta.machine_reality == 0.8
+ assert delta.human_reality is not None
+ assert delta.human_reality < delta.machine_reality
+ assert delta.machine_human == pytest.approx(0.7)
+
+
+def test_simplex_uses_reversionary_controller_when_blocked():
+ runtime = GovernedRuntime()
+ packet = {
+ "intent": "move_left",
+ "confidence": 0.1,
+ "uncertainty": 0.1,
+ }
+ decision = runtime.assess(packet)
+
+ simplex = SimplexController(
+ lambda p: {
+ "status": "success",
+ "action": p["intent"],
+ },
+ lambda action: {
+ "status": "safe-state",
+ "action": action,
+ },
+ )
+
+ out = simplex.execute(packet, decision)
+
+ assert out.controller == "reversionary"
+ assert out.result["action"] == "hold_position"
+
+
+def test_simplex_uses_advanced_controller_when_allowed():
+ runtime = GovernedRuntime()
+ packet = {
+ "intent": "move_left",
+ "confidence": 0.95,
+ "uncertainty": 0.1,
+ }
+ decision = runtime.assess(packet)
+
+ simplex = SimplexController(
+ lambda p: {
+ "status": "success",
+ "action": p["intent"],
+ },
+ lambda action: {
+ "status": "safe-state",
+ "action": action,
+ },
+ )
+
+ assert simplex.execute(packet, decision).controller == "advanced"
+
+
+def test_shadow_controller_never_executes_trusted_path():
+ calls = []
+
+ def policy(context):
+ calls.append(context)
+ return "experimental", {"score": 0.9}
+
+ shadow = ShadowController(policy)
+ result = shadow.evaluate("trusted", {"x": 1})
+
+ assert result.shadow_action == "experimental"
+ assert result.agreement is False
+ assert len(calls) == 1
\ No newline at end of file
diff --git a/synapdrive_ai/tests/test_v1_signing.py b/synapdrive_ai/tests/test_v1_signing.py
new file mode 100644
index 0000000..b032dbc
--- /dev/null
+++ b/synapdrive_ai/tests/test_v1_signing.py
@@ -0,0 +1,29 @@
+from synapdrive_ai.assurance.signing import Ed25519EvidenceSigner, verify_ed25519
+
+
+def test_ed25519_sign_and_verify(tmp_path):
+ signer = Ed25519EvidenceSigner.generate()
+ public = tmp_path / "public.pem"
+ signer.save_public_pem(public)
+ payload = b"evidence"
+ signature = signer.sign(payload)
+ assert verify_ed25519(public.read_bytes(), payload, signature) is True
+
+
+def test_ed25519_detects_tamper(tmp_path):
+ signer = Ed25519EvidenceSigner.generate()
+ public = tmp_path / "public.pem"
+ signer.save_public_pem(public)
+ signature = signer.sign(b"evidence")
+ assert verify_ed25519(public.read_bytes(), b"tampered", signature) is False
+
+
+def test_private_key_roundtrip(tmp_path):
+ signer = Ed25519EvidenceSigner.generate()
+ private = tmp_path / "private.pem"
+ public = tmp_path / "public.pem"
+ signer.save_private_pem(private)
+ signer.save_public_pem(public)
+ loaded = Ed25519EvidenceSigner.load_private_pem(private)
+ signature = loaded.sign(b"abc")
+ assert verify_ed25519(public.read_bytes(), b"abc", signature) is True
diff --git a/synapdrive_ai/tests/test_web_dashboard.py b/synapdrive_ai/tests/test_web_dashboard.py
index f862a3f..266efa5 100644
--- a/synapdrive_ai/tests/test_web_dashboard.py
+++ b/synapdrive_ai/tests/test_web_dashboard.py
@@ -1,44 +1,57 @@
from synapdrive_ai.interface import web_dashboard
-def test_dashboard_home_route():
+def test_dashboard_home_route() -> None:
client = web_dashboard.app.test_client()
- r = client.get("/")
- assert r.status_code == 200
- assert b"SynapDrive-AI Dashboard" in r.data
+ response = client.get("/")
+ assert response.status_code == 200
+ assert b"SynapDrive-AI Dashboard" in response.data
-def test_dashboard_text_api():
+def test_dashboard_text_api() -> None:
client = web_dashboard.app.test_client()
- r = client.post("/api/run/text", json={"text": "stop", "image": "hazard"})
- assert r.status_code == 200
- payload = r.get_json()
+ response = client.post("/api/run/text", json={"text": "stop", "image": "hazard"})
+ assert response.status_code == 200
+ payload = response.get_json()
assert "status" in payload
assert "result" in payload
assert payload["result"]["status"] in {"success", "blocked"}
-def test_dashboard_log_api():
+def test_dashboard_log_api() -> None:
client = web_dashboard.app.test_client()
- # generate one action so log is non-empty
client.post("/api/run/text", json={"text": "move left", "image": "road"})
- r = client.get("/api/log")
- assert r.status_code == 200
- payload = r.get_json()
+ response = client.get("/api/log")
+ assert response.status_code == 200
+ payload = response.get_json()
assert "count" in payload
assert "log" in payload
assert isinstance(payload["log"], list)
-def test_dashboard_assurance_api():
+def test_dashboard_assurance_api() -> None:
client = web_dashboard.app.test_client()
client.post("/api/run/text", json={"text": "move left", "image": "road"})
- r = client.get("/api/assurance")
- assert r.status_code == 200
- payload = r.get_json()
+ response = client.get("/api/assurance")
+ assert response.status_code == 200
+ payload = response.get_json()
assert "report" in payload
assert "receipts" in payload
assert payload["report"]["total_cycles"] >= 1
assert isinstance(payload["receipts"], list)
+
+
+def test_dashboard_rejects_non_object_json() -> None:
+ client = web_dashboard.app.test_client()
+ response = client.post("/api/run/text", json=["move left"])
+ assert response.status_code == 400
+ assert response.get_json()["error"] == "invalid-request"
+
+
+def test_dashboard_unknown_api_route_is_404() -> None:
+ client = web_dashboard.app.test_client()
+ response = client.get("/api/does-not-exist")
+ assert response.status_code == 404
+ assert response.get_json()["error"] == "not-found"
diff --git a/synapdrive_ai/vision/__init__.py b/synapdrive_ai/vision/__init__.py
new file mode 100644
index 0000000..bcca85f
--- /dev/null
+++ b/synapdrive_ai/vision/__init__.py
@@ -0,0 +1 @@
+"""Declared visual-context helpers with explicit provenance boundaries."""
diff --git a/synapdrive_ai/vision/visual_inference.py b/synapdrive_ai/vision/visual_inference.py
index 2456936..457fd94 100644
--- a/synapdrive_ai/vision/visual_inference.py
+++ b/synapdrive_ai/vision/visual_inference.py
@@ -1,13 +1,46 @@
-import random
+from __future__ import annotations
+from dataclasses import dataclass
+from typing import Dict
-class VisualInferenceEngine:
+
+@dataclass(frozen=True)
+class VisualContext:
+ """Declared visual context, not computer-vision inference.
+
+ The current repository does not ship an image model. A caller may provide a label
+ such as ``road`` or ``hazard`` to exercise context-aware control logic. The returned
+ certainty therefore describes whether the declaration mapped to a known vocabulary,
+ not confidence produced by a perception model.
"""
- Simulates basic vision processing for real-world awareness integration.
- Can be expanded with real models for robotics vision or camera inputs.
+
+ visual_tag: str
+ recognized: bool
+ evidence_kind: str = "declared-label"
+ model_used: bool = False
+
+ @property
+ def certainty(self) -> float:
+ return 1.0 if self.recognized else 0.0
+
+ def to_dict(self) -> Dict[str, object]:
+ return {
+ "visual_tag": self.visual_tag,
+ "certainty": self.certainty,
+ "recognized": self.recognized,
+ "evidence_kind": self.evidence_kind,
+ "model_used": self.model_used,
+ }
+
+
+class VisualInferenceEngine:
+ """Backward-compatible mapper for caller-declared visual labels.
+
+ The historical class name is retained so existing imports keep working. This class
+ performs no image inference and intentionally contains no pseudo-random confidence.
"""
- def __init__(self):
+ def __init__(self) -> None:
self.categories = {
"road": "navigation_path",
"person": "human_detected",
@@ -16,20 +49,8 @@ def __init__(self):
"none": "no_visual_target",
}
- def infer(self, image_label):
- """
- Simulates classification of a visual frame by label input.
-
- Args:
- image_label (str): Simplified label like "road", "person", etc.
-
- Returns:
- dict: {
- 'visual_tag': str,
- 'certainty': float
- }
- """
- tag = self.categories.get(image_label, "unknown_visual")
- certainty = round(random.uniform(0.7, 0.99), 2) if tag != "unknown_visual" else 0.4
-
- return {"visual_tag": tag, "certainty": certainty}
+ def infer(self, image_label: str | None) -> Dict[str, object]:
+ normalized = (image_label or "").strip().lower()
+ if normalized in self.categories:
+ return VisualContext(self.categories[normalized], True).to_dict()
+ return VisualContext("unknown_visual", False).to_dict()