tools.py

"""The nine public tools. Each returns JSON text. No orders and no state/ writes."""

from __future__ import annotations

import csv
import json
import math
import os
import tempfile
from pathlib import Path
from typing import Any

from aureon_public import paths
from aureon_public.guard_loader import aureon_source
from aureon_public.quiet import quiet
from aureon_public.text import AURIS_FOUR, HOLD_LINE

_AURIS_FIELDS = ("volatility", "momentum", "volume", "spread")


def _plain(value: Any) -> Any:
    if isinstance(value, dict):
        return {str(key): _plain(item) for key, item in value.items()}
    if isinstance(value, (list, tuple)):
        return [_plain(item) for item in value]
    if isinstance(value, bool) or value is None or isinstance(value, str):
        return value
    if isinstance(value, int):
        return int(value)
    if isinstance(value, float):
        return float(value)
    item = getattr(value, "item", None)
    if callable(item):
        try:
            return _plain(item())
        except Exception:
            pass
    return str(value)


_OMITTED_PUBLIC_KEYS = ("consciousness_level", "consciousness_psi")


def _omit_public_keys(body: dict[str, Any]) -> dict[str, Any]:
    for key in _OMITTED_PUBLIC_KEYS:
        body.pop(key, None)
    return body


def _strip_consciousness(value: Any) -> Any:
    if isinstance(value, dict):
        return {
            str(key): _strip_consciousness(item)
            for key, item in value.items()
            if not str(key).startswith("consciousness_")
        }
    if isinstance(value, list):
        return [_strip_consciousness(item) for item in value]
    return value


def _dumps(payload: dict[str, Any]) -> str:
    body = _strip_consciousness(_plain(payload))
    body["aureon_source"] = aureon_source()
    return json.dumps(body, ensure_ascii=False, allow_nan=False, sort_keys=True)


def _load(path: Path) -> dict[str, Any]:
    loaded = json.loads(path.read_text(encoding="utf-8"))
    if not isinstance(loaded, dict):
        raise TypeError(path.name)
    return loaded


def _finite(value: object) -> float | None:
    if isinstance(value, bool) or value is None:
        return None
    if not isinstance(value, (int, float)):
        return None
    number = float(value)
    if not math.isfinite(number):
        return None
    return number


def _brain() -> dict[str, Any]:
    return _load(paths.brain_json())


def _path_block(brain: dict[str, Any], enable_multiverse: bool) -> dict[str, Any]:
    hnc = brain.get("hnc")
    if not isinstance(hnc, dict):
        raise KeyError("hnc")
    if enable_multiverse:
        block = hnc.get("multiverse_extension")
    else:
        block = hnc
    if not isinstance(block, dict):
        raise KeyError("brain path")
    return block


def _cap(enable_multiverse: bool) -> int:
    steps = _path_block(_brain(), enable_multiverse).get("steps")
    if isinstance(steps, bool) or not isinstance(steps, int) or steps < 1:
        raise ValueError("committed step count is missing")
    return steps


def _hnc_doc() -> str:
    brain = _brain()
    on = json.dumps(_path_block(brain, True)["coherence"])
    off = json.dumps(_path_block(brain, False)["coherence"])
    return f"""Call HarmonicRealityField.step. Label: inferred. This is a rerun in this process.

The class default HarmonicRealityField() sets enable_multiverse=True (multiverse on). This tool's default is that class default. The output field enable_multiverse is the value used for this call.

Do not mix the two committed brain paths. Cited, huggingface/hnc-field/replay/brain.json: multiverse on coherence is {on}; multiverse off coherence is {off}.

steps is capped at the committed steps field for the selected path. There is no frequencies argument. Trade readiness: HOLD. {HOLD_LINE}

Args:
    alpha: Observer gain passed to HarmonicRealityField. Label: cited, DEFAULT_ALPHA in aureon/harmonic/aureon_harmonic_reality.py.
    beta: Echo gain passed to HarmonicRealityField. Label: cited, DEFAULT_BETA.
    tau: Echo delay passed to HarmonicRealityField. Label: cited, DEFAULT_TAU.
    g: Observer nonlinearity passed to HarmonicRealityField. Label: cited, DEFAULT_G.
    delta_t: Observer window passed to HarmonicRealityField. Label: cited, DEFAULT_DELTA_T.
    sample_rate: Samples per second. Signature default 1000.0. Label: cited, HarmonicRealityField.__init__.
    enable_multiverse: Class default is True (multiverse on). False selects the other committed brain path. Reported in the output. Label: cited.
    steps: Step count. Omitted uses the committed steps for the selected path. Values above that cap are unavailable. Label: cited, vendored brain.json steps.
"""


