diff --git a/CHANGELOG.md b/CHANGELOG.md
index 0b5eca6..49d248c 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -13,6 +13,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- **Break observer trajectory cycle detection and loop governance ([#93](https://github.com/ARPAHLS/aura/issues/93))** — enhanced `BreakObserver` (`preset: break`) with sliding-window trajectory cycle detection: 1-cycle action fixation ($A \to A \to A$), 2-cycle alternating state oscillation ($A \leftrightarrow B \leftrightarrow A \leftrightarrow B$), 3-cycle execution choreography ($A \to B \to C \to A \to B \to C$), error status fingerprint churn (consecutive identical error codes/messages), and reasoning inertia text similarity sensor (>85% SequenceMatcher ratio across consecutive turn thoughts); FinOps loop burn-velocity guard (`burn_velocity_guard`, `max_cycle_token_waste`) tracking wasted token churn during runaway cycles; non-repudiable cryptographic proof hashes (`cycle_entry_hash`, `cycle_recurrence_hash`) captured from the immutable SHA-256 audit spine conforming to EU AI Act Article 14 human oversight standards; actionable `remedial_directive` steering strings in `observer.alert` payloads; full integration with `profile.escalations[]` (`on: observer.alert`); audit report scorecard `scorecard.observers = {alerts, loop_cycles}` and finding `COGNITIVE_CYCLE_DETECTED` (`severity: warn`); optional Tier 2 `mode: "watchdog"` with pluggable callback; dedicated stress simulation `scripts/aura_break_loop_stress_sim.py` and test suite `tests/test_observer_break_loop.py`.
 - **Sequencer middleware and PII masking ([#58](https://github.com/ARPAHLS/aura/issues/58))** — two-tier PII sanitization engine: Tier 1 zero-dependency regex with algorithmic Luhn validation for credit cards, RFC emails, E.164/NANP phones, SSNs, IPv4/IPv6, API keys/JWTs/UUIDs, and IBAN; Tier 2 pluggable `PiiMaskerProvider` protocol for contextual NER models (Presidio, Skillware, ARPA `micro-f1-mask` 270M); masking strategies (`redact`, `mask`, `pseudonymize` / `tokenize`) with deterministic session-scoped pseudonym tokens preserving entity equivalence across multi-step agent reasoning; integration via `MiddlewarePolicy` / `MiddlewareStack` at step boundaries (`scope: per_step`, inbound/outbound) and explicit pipeline step op (`type: op, ref: middleware.pii_mask`); `strict: true` raises `ConstraintViolation` on detection; spine event `middleware.applied` captures detection metrics with zero raw PII written to spine; audit report scorecard `scorecard.middleware = {applied, pii_redactions}` and `PII_MASKED` finding (`severity: info`); dedicated stress simulation `scripts/aura_pii_stress_sim.py` and test suite `tests/test_sequencer_pii_mask.py`.
 - **Capability inject for multiple secrets and HTTP headers ([#86](https://github.com/ARPAHLS/aura/issues/86))** — capability declarations support multi-target secret injection via `inject: [{ref, as, prefix}]` alongside single-secret `secret: {ref, as}` backward compatibility; supports nested dot-paths into `args` (e.g. `aws_access_key_id`) and `headers` (e.g. `headers.Authorization`); optional value `prefix` (e.g. `"Bearer "`); `ToolHost.execute` protocol and `SkillwareHost.execute` forward `headers` to executor callables; agent spoofed header keys stripped at egress; all injected secret values registered for leak scrubbing on `tool.result`; `strict_args` reserves all injected target paths; `InjectTarget` exported from `aura` and `aura.core.capabilities`; CLI parity; stress sim scenarios `capability_multi_secret_dual` and `capability_header_authorization` (18 scenarios total).
 - **Egress schema and constitution checks ([#78](https://github.com/ARPAHLS/aura/issues/78))** — profile `variables.schema_refs` (named JSON Schema specs per tool) and explicit `schema_check` constraint rules validate `tool.call` args and/or `tool.result` payloads at egress; spectrum `high` / `full` auto-wires refs when `schema_enforcement` is not false; audit finding `SCHEMA_VIOLATION`; conformance `schema` check on close; `session.open` and `aura agent show` spectrum summary include `schema` block; `merged_profile_rules()` helper for session/CLI parity; stress sim `schema_high_bind` scenario; depends on `jsonschema`.
diff --git a/aura/agents/profile.py b/aura/agents/profile.py
index ad1ddb1..d73c0db 100644
--- a/aura/agents/profile.py
+++ b/aura/agents/profile.py
@@ -14,6 +14,17 @@ def _capabilities_from_dict(raw: Any) -> list[dict[str, Any]]:
     return capabilities_to_public(caps)
 
 
+def _sanitize_for_json(obj: Any) -> Any:
+    """Recursively convert un-serializable callables into safe descriptors."""
+    if callable(obj):
+        return f"<callable {getattr(obj, '__name__', 'unnamed')}>"
+    if isinstance(obj, dict):
+        return {k: _sanitize_for_json(v) for k, v in obj.items()}
+    if isinstance(obj, list):
+        return [_sanitize_for_json(v) for v in obj]
+    return obj
+
+
 @dataclass
 class AgentProfile:
     """Persistent agent record."""
@@ -48,7 +59,7 @@ def to_dict(self) -> dict[str, Any]:
             "rules": self.rules,
             "skills": self.skills,
             "sequencer": self.sequencer,
-            "observers": self.observers,
+            "observers": _sanitize_for_json(self.observers),
             "escalations": self.escalations,
             "spectrum": self.spectrum,
             "capabilities": self.capabilities,
diff --git a/aura/core/audit_report.py b/aura/core/audit_report.py
index 4975e75..a95ebd0 100644
--- a/aura/core/audit_report.py
+++ b/aura/core/audit_report.py
@@ -99,6 +99,14 @@ def build(
                 "goal_drifts": sum(1 for e in events if e.kind == "conformance.drift"),
                 "schedule_misses": sum(1 for e in events if e.kind == "slo.missed"),
             },
+            "observers": {
+                "alerts": sum(1 for e in events if e.kind == "observer.alert"),
+                "loop_cycles": sum(
+                    1
+                    for e in events
+                    if e.kind == "observer.alert" and (e.payload or {}).get("pattern")
+                ),
+            },
             "events": conformance.event_count,
         }
 
@@ -275,6 +283,27 @@ def build(
                         "event_id": event.event_id,
                     }
                 )
+            if event.kind == "observer.alert":
+                payload = event.payload or {}
+                pattern = payload.get("pattern")
+                if pattern:
+                    findings.append(
+                        {
+                            "severity": "warn",
+                            "code": "COGNITIVE_CYCLE_DETECTED",
+                            "message": payload.get("remedial_directive")
+                            or f"Observer loop cycle detected: {pattern}",
+                            "tool": payload.get("tool"),
+                            "pattern": pattern,
+                            "signature": payload.get("signature"),
+                            "cycle_entry_hash": payload.get("cycle_entry_hash"),
+                            "cycle_recurrence_hash": payload.get("cycle_recurrence_hash"),
+                            "event_id": event.event_id,
+                        }
+                    )
+                    directive = payload.get("remedial_directive")
+                    if directive and directive not in recommendations:
+                        recommendations.append(directive)
 
         open_evt = next((e for e in events if e.kind == "session.open"), None)
         spectrum = (open_evt.payload or {}).get("spectrum") if open_evt else None
diff --git a/aura/observers/presets/break_observer.py b/aura/observers/presets/break_observer.py
index f7b7dcc..8439e85 100644
--- a/aura/observers/presets/break_observer.py
+++ b/aura/observers/presets/break_observer.py
@@ -1,68 +1,442 @@
-"""Break observer preset — circuit-breaker alerts for runaway tool patterns."""
+"""Break observer preset — circuit-breaker alerts & trajectory cycle detector (#93).
+
+Monitors causal spine events to detect:
+1. Repeated Tool Intents: Global threshold count (backward compatible with V0.1).
+2. Action Fixation (1-cycle): Identical tool + normalized args repeating consecutively.
+3. State Oscillation (2-cycle): Alternating actions (A -> B -> A -> B) across turns.
+4. Cyclic Choreography (3-cycle): Repetitive action sequences (A -> B -> C -> A -> B -> C).
+5. Error Fingerprint Churn: Consecutive identical error codes or failure messages.
+6. Reasoning Inertia: Consecutive turns exhibiting high thought text similarity (>85%).
+7. FinOps Burn-Velocity Guard: Churn token tracking and rate alerts.
+8. Non-Repudiable Cryptographic Deadlock Proofs: Captures entry/exit spine hashes.
+9. Optional Watchdog Mode (Tier 2): Pluggable callable/model for semantic loop audit.
+"""
 
 from __future__ import annotations
 
+import difflib
+import hashlib
 import json
-from typing import Any, TYPE_CHECKING
+import uuid
+from dataclasses import dataclass
+from typing import Any, Callable, TYPE_CHECKING
 
 if TYPE_CHECKING:
     from aura.core.session import Session
 
 
+@dataclass
+class ActionRecord:
+    tool: str
+    args: dict[str, Any]
+    sig: str
+    turn: int
+    spine_hash: str | None = None
+    has_error: bool = False
+    error_fingerprint: str = ""
+
+
+@dataclass
+class TurnRecord:
+    turn: int
+    thought: str
+
+
 class BreakObserver:
     """
-    Detect repeated identical tool intents and emit observer.alert on the spine.
+    Detect repeated identical tool intents, trajectory cycles, and cognitive churn.
+    Emits observer.alert on the causal spine when runaway loops occur.
     Does not block egress — host or escalation layer acts on alerts.
     """
 
     def __init__(
-        self, observer_id: str, session: Session, config: dict[str, Any] | None = None
+        self,
+        observer_id: str,
+        session: Session,
+        config: dict[str, Any] | None = None,
     ) -> None:
         self.observer_id = observer_id
         self._session = session
         self._config = dict(config or {})
+
+        # Configuration parameters
+        self.window_size: int = int(self._config.get("window_size", 8))
+        self.max_identical_intents: int = int(self._config.get("max_identical_intents", 5))
+        self.max_identical_actions: int = int(self._config.get("max_identical_actions", 3))
+        self.max_oscillation_cycles: int = int(self._config.get("max_oscillation_cycles", 2))
+        self.max_choreography_cycles: int = int(self._config.get("max_choreography_cycles", 2))
+        self.max_error_churn: int = int(self._config.get("max_error_churn", 3))
+        self.thought_similarity_threshold: float = float(
+            self._config.get("thought_similarity_threshold", 0.85)
+        )
+        self.max_thought_inertia: int = int(self._config.get("max_thought_inertia", 3))
+        self.burn_velocity_guard: bool = bool(self._config.get("burn_velocity_guard", False))
+        self.max_cycle_token_waste: int = int(self._config.get("max_cycle_token_waste", 0))
+
+        self.action: str = str(self._config.get("action", "alert_only")).lower()
+        self.mode: str = str(self._config.get("mode", "default")).lower()
+        self.watchdog_fn: Callable[[list[dict[str, Any]]], dict[str, Any] | None] | None = (
+            self._config.get("watchdog_fn") or self._config.get("watchdog_model")
+        )
+
+        # In-memory sliding trajectory window & counters
+        self._current_turn: int = 1
         self._intent_signatures: dict[str, int] = {}
+        self._action_history: list[ActionRecord] = []
+        self._turn_history: list[TurnRecord] = []
         self._alerts: list[dict[str, Any]] = []
+        self._tripped_dedup_keys: set[str] = set()
+        self._cycle_token_accumulator: int = 0
 
     def on_event(self, event: dict[str, Any]) -> None:
-        if event.get("kind") != "tool.intent":
-            return
+        kind = event.get("kind")
         payload = dict(event.get("payload") or {})
+
+        if kind == "turn.start":
+            self._handle_turn_start(payload)
+        elif kind in ("tool.intent", "tool.call"):
+            self._handle_tool_action(kind, payload)
+        elif kind in ("tool.result", "tool.error"):
+            self._handle_tool_result(payload)
+        elif kind == "turn.end":
+            self._current_turn += 1
+
+    def _handle_turn_start(self, payload: dict[str, Any]) -> None:
+        thought = str(
+            payload.get("thought") or payload.get("reasoning") or payload.get("prompt") or ""
+        ).strip()
+        if not thought:
+            return
+
+        turn_rec = TurnRecord(turn=self._current_turn, thought=thought)
+        self._turn_history.append(turn_rec)
+        if len(self._turn_history) > self.window_size:
+            self._turn_history.pop(0)
+
+        # Check reasoning inertia across recent consecutive turns
+        if len(self._turn_history) >= self.max_thought_inertia:
+            recent_thoughts = [t.thought for t in self._turn_history[-self.max_thought_inertia :]]
+            similar_count = 0
+            for i in range(len(recent_thoughts) - 1):
+                t1, t2 = recent_thoughts[i], recent_thoughts[i + 1]
+                ratio = difflib.SequenceMatcher(None, t1, t2).ratio()
+                if ratio >= self.thought_similarity_threshold:
+                    similar_count += 1
+
+            if similar_count >= (self.max_thought_inertia - 1):
+                directive = (
+                    f"BREAK: Consecutive reasoning steps show >="
+                    f"{int(self.thought_similarity_threshold * 100)}% text similarity "
+                    f"across {self.max_thought_inertia} turns. Re-evaluate strategy or context."
+                )
+                self._emit_alert(
+                    alert_type="loop_cycle_detected",
+                    detail={
+                        "pattern": "REASONING_INERTIA",
+                        "tool": "host.reasoning",
+                        "count": self.max_thought_inertia,
+                        "threshold": self.max_thought_inertia,
+                        "signature": f"thought_inertia_turn_{self._current_turn}",
+                        "remedial_directive": directive,
+                    },
+                )
+
+    def _handle_tool_action(self, kind: str, payload: dict[str, Any]) -> None:
         tool = str(payload.get("tool") or "unknown")
-        args_key = json.dumps(payload.get("args") or {}, sort_keys=True)
+        raw_args = payload.get("args") or {}
+        args_key = self._normalize_args(raw_args)
         sig = f"{tool}:{args_key}"
+
+        # Global intent counting for 100% backward compatibility
         self._intent_signatures[sig] = self._intent_signatures.get(sig, 0) + 1
-        threshold = int(self._config.get("max_identical_intents", 5))
         count = self._intent_signatures[sig]
-        if threshold > 0 and count >= threshold:
+
+        # 1. Legacy global threshold alert
+        if self.max_identical_intents > 0 and count >= self.max_identical_intents:
             self._emit_alert(
-                "repeated_tool_intent",
-                {
+                alert_type="repeated_tool_intent",
+                detail={
                     "tool": tool,
                     "count": count,
-                    "threshold": threshold,
+                    "threshold": self.max_identical_intents,
                     "signature": sig,
+                    "pattern": "REPEATED_INTENT",
+                    "remedial_directive": (
+                        f"BREAK: Repeated identical tool intent '{tool}' "
+                        f"{count} times. Alter parameters or revise approach."
+                    ),
                 },
             )
 
-    def summary(self) -> dict[str, Any]:
-        return {
-            "intent_signatures": dict(self._intent_signatures),
-            "alerts_emitted": len(self._alerts),
-        }
+        # Record action in sliding trajectory window
+        spine = getattr(self._session, "spine", None)
+        last_hash = getattr(spine, "_last_content_hash", None) if spine else None
+
+        action_rec = ActionRecord(
+            tool=tool,
+            args=raw_args if isinstance(raw_args, dict) else {},
+            sig=sig,
+            turn=self._current_turn,
+            spine_hash=last_hash,
+        )
+        self._action_history.append(action_rec)
+        if len(self._action_history) > self.window_size:
+            self._action_history.pop(0)
+
+        # 2. Action Fixation (1-cycle: A -> A -> A consecutively in recent window)
+        self._check_action_fixation(tool, sig)
+
+        # 3. State Oscillation (2-cycle: A -> B -> A -> B)
+        self._check_oscillation_cycle()
+
+        # 4. Choreography Cycle (3-cycle: A -> B -> C -> A -> B -> C)
+        self._check_choreography_cycle()
+
+        # 5. Optional Watchdog Mode (Tier 2 Model/Callable)
+        if self.mode == "watchdog" and callable(self.watchdog_fn):
+            self._check_watchdog_model()
+
+    def _handle_tool_result(self, payload: dict[str, Any]) -> None:
+        tool = str(payload.get("tool") or "")
+        result = payload.get("result")
+        error = payload.get("error")
+        status = payload.get("status")
+
+        # Track tokens for burn-velocity guard
+        tokens = int(
+            payload.get("tokens") or payload.get("token_count") or payload.get("total_tokens") or 0
+        )
+        self._cycle_token_accumulator += tokens
+
+        has_error = False
+        err_msg = ""
+        if error:
+            has_error = True
+            err_msg = str(error)
+        elif isinstance(result, dict) and ("error" in result or result.get("success") is False):
+            has_error = True
+            err_msg = str(result.get("error") or result.get("message") or "failure")
+        elif isinstance(status, (int, str)) and str(status).isdigit() and int(status) >= 400:
+            has_error = True
+            err_msg = f"HTTP_{status}"
+
+        if not has_error:
+            return
+
+        fingerprint = hashlib.sha256(err_msg.encode("utf-8")).hexdigest()[:8]
+        if self._action_history:
+            last_action = self._action_history[-1]
+            last_action.has_error = True
+            last_action.error_fingerprint = fingerprint
+
+        # Check consecutive error status churn
+        recent_errors = [a for a in self._action_history if a.has_error]
+        if len(recent_errors) >= self.max_error_churn:
+            last_fps = [a.error_fingerprint for a in recent_errors[-self.max_error_churn :]]
+            if len(set(last_fps)) == 1:
+                directive = (
+                    f"BREAK: Tool '{tool}' returned identical error fingerprint "
+                    f"({fingerprint}) {self.max_error_churn} consecutive times. "
+                    f"Back off and resolve root cause before retrying."
+                )
+                self._emit_alert(
+                    alert_type="loop_cycle_detected",
+                    detail={
+                        "pattern": "ERROR_STATUS_CHURN",
+                        "tool": tool,
+                        "count": self.max_error_churn,
+                        "threshold": self.max_error_churn,
+                        "signature": f"err_{fingerprint}",
+                        "remedial_directive": directive,
+                    },
+                )
+
+    def _check_action_fixation(self, tool: str, sig: str) -> None:
+        """Detect identical action repeated N times consecutively in window."""
+        if len(self._action_history) < self.max_identical_actions:
+            return
+
+        recent = self._action_history[-self.max_identical_actions :]
+        if all(a.sig == sig for a in recent):
+            directive = (
+                f"BREAK: Repeated identical action '{tool}' "
+                f"{self.max_identical_actions} times consecutively. "
+                f"Alter parameters or revise approach."
+            )
+            self._emit_alert(
+                alert_type="loop_cycle_detected",
+                detail={
+                    "pattern": "ACTION_FIXATION",
+                    "tool": tool,
+                    "count": self.max_identical_actions,
+                    "threshold": self.max_identical_actions,
+                    "signature": sig,
+                    "remedial_directive": directive,
+                },
+            )
+
+    def _check_oscillation_cycle(self) -> None:
+        """Detect alternating 2-state oscillation: A -> B -> A -> B."""
+        needed_length = self.max_oscillation_cycles * 2
+        if len(self._action_history) < needed_length:
+            return
+
+        window = self._action_history[-needed_length:]
+        sigs = [a.sig for a in window]
+        a_sig, b_sig = sigs[0], sigs[1]
+
+        if a_sig == b_sig:
+            return  # Handled by action fixation
+
+        # Check for alternating pattern: [A, B, A, B, ...]
+        is_oscillating = True
+        for i, s in enumerate(sigs):
+            expected = a_sig if i % 2 == 0 else b_sig
+            if s != expected:
+                is_oscillating = False
+                break
+
+        if is_oscillating:
+            tools = f"{window[0].tool} <-> {window[1].tool}"
+            directive = (
+                f"BREAK: Alternating 2-state oscillation cycle detected "
+                f"({tools}) across {needed_length} steps. Reconcile conflicting "
+                f"objectives rather than toggling state."
+            )
+            self._emit_alert(
+                alert_type="loop_cycle_detected",
+                detail={
+                    "pattern": "OSCILLATION_CYCLE",
+                    "tool": tools,
+                    "count": self.max_oscillation_cycles,
+                    "threshold": self.max_oscillation_cycles,
+                    "signature": f"osc_{a_sig[:8]}_{b_sig[:8]}",
+                    "sequence": [a.tool for a in window],
+                    "remedial_directive": directive,
+                },
+            )
+
+    def _check_choreography_cycle(self) -> None:
+        """Detect 3-step action choreography cycle: A -> B -> C -> A -> B -> C."""
+        needed_length = self.max_choreography_cycles * 3
+        if len(self._action_history) < needed_length:
+            return
+
+        window = self._action_history[-needed_length:]
+        sigs = [a.sig for a in window]
+        a_sig, b_sig, c_sig = sigs[0], sigs[1], sigs[2]
+
+        if len({a_sig, b_sig, c_sig}) < 3:
+            return  # Handled by fixation or 2-cycle oscillation
+
+        is_choreography = True
+        pattern_seq = [a_sig, b_sig, c_sig]
+        for i, s in enumerate(sigs):
+            expected = pattern_seq[i % 3]
+            if s != expected:
+                is_choreography = False
+                break
+
+        if is_choreography:
+            tools = f"{window[0].tool} -> {window[1].tool} -> {window[2].tool}"
+            directive = (
+                f"BREAK: Repetitive 3-step execution choreography detected "
+                f"({tools}) across {needed_length} steps. Break execution sequence."
+            )
+            self._emit_alert(
+                alert_type="loop_cycle_detected",
+                detail={
+                    "pattern": "CHOREOGRAPHY_CYCLE",
+                    "tool": tools,
+                    "count": self.max_choreography_cycles,
+                    "threshold": self.max_choreography_cycles,
+                    "signature": f"choreo_{a_sig[:6]}_{b_sig[:6]}_{c_sig[:6]}",
+                    "sequence": [a.tool for a in window],
+                    "remedial_directive": directive,
+                },
+            )
+
+    def _check_watchdog_model(self) -> None:
+        """Tier 2: Optional model watchdog for semantic stagnation."""
+        if not callable(self.watchdog_fn):
+            return
+
+        trajectory_summary = [
+            {
+                "turn": a.turn,
+                "tool": a.tool,
+                "args": a.args,
+                "has_error": a.has_error,
+            }
+            for a in self._action_history
+        ]
+        try:
+            res = self.watchdog_fn(trajectory_summary)
+            if isinstance(res, dict) and res.get("loop_detected"):
+                directive = str(
+                    res.get("remedial_directive")
+                    or res.get("reason")
+                    or "Watchdog model detected semantic execution stagnation."
+                )
+                self._emit_alert(
+                    alert_type="loop_cycle_detected",
+                    detail={
+                        "pattern": "SEMANTIC_STAGNATION",
+                        "tool": (
+                            self._action_history[-1].tool if self._action_history else "unknown"
+                        ),
+                        "count": 1,
+                        "threshold": 1,
+                        "signature": f"watchdog_{self._current_turn}",
+                        "remedial_directive": directive,
+                    },
+                )
+        except Exception:
+            pass  # Watchdog failure must never break the main loop
 
     def _emit_alert(self, alert_type: str, detail: dict[str, Any]) -> None:
-        last = self._alerts[-1] if self._alerts else None
-        if (
-            last
-            and last.get("type") == alert_type
-            and last.get("signature") == detail.get("signature")
-        ):
-            if last.get("count") == detail.get("count"):
-                return
-        alert = {"type": alert_type, **detail, "observer_id": self.observer_id}
+        """Deduplicate, enrich with cryptographic spine hashes, and emit alert."""
+        sig = str(detail.get("signature") or "")
+        count = int(detail.get("count") or 0)
+        pattern = str(detail.get("pattern") or "")
+        dedup_key = f"{alert_type}:{pattern}:{sig}:{count}"
+        if dedup_key in self._tripped_dedup_keys:
+            return
+        self._tripped_dedup_keys.add(dedup_key)
+
+        alert = {
+            "type": alert_type,
+            **detail,
+            "observer_id": self.observer_id,
+            "window_size": self.window_size,
+        }
+
+        # Cryptographic spine hash capture for non-repudiable audit proofs
+        spine = getattr(self._session, "spine", None)
+        if spine is not None and hasattr(spine, "_last_content_hash"):
+            rec_hash = spine._last_content_hash
+            entry_hash = rec_hash
+            if self._action_history:
+                entry_hash = self._action_history[0].spine_hash or rec_hash
+            if entry_hash:
+                alert.setdefault("cycle_entry_hash", entry_hash)
+            if rec_hash:
+                alert.setdefault("cycle_recurrence_hash", rec_hash)
+
+        # FinOps burn velocity guard check
+        if self.burn_velocity_guard or self.max_cycle_token_waste > 0:
+            alert["churn_tokens_estimate"] = self._cycle_token_accumulator
+            if (
+                self.max_cycle_token_waste > 0
+                and self._cycle_token_accumulator >= self.max_cycle_token_waste
+            ):
+                alert["burn_rate_spike"] = True
+
         self._alerts.append(alert)
-        spine = self._session.spine
+
+        # Enforce action if configured
+        self._apply_configured_action(alert)
+
         if spine is not None:
             spine.append(
                 "observer.alert",
@@ -70,6 +444,49 @@ def _emit_alert(self, alert_type: str, detail: dict[str, Any]) -> None:
                 agent_ids=self._session.agent_ids_trailer(),
             )
 
+    def _apply_configured_action(self, alert: dict[str, Any]) -> None:
+        """Enforce configured remedial action (pause session or trip egress circuit breaker)."""
+        if self.action in ("pause", "pause_session"):
+            pause_rule = {
+                "type": "break_pause",
+                "request_id": f"break_pause_{uuid.uuid4().hex[:10]}",
+                "source": "break_observer",
+                "pattern": alert.get("pattern"),
+                "remedial_directive": alert.get("remedial_directive"),
+            }
+            if hasattr(self._session, "rules"):
+                self._session.rules.append(pause_rule)
+            if hasattr(self._session, "state"):
+                self._session.state["_break_tripped"] = True
+                self._session.state["_break_pause_rule"] = pause_rule
+
+        if self.action in ("circuit_break", "trip"):
+            circuit_rule = {
+                "type": "circuit_break",
+                "tool": alert.get("tool"),
+                "signature": alert.get("signature"),
+                "remedial_directive": alert.get("remedial_directive"),
+            }
+            if hasattr(self._session, "rules"):
+                self._session.rules.append(circuit_rule)
+            if hasattr(self._session, "state"):
+                self._session.state["_break_tripped"] = True
+
+    def summary(self) -> dict[str, Any]:
+        return {
+            "intent_signatures": dict(self._intent_signatures),
+            "alerts_emitted": len(self._alerts),
+            "cycles_detected": sum(1 for a in self._alerts if a.get("pattern")),
+            "patterns": sorted(list({a.get("pattern") for a in self._alerts if a.get("pattern")})),
+            "tripped": len(self._alerts) > 0,
+        }
+
+    @staticmethod
+    def _normalize_args(args: Any) -> str:
+        if not isinstance(args, dict):
+            return str(args)
+        return json.dumps(args, sort_keys=True, default=str)
+
 
 def create_break_observer(session: Session, entry: dict[str, Any]) -> BreakObserver:
     obs_id = str(entry.get("id") or "break")
diff --git a/docs/observers.md b/docs/observers.md
index 6b6d327..52a8ab0 100644
--- a/docs/observers.md
+++ b/docs/observers.md
@@ -138,12 +138,31 @@ config:
 
 ### break
 
+Detects runaway loops, multi-step action cycles, and cognitive deadlocks within a sliding window.
+
 ```yaml
 config:
-  max_identical_intents: 5
-  window_seconds: 60
+  window_size: 8                     # rolling trajectory window (default 8)
+  max_identical_actions: 3           # 1-cycle: consecutive identical tool + normalized args
+  max_oscillation_cycles: 2          # 2-cycle: alternating oscillation (A <-> B <-> A <-> B)
+  max_choreography_cycles: 2         # 3-cycle: execution choreography (A -> B -> C -> A -> B -> C)
+  max_error_churn: 3                 # consecutive identical error code / fingerprint churn
+  thought_similarity_threshold: 0.85 # reasoning text similarity threshold across turns
+  max_thought_inertia: 3             # consecutive turns with high thought similarity
+  burn_velocity_guard: true          # FinOps guard: track token churn during cycles
+  max_cycle_token_waste: 2000        # token threshold tripping burn rate alert
+  max_identical_intents: 5           # legacy global threshold (0 = disabled)
+  action: alert_only                 # alert_only (default) | pause | circuit_break
+  mode: default                      # default | watchdog (Tier 2 model/heuristic hook)
 ```
 
+**Spine output:** `observer.alert` (`type: loop_cycle_detected` or `repeated_tool_intent`), enriched with:
+- `pattern`: `ACTION_FIXATION`, `OSCILLATION_CYCLE`, `CHOREOGRAPHY_CYCLE`, `ERROR_STATUS_CHURN`, `REASONING_INERTIA`
+- `remedial_directive`: Actionable prompt-steering recommendation to break the deadlock
+- `cycle_entry_hash` & `cycle_recurrence_hash`: Immutable SHA-256 spine hashes providing non-repudiable audit proofs of algorithmic deadlock (EU AI Act Art. 14)
+- `churn_tokens_estimate` & `burn_rate_spike`: FinOps cost indicators
+
+
 ### limit
 
 ```yaml
diff --git a/docs/outputs.md b/docs/outputs.md
index 35fa5c2..9863afb 100644
--- a/docs/outputs.md
+++ b/docs/outputs.md
@@ -32,6 +32,9 @@ CLI: `aura report show <session_id>`, `aura report show <session_id> --json`, `a
 
 **Spine events (middleware & PII masking):** When sequencer steps or boundaries execute middleware (e.g. `middleware.pii_mask` [#58](https://github.com/ARPAHLS/aura/issues/58)), the session emits `middleware.applied` containing detection counts (`entities_detected`), total redactions, strategy (`redact`, `mask`, `pseudonymize`), and modified fields. Raw PII is strictly forbidden from the spine. Audit scorecard records `scorecard.middleware = {"applied": count, "pii_redactions": count}` and surfaces finding **`PII_MASKED`** (`severity: info`). See [sequencer.md](sequencer.md#middleware-pii-masking).
 
+**Spine events (break preset & observer alerts):** When observer presets (such as `break` [#93](https://github.com/ARPAHLS/aura/issues/93)) detect runaway cycles or repetition, the session emits `observer.alert` containing pattern (`ACTION_FIXATION`, `OSCILLATION_CYCLE`, `CHOREOGRAPHY_CYCLE`, `ERROR_STATUS_CHURN`, `REASONING_INERTIA`), structured `remedial_directive`, and non-repudiable cryptographic hashes (`cycle_entry_hash`, `cycle_recurrence_hash`). Audit scorecard records `scorecard.observers = {"alerts": count, "loop_cycles": count}` and surfaces finding **`COGNITIVE_CYCLE_DETECTED`** (`severity: warn`). See [observers.md](observers.md#break).
+
+
 ---
 
 ## Audit report (summary JSON)
diff --git a/scripts/aura_break_loop_stress_sim.py b/scripts/aura_break_loop_stress_sim.py
new file mode 100644
index 0000000..648717f
--- /dev/null
+++ b/scripts/aura_break_loop_stress_sim.py
@@ -0,0 +1,340 @@
+#!/usr/bin/env python3
+"""
+Break Observer & Trajectory Cycle Governance Stress Simulation (#93).
+
+Stress-tests:
+1. High-throughput sliding window stream (5,000 events; measures throughput & memory locality).
+2. Multi-state oscillation detection (A <-> B) amidst noisy background tool calls.
+3. 3-step choreography detection (A -> B -> C -> A -> B -> C) under high token load.
+4. Error fingerprint churn detection under consecutive HTTP and API failure codes.
+5. Reasoning inertia sensor across continuous thought text trajectories.
+6. FinOps burn-velocity circuit breaking and cryptographic spine hash non-repudiation.
+7. Full Audit Report scorecard and COGNITIVE_CYCLE_DETECTED finding synthesis.
+
+Usage:
+  python scripts/aura_break_loop_stress_sim.py
+
+Exit 0 when all scenarios pass; 1 on any failure.
+"""
+
+from __future__ import annotations
+
+import sys
+import tempfile
+import time
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any
+
+_REPO = Path(__file__).resolve().parents[1]
+if str(_REPO) not in sys.path:
+    sys.path.insert(0, str(_REPO))
+
+from aura import configure, agent  # noqa: E402
+from aura.core.audit_report import AuditReportBuilder  # noqa: E402
+from aura.core.conformance import ConformanceEngine  # noqa: E402
+from aura.observers.presets.break_observer import BreakObserver  # noqa: E402
+
+
+@dataclass
+class ScenarioResult:
+    name: str
+    passed: bool
+    reason: str = ""
+    metrics: dict[str, Any] = field(default_factory=dict)
+
+
+def scenario_high_throughput_sliding_window() -> ScenarioResult:
+    """Benchmark high-throughput event processing and sliding window pruning."""
+    with tempfile.TemporaryDirectory() as tmp:
+        configure(home=Path(tmp))
+        ag = agent(
+            "bench-agent",
+            observers=[
+                {
+                    "preset": "break",
+                    "config": {
+                        "window_size": 10,
+                        "max_identical_actions": 5,
+                        "max_identical_intents": 0,
+                    },
+                }
+            ],
+        )
+
+        event_count = 5000
+        start_time = time.monotonic()
+        with ag.session(export=False) as run:
+            for i in range(event_count):
+                # Varied tools so it does not trip fixation
+                tool_name = f"tool_{i % 8}"
+                run.emit("tool.intent", {"tool": tool_name, "args": {"step": i}})
+        elapsed = time.monotonic() - start_time
+        rate = event_count / elapsed if elapsed > 0 else 0
+
+        # Verify window size did not grow beyond limit
+        observer = run._session._observers[0]
+        assert isinstance(observer, BreakObserver)
+        window_len = len(observer._action_history)
+
+        if window_len > 10:
+            return ScenarioResult(
+                name="High-throughput sliding window",
+                passed=False,
+                reason=f"Window leaked: {window_len} > 10",
+            )
+
+        return ScenarioResult(
+            name="High-throughput sliding window",
+            passed=True,
+            metrics={"events": event_count, "elapsed_s": round(elapsed, 4), "rate_eps": int(rate)},
+        )
+
+
+def scenario_noisy_oscillation_detection() -> ScenarioResult:
+    """Verify detection of A <-> B oscillation when surrounded by noise."""
+    with tempfile.TemporaryDirectory() as tmp:
+        configure(home=Path(tmp))
+        ag = agent(
+            "noisy-osc-agent",
+            observers=[
+                {
+                    "preset": "break",
+                    "config": {
+                        "window_size": 8,
+                        "max_oscillation_cycles": 2,
+                        "max_identical_actions": 5,
+                        "max_identical_intents": 0,
+                    },
+                }
+            ],
+        )
+
+        with ag.session(export=False) as run:
+            # Noise steps
+            for i in range(5):
+                run.emit("tool.intent", {"tool": f"init_{i}", "args": {}})
+
+            # Alternating oscillation
+            run.emit("tool.intent", {"tool": "db_read", "args": {"key": "x"}})
+            run.emit("tool.intent", {"tool": "cache_clear", "args": {"all": True}})
+            run.emit("tool.intent", {"tool": "db_read", "args": {"key": "x"}})
+            run.emit("tool.intent", {"tool": "cache_clear", "args": {"all": True}})
+
+        alerts = [e for e in run._session.spine.stream() if e.kind == "observer.alert"]
+        if len(alerts) != 1 or alerts[0].payload.get("pattern") != "OSCILLATION_CYCLE":
+            return ScenarioResult(
+                name="Noisy oscillation detection",
+                passed=False,
+                reason=f"Expected 1 OSCILLATION_CYCLE alert, got {len(alerts)}",
+            )
+
+        return ScenarioResult(
+            name="Noisy oscillation detection",
+            passed=True,
+            metrics={"alerts": len(alerts), "pattern": alerts[0].payload.get("pattern")},
+        )
+
+
+def scenario_choreography_and_burn_velocity() -> ScenarioResult:
+    """Verify 3-step choreography with token accumulation."""
+    with tempfile.TemporaryDirectory() as tmp:
+        configure(home=Path(tmp))
+        ag = agent(
+            "choreo-burn-agent",
+            observers=[
+                {
+                    "preset": "break",
+                    "config": {
+                        "window_size": 8,
+                        "max_choreography_cycles": 2,
+                        "burn_velocity_guard": True,
+                        "max_cycle_token_waste": 1200,
+                        "max_identical_intents": 0,
+                    },
+                }
+            ],
+        )
+
+        with ag.session(export=False) as run:
+            seq = [("read_code", 500), ("edit_code", 500), ("lint_check", 500)]
+            for _ in range(2):
+                for tool, tokens in seq:
+                    run.emit("tool.intent", {"tool": tool, "args": {"f": "app.py"}})
+                    run.emit("tool.result", {"tool": tool, "tokens": tokens})
+
+        alerts = [e for e in run._session.spine.stream() if e.kind == "observer.alert"]
+        if not alerts:
+            return ScenarioResult(
+                name="Choreography & burn velocity",
+                passed=False,
+                reason="No choreography alert emitted",
+            )
+
+        alert = alerts[0].payload
+        if alert.get("pattern") != "CHOREOGRAPHY_CYCLE":
+            return ScenarioResult(
+                name="Choreography & burn velocity",
+                passed=False,
+                reason=f"Wrong pattern: {alert.get('pattern')}",
+            )
+
+        if not alert.get("burn_rate_spike"):
+            return ScenarioResult(
+                name="Choreography & burn velocity",
+                passed=False,
+                reason=f"Expected burn_rate_spike=True, got {alert.get('burn_rate_spike')}",
+            )
+
+        return ScenarioResult(
+            name="Choreography & burn velocity",
+            passed=True,
+            metrics={
+                "churn_tokens": alert.get("churn_tokens_estimate"),
+                "burn_spike": alert.get("burn_rate_spike"),
+            },
+        )
+
+
+def scenario_error_churn_stress() -> ScenarioResult:
+    """Stress test consecutive error churn with varying tools but same root error code."""
+    with tempfile.TemporaryDirectory() as tmp:
+        configure(home=Path(tmp))
+        ag = agent(
+            "err-stress-agent",
+            observers=[
+                {
+                    "preset": "break",
+                    "config": {
+                        "window_size": 8,
+                        "max_error_churn": 4,
+                        "max_identical_actions": 10,
+                        "max_identical_intents": 0,
+                    },
+                }
+            ],
+        )
+
+        with ag.session(export=False) as run:
+            for i in range(4):
+                run.emit("tool.intent", {"tool": f"endpoint_{i}", "args": {}})
+                run.emit(
+                    "tool.result",
+                    {"tool": f"endpoint_{i}", "status": 503, "error": "ServiceUnavailable"},
+                )
+
+        alerts = [e for e in run._session.spine.stream() if e.kind == "observer.alert"]
+        if len(alerts) != 1 or alerts[0].payload.get("pattern") != "ERROR_STATUS_CHURN":
+            return ScenarioResult(
+                name="Error churn stress",
+                passed=False,
+                reason=f"Expected 1 ERROR_STATUS_CHURN alert, got {len(alerts)}",
+            )
+
+        return ScenarioResult(
+            name="Error churn stress",
+            passed=True,
+            metrics={"count": alerts[0].payload.get("count"), "pattern": "ERROR_STATUS_CHURN"},
+        )
+
+
+def scenario_audit_report_and_cryptographic_proof() -> ScenarioResult:
+    """Verify non-repudiable spine hashes and audit scorecard synthesis."""
+    with tempfile.TemporaryDirectory() as tmp:
+        configure(home=Path(tmp))
+        ag = agent(
+            "audit-proof-agent",
+            observers=[
+                {
+                    "preset": "break",
+                    "config": {
+                        "max_identical_actions": 2,
+                        "max_identical_intents": 0,
+                    },
+                }
+            ],
+            escalations=[
+                {
+                    "on": "observer.alert",
+                    "actions": ["log", "nudge:Break cognitive loop immediately"],
+                }
+            ],
+        )
+
+        with ag.session(export=False) as run:
+            run.emit("tool.intent", {"tool": "retry_op", "args": {"attempt": 1}})
+            run.emit("tool.intent", {"tool": "retry_op", "args": {"attempt": 1}})
+
+        conf = ConformanceEngine().summarize(run._session.spine, [])
+        report = AuditReportBuilder().build(run._session.spine, conf)
+
+        scorecard = report.scorecard
+        if scorecard.get("observers", {}).get("loop_cycles") != 1:
+            return ScenarioResult(
+                name="Audit report & cryptographic proof",
+                passed=False,
+                reason=f"Expected 1 loop cycle in scorecard, got {scorecard.get('observers')}",
+            )
+
+        cycle_finding = next(
+            (f for f in report.findings if f["code"] == "COGNITIVE_CYCLE_DETECTED"), None
+        )
+        if not cycle_finding:
+            return ScenarioResult(
+                name="Audit report & cryptographic proof",
+                passed=False,
+                reason="COGNITIVE_CYCLE_DETECTED finding missing in audit report",
+            )
+
+        entry_h = cycle_finding.get("cycle_entry_hash")
+        rec_h = cycle_finding.get("cycle_recurrence_hash")
+        if not entry_h or not rec_h or len(entry_h) != 64 or len(rec_h) != 64:
+            return ScenarioResult(
+                name="Audit report & cryptographic proof",
+                passed=False,
+                reason=f"Invalid cryptographic proof hashes: entry={entry_h}, rec={rec_h}",
+            )
+
+        return ScenarioResult(
+            name="Audit report & cryptographic proof",
+            passed=True,
+            metrics={"entry_hash": entry_h[:12] + "...", "rec_hash": rec_h[:12] + "..."},
+        )
+
+
+def run_all_scenarios() -> int:
+    scenarios = [
+        scenario_high_throughput_sliding_window,
+        scenario_noisy_oscillation_detection,
+        scenario_choreography_and_burn_velocity,
+        scenario_error_churn_stress,
+        scenario_audit_report_and_cryptographic_proof,
+    ]
+
+    print("\n" + "=" * 70)
+    print(" AURA Break Observer Trajectory & Loop Governance Stress Suite (#93)")
+    print("=" * 70 + "\n")
+
+    failed = False
+    for sc in scenarios:
+        res = sc()
+        status_str = "[PASS]" if res.passed else "[FAIL]"
+        print(f" {status_str} {res.name}")
+        if res.metrics:
+            for k, v in res.metrics.items():
+                print(f"        - {k}: {v}")
+        if not res.passed:
+            print(f"        ERROR: {res.reason}")
+            failed = True
+
+    print("\n" + "-" * 70)
+    if failed:
+        print(" RESULT: FAILED — Some stress scenarios encountered regressions.")
+        return 1
+    print(" RESULT: ALL SCENARIOS PASSED SUCCESSFULLY.")
+    print("-" * 70 + "\n")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(run_all_scenarios())
diff --git a/tests/test_break_loop_stress_sim.py b/tests/test_break_loop_stress_sim.py
new file mode 100644
index 0000000..1bf1ae0
--- /dev/null
+++ b/tests/test_break_loop_stress_sim.py
@@ -0,0 +1,23 @@
+"""Run the AURA break observer trajectory cycle governance stress simulation (#93)."""
+
+from __future__ import annotations
+
+import subprocess
+import sys
+from pathlib import Path
+
+REPO = Path(__file__).resolve().parents[1]
+SIM = REPO / "scripts" / "aura_break_loop_stress_sim.py"
+
+
+def test_break_loop_stress_simulation():
+    result = subprocess.run(
+        [sys.executable, str(SIM)],
+        cwd=str(REPO),
+        capture_output=True,
+        text=True,
+        encoding="utf-8",
+        errors="replace",
+    )
+    assert result.returncode == 0, result.stderr or result.stdout
+    assert "ALL SCENARIOS PASSED SUCCESSFULLY" in result.stdout
diff --git a/tests/test_observer_break_loop.py b/tests/test_observer_break_loop.py
new file mode 100644
index 0000000..112e691
--- /dev/null
+++ b/tests/test_observer_break_loop.py
@@ -0,0 +1,467 @@
+"""Tests for BreakObserver trajectory cycle detection, sliding window, and FinOps guards (#93)."""
+
+from __future__ import annotations
+
+from aura.api import agent
+from aura.core.audit_report import AuditReportBuilder
+from aura.core.conformance import ConformanceEngine
+
+
+def test_break_observer_legacy_identical_intents_compatibility(aura_home):
+    """Verify 100% backward compatibility for legacy max_identical_intents config."""
+    ag = agent(
+        "legacy-break",
+        observers=[{"preset": "break", "config": {"max_identical_intents": 2}}],
+    )
+    with ag.session(export=False) as run:
+        for _ in range(3):
+            run.emit("tool.intent", {"tool": "search", "args": {"q": "term"}})
+
+    kinds = [e.kind for e in run._session.spine.stream()]
+    assert "observer.alert" in kinds
+    alerts = [e for e in run._session.spine.stream() if e.kind == "observer.alert"]
+    assert len(alerts) >= 1
+    first_alert = alerts[0].payload
+    assert first_alert["tool"] == "search"
+    assert first_alert["threshold"] == 2
+    assert "remedial_directive" in first_alert
+
+
+def test_action_fixation_detection_in_sliding_window(aura_home):
+    """Detect 1-cycle: same action and arguments repeated consecutively."""
+    ag = agent(
+        "fixation-agent",
+        observers=[
+            {
+                "preset": "break",
+                "config": {
+                    "window_size": 6,
+                    "max_identical_actions": 3,
+                    "max_identical_intents": 0,  # disable legacy global count
+                },
+            }
+        ],
+    )
+    with ag.session(export=False) as run:
+        # 2 calls: should not trip
+        run.emit("tool.intent", {"tool": "db_query", "args": {"sql": "SELECT 1"}})
+        run.emit("tool.intent", {"tool": "db_query", "args": {"sql": "SELECT 1"}})
+        assert not any(e.kind == "observer.alert" for e in run._session.spine.stream())
+
+        # 3rd identical call: trips ACTION_FIXATION
+        run.emit("tool.intent", {"tool": "db_query", "args": {"sql": "SELECT 1"}})
+
+    alerts = [e for e in run._session.spine.stream() if e.kind == "observer.alert"]
+    assert len(alerts) == 1
+    alert = alerts[0].payload
+    assert alert["pattern"] == "ACTION_FIXATION"
+    assert alert["tool"] == "db_query"
+    assert alert["count"] == 3
+    assert "BREAK" in alert["remedial_directive"]
+
+
+def test_state_oscillation_cycle_detection(aura_home):
+    """Detect 2-cycle alternating state oscillation: A -> B -> A -> B."""
+    ag = agent(
+        "oscillation-agent",
+        observers=[
+            {
+                "preset": "break",
+                "config": {
+                    "window_size": 8,
+                    "max_oscillation_cycles": 2,
+                    "max_identical_actions": 5,
+                    "max_identical_intents": 0,
+                },
+            }
+        ],
+    )
+    with ag.session(export=False) as run:
+        # A -> B -> A -> B
+        run.emit("tool.intent", {"tool": "git_pull", "args": {"branch": "main"}})
+        run.emit("tool.intent", {"tool": "git_stash", "args": {"pop": True}})
+        run.emit("tool.intent", {"tool": "git_pull", "args": {"branch": "main"}})
+        run.emit("tool.intent", {"tool": "git_stash", "args": {"pop": True}})
+
+    alerts = [e for e in run._session.spine.stream() if e.kind == "observer.alert"]
+    assert len(alerts) == 1
+    alert = alerts[0].payload
+    assert alert["pattern"] == "OSCILLATION_CYCLE"
+    assert "git_pull" in alert["tool"] and "git_stash" in alert["tool"]
+    assert alert["count"] == 2
+    assert "Alternating 2-state oscillation cycle detected" in alert["remedial_directive"]
+
+
+def test_choreography_cycle_detection(aura_home):
+    """Detect 3-cycle execution choreography: A -> B -> C -> A -> B -> C."""
+    ag = agent(
+        "choreo-agent",
+        observers=[
+            {
+                "preset": "break",
+                "config": {
+                    "window_size": 8,
+                    "max_choreography_cycles": 2,
+                    "max_identical_actions": 5,
+                    "max_oscillation_cycles": 5,
+                    "max_identical_intents": 0,
+                },
+            }
+        ],
+    )
+    with ag.session(export=False) as run:
+        # Sequence: Read -> Edit -> Test -> Read -> Edit -> Test
+        run.emit("tool.intent", {"tool": "file_read", "args": {"path": "main.py"}})
+        run.emit("tool.intent", {"tool": "file_edit", "args": {"path": "main.py"}})
+        run.emit("tool.intent", {"tool": "test_run", "args": {"cmd": "pytest"}})
+        run.emit("tool.intent", {"tool": "file_read", "args": {"path": "main.py"}})
+        run.emit("tool.intent", {"tool": "file_edit", "args": {"path": "main.py"}})
+        run.emit("tool.intent", {"tool": "test_run", "args": {"cmd": "pytest"}})
+
+    alerts = [e for e in run._session.spine.stream() if e.kind == "observer.alert"]
+    assert len(alerts) == 1
+    alert = alerts[0].payload
+    assert alert["pattern"] == "CHOREOGRAPHY_CYCLE"
+    assert "file_read -> file_edit -> test_run" in alert["tool"]
+    assert alert["count"] == 2
+    assert "Repetitive 3-step execution choreography detected" in alert["remedial_directive"]
+
+
+def test_error_status_churn_detection(aura_home):
+    """Detect consecutive identical error fingerprints."""
+    ag = agent(
+        "error-churn-agent",
+        observers=[
+            {
+                "preset": "break",
+                "config": {
+                    "window_size": 6,
+                    "max_error_churn": 3,
+                    "max_identical_actions": 10,
+                    "max_identical_intents": 0,
+                },
+            }
+        ],
+    )
+    with ag.session(export=False) as run:
+        for i in range(3):
+            run.emit("tool.intent", {"tool": "fetch_api", "args": {"url": f"https://api.test/{i}"}})
+            run.emit(
+                "tool.result",
+                {"tool": "fetch_api", "status": 403, "error": "Forbidden - Invalid Token"},
+            )
+
+    alerts = [e for e in run._session.spine.stream() if e.kind == "observer.alert"]
+    assert len(alerts) == 1
+    alert = alerts[0].payload
+    assert alert["pattern"] == "ERROR_STATUS_CHURN"
+    assert alert["tool"] == "fetch_api"
+    assert alert["count"] == 3
+    assert "identical error fingerprint" in alert["remedial_directive"]
+
+
+def test_reasoning_inertia_detection(aura_home):
+    """Detect consecutive turns exhibiting >=85% thought text similarity."""
+    ag = agent(
+        "thought-agent",
+        observers=[
+            {
+                "preset": "break",
+                "config": {
+                    "thought_similarity_threshold": 0.85,
+                    "max_thought_inertia": 3,
+                    "max_identical_intents": 0,
+                },
+            }
+        ],
+    )
+    with ag.session(export=False) as run:
+        # Turn 1
+        run.emit(
+            "turn.start",
+            {"thought": "I should search for the user authentication tokens in settings."},
+        )
+        run.emit("turn.end", {})
+        # Turn 2
+        run.emit(
+            "turn.start",
+            {"thought": "I should search for the user authentication tokens in settings config."},
+        )
+        run.emit("turn.end", {})
+        # Turn 3: highly similar thought
+        run.emit(
+            "turn.start",
+            {"thought": "I should search for the user authentication tokens in settings file."},
+        )
+        run.emit("turn.end", {})
+
+    alerts = [e for e in run._session.spine.stream() if e.kind == "observer.alert"]
+    assert len(alerts) == 1
+    alert = alerts[0].payload
+    assert alert["pattern"] == "REASONING_INERTIA"
+    assert alert["tool"] == "host.reasoning"
+    assert "Consecutive reasoning steps show >=" in alert["remedial_directive"]
+
+
+def test_sliding_window_locality_avoids_false_positives(aura_home):
+    """Spaced out identical tool calls across a long session do not falsely trigger."""
+    ag = agent(
+        "spaced-agent",
+        observers=[
+            {
+                "preset": "break",
+                "config": {
+                    "window_size": 4,
+                    "max_identical_actions": 3,
+                    "max_identical_intents": 0,
+                },
+            }
+        ],
+    )
+    with ag.session(export=False) as run:
+        # Call X, then intermediate Y, Z, W, then X again -> window never contains 3 consecutive X's
+        for i in range(10):
+            run.emit("tool.intent", {"tool": "search", "args": {"q": "aura"}})
+            run.emit("tool.intent", {"tool": "step_a", "args": {"step": i}})
+            run.emit("tool.intent", {"tool": "step_b", "args": {"step": i}})
+            run.emit("tool.intent", {"tool": "step_c", "args": {"step": i}})
+
+    alerts = [e for e in run._session.spine.stream() if e.kind == "observer.alert"]
+    assert len(alerts) == 0
+
+
+def test_burn_velocity_token_guard(aura_home):
+    """FinOps token accumulator flags burn rate spikes."""
+    ag = agent(
+        "finops-agent",
+        observers=[
+            {
+                "preset": "break",
+                "config": {
+                    "burn_velocity_guard": True,
+                    "max_cycle_token_waste": 800,
+                    "max_identical_actions": 2,
+                    "max_identical_intents": 0,
+                },
+            }
+        ],
+    )
+    with ag.session(export=False) as run:
+        run.emit("tool.intent", {"tool": "rag_query", "args": {"q": "big context"}})
+        run.emit("tool.result", {"tool": "rag_query", "tokens": 1000})
+
+        run.emit("tool.intent", {"tool": "rag_query", "args": {"q": "big context"}})
+        run.emit("tool.result", {"tool": "rag_query", "tokens": 1000})
+
+    alerts = [e for e in run._session.spine.stream() if e.kind == "observer.alert"]
+    assert len(alerts) == 1
+    alert = alerts[0].payload
+    assert alert.get("churn_tokens_estimate") == 1000
+    assert alert.get("burn_rate_spike") is True
+
+
+def test_cryptographic_spine_hash_capture(aura_home):
+    """Ensure non-repudiable spine event hashes are attached to alerts."""
+    ag = agent(
+        "crypto-proof-agent",
+        observers=[
+            {
+                "preset": "break",
+                "config": {
+                    "max_identical_actions": 2,
+                    "max_identical_intents": 0,
+                },
+            }
+        ],
+    )
+    with ag.session(export=False) as run:
+        run.emit("tool.intent", {"tool": "echo", "args": {"msg": "hello"}})
+        run.emit("tool.intent", {"tool": "echo", "args": {"msg": "hello"}})
+
+    alerts = [e for e in run._session.spine.stream() if e.kind == "observer.alert"]
+    assert len(alerts) == 1
+    alert = alerts[0].payload
+    assert "cycle_entry_hash" in alert
+    assert "cycle_recurrence_hash" in alert
+    assert len(alert["cycle_entry_hash"]) == 64
+    assert len(alert["cycle_recurrence_hash"]) == 64
+
+
+def test_escalation_playbook_trigger_on_break_alert(aura_home):
+    """Verify that observer.alert triggers escalation playbooks (e.g. nudge, log)."""
+    ag = agent(
+        "esc-break-agent",
+        observers=[
+            {
+                "preset": "break",
+                "config": {"max_identical_actions": 2, "max_identical_intents": 0},
+            }
+        ],
+        escalations=[
+            {
+                "on": "observer.alert",
+                "actions": ["log", "nudge:Break loop — strategy shift required"],
+            }
+        ],
+    )
+    with ag.session(export=False) as run:
+        run.emit("tool.intent", {"tool": "stuck_tool", "args": {}})
+        run.emit("tool.intent", {"tool": "stuck_tool", "args": {}})
+
+    kinds = [e.kind for e in run._session.spine.stream()]
+    assert "observer.alert" in kinds
+    assert "escalation.fired" in kinds
+    assert "membrane.nudge" in kinds
+
+
+def test_break_preset_audit_report(aura_home):
+    """Test preset: break audit report scorecard/findings synthesis."""
+    ag = agent(
+        "break-report-agent",
+        observers=[
+            {
+                "preset": "break",
+                "config": {
+                    "max_oscillation_cycles": 2,
+                    "max_identical_actions": 5,
+                    "max_identical_intents": 0,
+                },
+            }
+        ],
+    )
+    with ag.session(export=False) as run:
+        run.emit("tool.intent", {"tool": "step_a", "args": {"id": 1}})
+        run.emit("tool.intent", {"tool": "step_b", "args": {"id": 2}})
+        run.emit("tool.intent", {"tool": "step_a", "args": {"id": 1}})
+        run.emit("tool.intent", {"tool": "step_b", "args": {"id": 2}})
+
+    conf = ConformanceEngine().summarize(run._session.spine, [])
+    report = AuditReportBuilder().build(run._session.spine, conf)
+
+    # Scorecard verification
+    scorecard = report.scorecard
+    assert "observers" in scorecard
+    assert scorecard["observers"]["alerts"] == 1
+    assert scorecard["observers"]["loop_cycles"] == 1
+
+    # Findings verification
+    codes = [f["code"] for f in report.findings]
+    assert "COGNITIVE_CYCLE_DETECTED" in codes
+    cycle_finding = next(f for f in report.findings if f["code"] == "COGNITIVE_CYCLE_DETECTED")
+    assert cycle_finding["pattern"] == "OSCILLATION_CYCLE"
+    assert "cycle_entry_hash" in cycle_finding
+    assert "Alternating 2-state oscillation cycle detected" in cycle_finding["message"]
+    assert any(
+        "Alternating 2-state oscillation cycle detected" in r for r in report.recommendations
+    )
+
+
+def test_tier2_watchdog_mode(aura_home):
+    """Test optional mode='watchdog' with pluggable watchdog_fn callback."""
+    called = []
+
+    def mock_watchdog(trajectory):
+        called.append(trajectory)
+        if len(trajectory) >= 2:
+            return {
+                "loop_detected": True,
+                "remedial_directive": "Watchdog: Semantic goal stalling detected.",
+            }
+        return {"loop_detected": False}
+
+    ag = agent(
+        "watchdog-agent",
+        observers=[
+            {
+                "preset": "break",
+                "config": {
+                    "mode": "watchdog",
+                    "watchdog_fn": mock_watchdog,
+                    "max_identical_actions": 10,
+                    "max_identical_intents": 0,
+                },
+            }
+        ],
+    )
+    with ag.session(export=False) as run:
+        run.emit("tool.intent", {"tool": "probe_a", "args": {"target": "alpha"}})
+        run.emit("tool.intent", {"tool": "probe_b", "args": {"target": "beta"}})
+
+    assert len(called) >= 1
+    alerts = [e for e in run._session.spine.stream() if e.kind == "observer.alert"]
+    assert len(alerts) == 1
+    alert = alerts[0].payload
+    assert alert["pattern"] == "SEMANTIC_STAGNATION"
+    assert "Watchdog: Semantic goal stalling detected." in alert["remedial_directive"]
+
+
+def test_break_observer_action_pause(aura_home):
+    """Test action='pause' appends break_pause rule and trips session state."""
+    ag = agent(
+        "pause-break-agent",
+        observers=[
+            {
+                "preset": "break",
+                "config": {
+                    "action": "pause",
+                    "max_identical_actions": 2,
+                    "max_identical_intents": 0,
+                },
+            }
+        ],
+    )
+    with ag.session(export=False) as run:
+        run.emit("tool.intent", {"tool": "freeze_me", "args": {}})
+        run.emit("tool.intent", {"tool": "freeze_me", "args": {}})
+
+        assert run._session.state.get("_break_tripped") is True
+        pause_rules = [r for r in run._session.rules if r.get("type") == "break_pause"]
+        assert len(pause_rules) == 1
+        assert pause_rules[0]["pattern"] == "ACTION_FIXATION"
+        assert "BREAK:" in pause_rules[0]["remedial_directive"]
+
+
+def test_break_observer_action_circuit_break(aura_home):
+    """Test action='circuit_break' appends circuit_break rule and trips session state."""
+    ag = agent(
+        "trip-break-agent",
+        observers=[
+            {
+                "preset": "break",
+                "config": {
+                    "action": "circuit_break",
+                    "max_identical_actions": 2,
+                    "max_identical_intents": 0,
+                },
+            }
+        ],
+    )
+    with ag.session(export=False) as run:
+        run.emit("tool.intent", {"tool": "crash_tool", "args": {"mode": "fail"}})
+        run.emit("tool.intent", {"tool": "crash_tool", "args": {"mode": "fail"}})
+
+        assert run._session.state.get("_break_tripped") is True
+        trip_rules = [r for r in run._session.rules if r.get("type") == "circuit_break"]
+        assert len(trip_rules) == 1
+        assert trip_rules[0]["tool"] == "crash_tool"
+
+
+def test_break_observer_summary_metrics(aura_home):
+    """Test BreakObserver summary() reports counts, patterns, and tripped state."""
+    from aura.observers.presets.break_observer import BreakObserver
+
+    ag = agent("summary-agent")
+    with ag.session(export=False) as run:
+        obs = BreakObserver(
+            "test_break",
+            run._session,
+            {"max_identical_actions": 2, "max_identical_intents": 0},
+        )
+        obs.on_event({"kind": "tool.intent", "payload": {"tool": "loop_fn", "args": {}}})
+        obs.on_event({"kind": "tool.intent", "payload": {"tool": "loop_fn", "args": {}}})
+
+        s = obs.summary()
+        assert s["alerts_emitted"] == 1
+        assert s["cycles_detected"] == 1
+        assert "ACTION_FIXATION" in s["patterns"]
+        assert s["tripped"] is True