@quiet
def hnc_reality_step(
    alpha: float = 0.85,
    beta: float = 0.9,
    tau: float = 0.05,
    g: float = 1.5,
    delta_t: float = 0.01,
    sample_rate: float = 1000.0,
    enable_multiverse: bool = True,
    steps: int | None = None,
) -> str:
    turned_on = enable_multiverse is True
    if not isinstance(enable_multiverse, bool):
        return _dumps(
            {
                "class_default_enable_multiverse": True,
                "enable_multiverse": None,
                "label": "cited",
                "reason": "enable_multiverse is missing",
                "status": "unavailable",
            }
        )
    try:
        cap = _cap(turned_on)
    except (OSError, ValueError, KeyError, TypeError):
        return _dumps(
            {
                "class_default_enable_multiverse": True,
                "enable_multiverse": turned_on,
                "label": "cited",
                "reason": "The committed step count is missing.",
                "status": "unavailable",
            }
        )
    chosen: int | None
    if steps is None:
        chosen = cap
    elif isinstance(steps, bool) or not isinstance(steps, int) or steps < 1:
        chosen = None
    elif steps > cap:
        return _dumps(
            {
                "class_default_enable_multiverse": True,
                "enable_multiverse": turned_on,
                "label": "cited. The cap is the steps field on the selected brain path.",
                "reason": "steps is above the committed replay length for this path",
                "status": "unavailable",
                "steps_cap": cap,
            }
        )
    else:
        chosen = steps
    numbers = {
        "alpha": _finite(alpha),
        "beta": _finite(beta),
        "tau": _finite(tau),
        "g": _finite(g),
        "delta_t": _finite(delta_t),
        "sample_rate": _finite(sample_rate),
    }
    if (
        chosen is None
        or any(value is None for value in numbers.values())
        or numbers["sample_rate"] is None
        or numbers["sample_rate"] <= 0
    ):
        return _dumps(
            {
                "class_default_enable_multiverse": True,
                "enable_multiverse": turned_on,
                "label": "cited",
                "reason": "A parameter is missing.",
                "status": "unavailable",
                "steps_cap": cap,
            }
        )
    from aureon.harmonic.aureon_harmonic_reality import HarmonicRealityField

    field = HarmonicRealityField(
        alpha=float(numbers["alpha"]),
        beta=float(numbers["beta"]),
        tau=float(numbers["tau"]),
        g=float(numbers["g"]),
        delta_t=float(numbers["delta_t"]),
        sample_rate=float(numbers["sample_rate"]),
        enable_multiverse=turned_on,
    )
    series = [float(field.step()) for _ in range(chosen)]
    payload: dict[str, Any] = {
        "class_default_enable_multiverse": True,
        "coherence": float(field.coherence),
        "committed_replay_match": None,
        "committed_replay_path": "multiverse_extension" if turned_on else "hnc",
        "enable_multiverse": bool(field.enable_multiverse),
        "get_state": field.get_state(),
        "label": "inferred. Rerun of HarmonicRealityField.step in this process.",
        "lambda_last": series[-1],
        "series": series,
        "state_name": field.state.name,
        "state_value": field.state.value,
        "status": "inferred",
        "steps": chosen,
        "steps_cap": cap,
    }
    payload["committed_replay_match"] = _committed_match(
        payload,
        {key: float(value) for key, value in numbers.items()},
        turned_on,
    )
    return _dumps(payload)


def _committed_match(payload: dict[str, Any], numbers: dict[str, float], enable_multiverse: bool) -> bool | None:
    brain = _brain()
    parameters = brain.get("hnc", {}).get("parameters")
    if not isinstance(parameters, dict):
        return None
    expected = {
        "alpha": parameters.get("alpha"),
        "beta": parameters.get("beta"),
        "delta_t": parameters.get("delta_t"),
        "g": parameters.get("g"),
        "sample_rate": parameters.get("sample_rate"),
        "tau": parameters.get("tau"),
    }
    if any(numbers[key] != expected[key] for key in expected):
        return None
    block = _path_block(brain, enable_multiverse)
    if payload["steps"] != block.get("steps"):
        return None
    fresh = {
        "coherence": payload["coherence"],
        "lambda_last": payload["lambda_last"],
        "series": payload["series"],
    }
    rounded = json.loads(json.dumps(fresh))
    return (
        rounded["coherence"] == block.get("coherence")
        and rounded["lambda_last"] == block.get("lambda_last")
        and rounded["series"] == block.get("series")
        and payload["state_value"] == block.get("state")
    )


hnc_reality_step.__doc__ = _hnc_doc()


@quiet
def hnc_canonical_read(envelope_jsonl: str | None = None) -> str:
    """Read CanonicalField. An empty envelope does not touch state/.

    The public JSON drops consciousness_level and consciousness_psi. Those fields have defaults on CanonicalField, so CanonicalField(**public_dict) still constructs and does not raise. It is not a full to_dict() round-trip: the rebuilt fields are the defaults, not whatever to_dict() would have carried.

    Label: inferred when an envelope was supplied and available is true. Otherwise the field is unavailable. Cited: read_canonical_field does not run unless a temp trace was set. The bundled brain replay is a formula record, not a canonical envelope.

    Args:
        envelope_jsonl: Optional JSONL text for one canonical envelope. Empty or omitted returns the unavailable field and does not read state/. Label: cited, aureon/core/hnc_field.py read_canonical_field.
    """

    if envelope_jsonl is None or not str(envelope_jsonl).strip():
        from aureon.core.hnc_field import CanonicalField

        body = CanonicalField().to_dict()
        body["label"] = (
            "cited. No envelope was supplied, so state/ was not read. "
            "The bundled brain replay is a formula record, not a canonical envelope."
        )
        body["status"] = "unavailable"
        return _dumps(_omit_public_keys(body))
    return _dumps(_omit_public_keys(_with_envelope(str(envelope_jsonl), blend=False)))


@quiet
def hnc_blend_read(envelope_jsonl: str | None = None) -> str:
    """Read BlendedField. Does not call publish_subfield. An empty envelope does not touch state/.

    BlendedField.to_dict does not include consciousness_level or consciousness_psi. The same two keys are dropped if they appear. That drop does not change a BlendedField round-trip.

    Label: inferred when an envelope was supplied and available is true. Otherwise unavailable.

    Args:
        envelope_jsonl: Optional JSONL text. Empty or omitted returns the unavailable blend and does not read state/. Label: cited, aureon/core/hnc_field.py blend_field.
    """

    if envelope_jsonl is None or not str(envelope_jsonl).strip():
        from aureon.core.hnc_field import BlendedField

        body = BlendedField().to_dict()
        body["label"] = "cited. No envelope was supplied, so state/ was not read."
        body["status"] = "unavailable"
        return _dumps(_omit_public_keys(body))
    return _dumps(_omit_public_keys(_with_envelope(str(envelope_jsonl), blend=True)))


def _with_envelope(text: str, *, blend: bool) -> dict[str, Any]:
    from aureon.core.hnc_field import blend_field, read_canonical_field

    previous_trace = os.environ.get("AUREON_HNC_TRACE_PATH")
    previous_bus = os.environ.get("AUREON_BUS_TRACE_DIR")
    directory = Path(tempfile.mkdtemp(prefix="aureon-public-field-"))
    trace = directory / "envelope.jsonl"
    bus_dir = directory / "bus"
    bus_dir.mkdir()
    try:
        payload = text if text.endswith("\n") else text + "\n"
        trace.write_text(payload, encoding="utf-8")
        os.environ["AUREON_HNC_TRACE_PATH"] = str(trace)
        os.environ["AUREON_BUS_TRACE_DIR"] = str(bus_dir)
        field = read_canonical_field(bus=object())
        if blend:
            body = blend_field(bus=object()).to_dict()
        else:
            body = field.to_dict()
    finally:
        if previous_trace is None:
            os.environ.pop("AUREON_HNC_TRACE_PATH", None)
        else:
            os.environ["AUREON_HNC_TRACE_PATH"] = previous_trace
        if previous_bus is None:
            os.environ.pop("AUREON_BUS_TRACE_DIR", None)
        else:
            os.environ["AUREON_BUS_TRACE_DIR"] = previous_bus
        trace.unlink(missing_ok=True)
        for child in bus_dir.glob("*"):
            child.unlink(missing_ok=True)
        bus_dir.rmdir()
        directory.rmdir()
    body["label"] = "inferred. Read from the envelope supplied to this process."
    body["status"] = "inferred" if body.get("available") is True else "unavailable"
    return body


@quiet
def lambda_engine_step(
    volatility: float | None = None,
    readings: list[dict[str, Any]] | None = None,
) -> str:
    """Run LambdaEngine.step, or return status unavailable.

    This is not HarmonicRealityField and it is not the committed brain replay. The committed replay ships no readings. A missing or short reading list returns status unavailable and does not fill lambda or coherence with 0. Auris-to-Lambda link: HOLD.

    Args:
        volatility: Finite volatility. Omitted is unavailable, not 0. Label: cited, LambdaEngine.step.
        readings: Objects with name, value, confidence, and state. Fewer than two returns status unavailable. Label: cited, huggingface/hnc-field/app.py. Auris-to-Lambda link: HOLD.
    """

    parsed, reason = _parse_readings(readings)
    vol = _finite(volatility)
    if parsed is None or vol is None:
        why = reason or "volatility is missing"
        return _dumps(
            {
                "auris_to_lambda": "HOLD",
                "label": (
                    "cited. The committed brain replay ships no LambdaEngine readings. "
                    "This is not HarmonicRealityField. Auris-to-Lambda link: HOLD."
                ),
                "lambda_t": None,
                "reason": why,
                "status": "unavailable",
            }
        )
    fd, name = tempfile.mkstemp(prefix="aureon-public-lambda-", suffix=".json")
    os.close(fd)
    path = Path(name)
    try:
        from aureon.core.aureon_lambda_engine import LambdaEngine, SubsystemReading

        engine = LambdaEngine(state_path=path)
        state = engine.step(parsed_readings(parsed, SubsystemReading), volatility=vol, auto_persist=False)
        body = state.to_dict()
    finally:
        path.unlink(missing_ok=True)
    body["auris_to_lambda"] = "HOLD"
    body["label"] = (
        "inferred. LambdaEngine.step on the supplied readings. "
        "Not HarmonicRealityField. Not the committed brain replay. "
        "Field names are the keys LambdaState.to_dict already returns. "
        "Auris-to-Lambda link: HOLD."
    )
    body["status"] = "inferred"
    return _dumps(body)


def parsed_readings(rows: list[dict[str, Any]], reading_type: Any) -> list[Any]:
    return [
        reading_type(
            name=row["name"],
            value=row["value"],
            confidence=row["confidence"],
            state=row["state"],
        )
        for row in rows
    ]


def _parse_readings(readings: list[dict[str, Any]] | None) -> tuple[list[dict[str, Any]] | None, str | None]:
    if readings is None:
        return None, "LambdaEngine path: unavailable. No readings were supplied."
    if not isinstance(readings, list) or len(readings) < 2:
        return None, "LambdaEngine path: unavailable. At least two readings are required."
    parsed: list[dict[str, Any]] = []
    for row in readings:
        if not isinstance(row, dict):
            return None, "LambdaEngine path: unavailable. A reading is missing fields."
        name = row.get("name")
        value = _finite(row.get("value"))
        confidence = _finite(row.get("confidence"))
        state = row.get("state")
        if not isinstance(name, str) or not name.strip() or value is None or confidence is None:
            return None, "LambdaEngine path: unavailable. A reading is missing fields."
        if not isinstance(state, str) or not state.strip():
            return None, "LambdaEngine path: unavailable. A reading is missing fields."
        parsed.append(
            {
                "name": name.strip(),
                "value": value,
                "confidence": confidence,
                "state": state.strip(),
            }
        )
    return parsed, None


def _texture(
    volatility: float | None,
    momentum: float | None,
    volume: float | None,
    spread: float | None,
) -> dict[str, float] | None:
    supplied: dict[str, float] = {}
    for key, raw in (
        ("volatility", volatility),
        ("momentum", momentum),
        ("volume", volume),
        ("spread", spread),
    ):
        if raw is None:
            continue
        number = _finite(raw)
        if number is None:
            continue
        supplied[key] = number
    return supplied or None


def _queen() -> Any:
    from aureon.utils.aureon_queen_hive_mind import QueenHiveMind

    return QueenHiveMind.__new__(QueenHiveMind)


@quiet
def auris_nodes(
    volatility: float | None = None,
    momentum: float | None = None,
    volume: float | None = None,
    spread: float | None = None,
) -> str:
    """Nine nodes. Real function: QueenHiveMind.read_auris_nodes.

    This tool is the nine nodes. It is not the conjecture (not called), not the persona (not scored), and not the coherence evaluator. The coherence evaluator is auris_coherence. Only the nine nodes and get_auris_coherence are real functions exposed here.

    A missing field is omitted. Nothing is filled with zero. Label: inferred.

    Args:
        volatility: Optional Tiger and related input. Omitted stays omitted. Label: cited, _auris_input.
        momentum: Optional Falcon and Dolphin input. Omitted stays omitted. Label: cited, _auris_input.
        volume: Optional Falcon and CargoShip input. Omitted stays omitted. Label: cited, _auris_input.
        spread: Optional Deer input. Omitted stays omitted. Label: cited, _auris_input.
    """

    market = _texture(volatility, momentum, volume, spread)
    nodes = _queen().read_auris_nodes(market)
    missing = sorted(name for name, row in nodes.items() if row.get("value") is None)
    return _dumps(
        {
            "auris": "nine nodes",
            "label": "inferred. QueenHiveMind.read_auris_nodes. A missing field is not filled.",
            "missing_nodes": missing,
            "nodes": nodes,
            "which_auris": AURIS_FOUR,
        }
    )


@quiet
def auris_coherence(
    volatility: float | None = None,
    momentum: float | None = None,
    volume: float | None = None,
    spread: float | None = None,
) -> str:
    """Coherence evaluator. Real function: QueenHiveMind.get_auris_coherence.

    This tool is the coherence evaluator. It is not the conjecture (not called), not the nine nodes (those are read_auris_nodes and AURIS_NODES on auris_nodes), and not the persona (not scored). Only the nine nodes and get_auris_coherence are real functions exposed here.

    Whenever any of the nine nodes has no reading, status is no_data and coherence is null. The missing node names are listed. Nothing is filled with zero. Auris-to-Lambda link: HOLD. Trade readiness: HOLD.

    Args:
        volatility: Optional. Omitted is not filled with zero. Label: cited, _auris_input.
        momentum: Optional. Omitted is not filled with zero. Label: cited, _auris_input.
        volume: Optional. Omitted is not filled with zero. Label: cited, _auris_input.
        spread: Optional. Omitted is not filled with zero. Label: cited, _auris_input.
    """

    market = _texture(volatility, momentum, volume, spread)
    queen = _queen()
    nodes = queen.read_auris_nodes(market)
    coherence, status = queen.get_auris_coherence(market)
    missing = sorted(name for name, row in nodes.items() if row.get("value") is None)
    if missing or coherence is None:
        return _dumps(
            {
                "auris": "coherence evaluator",
                "auris_to_lambda": "HOLD",
                "coherence": None,
                "label": "inferred. get_auris_coherence returned no complete nine-node reading. Auris-to-Lambda link: HOLD.",
                "missing_nodes": missing,
                "status": "no_data",
                "trade_readiness": "HOLD",
                "which_auris": AURIS_FOUR,
            }
        )
    return _dumps(
        {
            "auris": "coherence evaluator",
            "auris_to_lambda": "HOLD",
            "coherence": coherence,
            "label": "inferred. This is the evaluator return, not a track record. Auris-to-Lambda link: HOLD.",
            "missing_nodes": missing,
            "status": status,
            "trade_readiness": "HOLD",
            "which_auris": AURIS_FOUR,
        }
    )


def _lambda_column() -> list[float]:
    path = paths.qgita_series()
    values: list[float] = []
    with path.open(encoding="utf-8", newline="") as handle:
        reader = csv.DictReader(handle)
        if reader.fieldnames is None or "Lambda" not in reader.fieldnames:
            return []
        for row in reader:
            try:
                values.append(float(row["Lambda"]))
            except (TypeError, ValueError):
                continue
    return values


def _event_record(item: Any, kind: str) -> dict[str, Any]:
    if kind == "ftcp":
        return {
            "adapter": "QGITAMarketAnalyzer.analyze",
            "curvature": float(item.curvature),
            "is_valid": bool(item.is_valid),
            "kind": "FTCP",
            "knot_index": int(item.knot_index),
            "timestamp": float(item.timestamp),
        }
    return {
        "adapter": "QGITAMarketAnalyzer.analyze",
        "event_type": str(item.event_type.value),
        "kind": "lighthouse",
        "lighthouse_intensity": float(item.lighthouse_intensity),
        "timestamp": float(item.timestamp),
    }


@quiet
def qgita_analyze(
    values: list[float] | None = None,
    timestamps: list[float] | None = None,
) -> str:
    """Run QGITAMarketAnalyzer.analyze and drop the signals object.

    Events are the adapter return, not a detection rate. Detection rate: HOLD. False-alarm rate: HOLD. The bundled series is cited-unverified and is not public market data.

    Args:
        values: Optional series. Omitted loads the vendored de440 Lambda column. A supplied series is inferred and is not the committed replay. Length must be at least 10. Label: cited for the file, inferred for a pasted series.
        timestamps: Optional clocks, one per value. Omitted uses the sample index. Label: cited, the Space clock.
    """

    supplied = values is not None
    series = list(values) if supplied else _lambda_column()
    if supplied:
        cleaned: list[float] = []
        for item in series:
            number = _finite(item)
            if number is None:
                return _dumps(
                    {
                        "detection_rate": "HOLD",
                        "false_alarm_rate": "HOLD",
                        "label": "inferred",
                        "reason": "A value is missing.",
                        "status": "unavailable",
                    }
                )
            cleaned.append(number)
        series = cleaned
    if len(series) < 10:
        return _dumps(
            {
                "detection_rate": "HOLD",
                "false_alarm_rate": "HOLD",
                "label": "cited. analyze returns insufficient_data below 10 samples.",
                "reason": "At least ten values are required.",
                "status": "unavailable",
            }
        )
    if timestamps is None:
        clocks = [float(index) for index in range(len(series))]
        clock_label = "cited. Sample index, the Space clock."
    else:
        if len(timestamps) != len(series):
            return _dumps(
                {
                    "detection_rate": "HOLD",
                    "false_alarm_rate": "HOLD",
                    "label": "inferred",
                    "reason": "timestamps and values differ in length.",
                    "status": "unavailable",
                }
            )
        clocks = []
        for item in timestamps:
            number = _finite(item)
            if number is None:
                return _dumps(
                    {
                        "detection_rate": "HOLD",
                        "false_alarm_rate": "HOLD",
                        "label": "inferred",
                        "reason": "A timestamp is missing.",
                        "status": "unavailable",
                    }
                )
            clocks.append(number)
        clock_label = "inferred. Caller-supplied clocks."
    from aureon.wisdom.aureon_qgita_framework import QGITAMarketAnalyzer

    analyzer = QGITAMarketAnalyzer()
    for index, value in enumerate(series):
        analyzer.feed_price(float(value), timestamp=clocks[index])
    raw = analyzer.analyze()
    stage1 = raw.get("stage1") if isinstance(raw.get("stage1"), dict) else {}
    stage2 = raw.get("stage2") if isinstance(raw.get("stage2"), dict) else {}
    valid = list(stage1.get("valid_ftcps") or [])
    events = list(stage2.get("lighthouse_events") or [])
    body: dict[str, Any] = {
        "clock": clock_label,
        "coherence": raw.get("coherence"),
        "detection_rate": "HOLD",
        "false_alarm_rate": "HOLD",
        "ftcp_count": stage1.get("ftcp_count"),
        "ftcps": [_event_record(item, "ftcp") for item in valid],
        "label": (
            "cited-unverified bundled series. Not public market data."
            if not supplied
            else "inferred. Adapter run on the supplied series. Not the committed replay."
        ),
        "lhe_count": stage2.get("lhe_count"),
        "lighthouse_events": [_event_record(item, "lhe") for item in events],
        "regime": raw.get("regime"),
        "samples_fed": len(series),
        "status": raw.get("status", "unavailable"),
    }
    return _dumps(body)


@quiet
def organism_replay_read() -> str:
    """Read the committed circulation replay file. This does not start a seat.

    Connectivity figures are cited from the file. The fault block is seeded synthetic input, not telemetry. A claim that those faults are recorded seat faults is HOLD.

    Args:
        none: This tool takes no inputs. Label: cited, huggingface/organism-replay/replay/organism.json.
    """

    record = _load(paths.organism_json())
    connectivity = record.get("connectivity") if isinstance(record.get("connectivity"), dict) else {}
    self_model = record.get("self_model") if isinstance(record.get("self_model"), dict) else {}
    prediction = self_model.get("prediction") if isinstance(self_model.get("prediction"), dict) else {}
    fault = self_model.get("fault") if isinstance(self_model.get("fault"), dict) else {}
    seats = connectivity.get("seats") if isinstance(connectivity.get("seats"), dict) else {}
    isolated = connectivity.get("isolated")
    topics = _dashboard_topics()
    return _dumps(
        {
            "cited": {
                "connected": connectivity.get("connected"),
                "connectivity_score": connectivity.get("score"),
                "http_seat_bound": connectivity.get("http_seat_bound"),
                "isolated": isolated,
                "label": "cited, huggingface/organism-replay/replay/organism.json",
                "prediction_accuracy": prediction.get("accuracy"),
                "prediction_baseline_accuracy": prediction.get("baseline_accuracy"),
                "roster": connectivity.get("roster"),
                "seats": seats,
                "verdict": connectivity.get("verdict"),
            },
            "dashboard_topics": topics,
            "label": "cited replay file. Circulation was not started.",
            "seeded_synthetic_fault": {
                "detection_rate_in_file": fault.get("detection_rate"),
                "false_alarm_rate_in_file": fault.get("false_alarm_rate"),
                "hold": (
                    "A claim that these are recorded seat faults is HOLD. "
                    "These figures are not a QGITA detection rate and not telemetry."
                ),
                "label": "Seeded synthetic input, not telemetry.",
                "origin": fault.get("origin"),
                "seed": fault.get("seed"),
            },
        }
    )


def _dashboard_topics() -> list[str]:
    topics = (
        "ics.body.state",
        "ics.mind.state",
        "ics.source.state",
        "ics.soul.state",
        "ics.hnc.state",
        "ics.monologue",
    )
    try:
        from aureon.core.aureon_thought_bus import ThoughtBus
        from aureon.core.cognitive_dashboard import CognitiveDashboard

        bus = ThoughtBus(persist_path=None)
        CognitiveDashboard(thought_bus=bus)
    except Exception:
        return [f"Dashboard topic {topic}: unavailable" for topic in topics]
    lines = []
    for topic in topics:
        found = bus.recall(topic, limit=1)
        if found:
            lines.append(f"Dashboard topic {topic}: present in the replay")
        else:
            lines.append(f"Dashboard topic {topic}: unavailable")
    return lines


@quiet
def organism_catalog() -> str:
    """List circulation seats, algorithmic nodes, and deterministic systems.

    These are names in source. This tool does not start circulation, self-repair, or self-code. Label: cited.

    Args:
        none: This tool takes no inputs. Label: cited, circulation_seats, list_nodes, and list_systems.
    """

    from aureon.core.algorithmic_nodes import list_nodes
    from aureon.core.deterministic_systems import list_systems
    from aureon.core.organism_circulation import circulation_seats

    return _dumps(
        {
            "label": "cited. Names in source. Seats are not running.",
            "nodes": list_nodes(),
            "seats": list(circulation_seats()),
            "systems": list_systems(),
        }
    )