Merge pull request #572 from AssetOverflow/feat/l11-identity-continuity

feat(identity): L11 identity continuity — same identity across reboot, not just same bytes

chat/runtime.py

@@ -4,6 +4,7 @@ from dataclasses import dataclass, replace
 import hashlib
 import json
 import re
+import warnings
 from collections.abc import Sequence
 from typing import Any, List
 
@@ -52,7 +53,8 @@ from teaching.discovery import (
     format_candidate_jsonl,
 )
 from teaching.discovery_sink import DiscoveryCandidateSink
-from engine_state import EngineStateStore
+from engine_state import EngineStateStore, get_git_revision
+from core.engine_identity import engine_identity_for_config
 from recognition.anti_unifier import derive_recognizer
 from recognition.outcome import FeatureBundle
 from recognition.registry import RecognizerRegistry
@@ -674,6 +676,16 @@ class ChatRuntime:
         # checkpoint is loaded, flushed to the sink on attach_telemetry_sink.
         # None means no reboot was detected this session.
         self._pending_reboot_payload: str | None = None
+        # L11 — the engine's content-derived identity (who am I), and the
+        # identity stamped in the loaded checkpoint (the lineage parent for the
+        # next checkpoint). ``_loaded_engine_identity`` stays "" at genesis.
+        self._engine_identity: str = engine_identity_for_config(
+            self.config, get_git_revision()
+        )
+        self._loaded_engine_identity: str = ""
+        # L11 — set True on reboot when the stamped checkpoint identity differs
+        # from the recomputed identity (the ratified substrate changed while down).
+        self.identity_continuity_break: bool = False
         if self._engine_state_store is not None and self._engine_state_store.exists():
             self._load_engine_state()
 
@@ -689,6 +701,29 @@ class ChatRuntime:
         self._recognizer_registry = RecognizerRegistry.from_recognizers(recognizers)
         self._pending_candidates = store.load_discovery_candidates()
         self._turn_count = int(manifest.get("turn_count", 0))
+        # L11 — the identity this checkpoint was written under becomes the lineage
+        # parent of the next checkpoint we write. If it differs from the identity
+        # we recomputed at boot, the ratified substrate changed during downtime:
+        # we would resume the lived state under a DIFFERENT identity. Surface it
+        # (warn + flag); refuse only under strict_identity_continuity.
+        self._loaded_engine_identity = str(manifest.get("engine_identity", ""))
+        if (
+            self._loaded_engine_identity
+            and self._loaded_engine_identity != self._engine_identity
+        ):
+            self.identity_continuity_break = True
+            message = (
+                "engine identity continuity break: checkpoint was written under "
+                f"{self._loaded_engine_identity[:12]}… but this build computes "
+                f"{self._engine_identity[:12]}… — the ratified identity substrate "
+                "changed while the engine was down. Resuming would carry the lived "
+                "state into a different identity."
+            )
+            if self.config.strict_identity_continuity:
+                from core.engine_identity import IdentityContinuityError
+
+                raise IdentityContinuityError(message)
+            warnings.warn(message, RuntimeWarning, stacklevel=2)
         # Shape B+ (schema v2): restore the lived session state into the live
         # context so a reboot resumes the SAME life (field/vault/anchor/graph/
         # referents/dialogue). Opt-in (config.persist_session_state); None for a
@@ -738,7 +773,15 @@ class ChatRuntime:
         # default so one-shot runtimes don't pay the per-turn snapshot cost.
         if self._context is not None and self.config.persist_session_state:
             store.save_session_state(self._context.snapshot())
-        store.save_manifest(self._turn_count)
+        # L11 — stamp the engine's identity and its lineage parent (the identity
+        # of the prior checkpoint). Same substrate -> identity == parent (a stable
+        # life); a ratified substrate change -> identity != parent (the bump).
+        store.save_manifest(
+            self._turn_count,
+            engine_identity=self._engine_identity,
+            parent_engine_identity=self._loaded_engine_identity,
+        )
+        self._loaded_engine_identity = self._engine_identity
 
     def record_recognition_example(
         self,

conftest.py

@@ -25,6 +25,26 @@ from __future__ import annotations
 
 import pytest
 
+import engine_state
+
+
+@pytest.fixture(autouse=True)
+def _isolate_engine_state_default(tmp_path_factory, monkeypatch):
+    """Isolate the default engine-state checkpoint dir per test.
+
+    A bare ``ChatRuntime()`` (no ``engine_state_path``) falls back to
+    ``engine_state._DEFAULT_DIR`` — the shared repo ``engine_state/`` directory.
+    Tests must not share that mutable dir: one test's checkpoint (recognizers,
+    candidates, the stamped engine-identity, and — under resume mode — the lived
+    session_state) leaks into another test's fresh-state assumptions (and, since
+    L11, raises spurious identity-continuity-break warnings when a later test
+    boots under a different identity over the same dir). Point the default at a
+    fresh per-test temp dir. Tests passing an explicit ``engine_state_path`` are
+    unaffected; within one test, repeated ``ChatRuntime()`` share this dir.
+    """
+    isolated = tmp_path_factory.mktemp("engine_state_default")
+    monkeypatch.setattr(engine_state, "_DEFAULT_DIR", isolated)
+
 
 QUARANTINE: frozenset[str] = frozenset()
 

core/config.py

@@ -284,6 +284,14 @@ class RuntimeConfig:
     # use. Enabled by the L10 continuity lane and the production L10 process.
     persist_session_state: bool = False
 
+    # L11 — on reboot, if the stamped checkpoint identity != the recomputed
+    # engine identity (the ratified substrate changed during downtime), REFUSE to
+    # start (raise IdentityContinuityError) rather than the default warn-and-flag.
+    # OFF by default: reboot is recovery, not control flow (ADR-0157), and the
+    # operator must not be bricked by a benign ratified pack swap. Deployments
+    # wanting a hard identity-continuity guarantee opt in.
+    strict_identity_continuity: bool = False
+
 
 DEFAULT_IDENTITY_PACK: str = "default_general_v1"
 DEFAULT_ETHICS_PACK: str = "default_general_ethics_v1"

core/engine_identity.py

@@ -0,0 +1,134 @@
+"""EngineIdentity — content-derived identity of the ratified substrate (L11).
+
+``EngineIdentity`` is the sha256 of the canonical serialization of the engine's
+ratified PERSONALITY substrate — the active identity / safety / ethics / register
+/ anchor-lens pack files — plus the code revision. It is **content-derived, NOT
+entropy-based**: two engines with the same ratified substrate compute the SAME
+identity, because they ARE the same engine functionally (substrate is shareable).
+
+This is the "who am I" hash. It is bumped only by a ratified change to the
+identity substrate (a new identity pack, a safety-axis change), NOT by lived
+learning (recall, teaching) — that is the engine's *experience*, carried across
+reboot by the Shape B+ lived-state lineage, not its identity. The git-like
+lineage chain (parent links on ratification) and the reboot identity verification
+build on this primitive.
+
+Honest scope (per the EngineIdentity candidate note): this is a *convention*
+naming the existing per-pack content hashes, not a new pack format. The ratified
+teaching corpus / recognizer-registry head can be folded into the tuple later
+(additive — a new key changes the identity, which is the correct semantic for a
+ratified-substrate change).
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+from pathlib import Path
+from typing import Any
+
+from core.config import (
+    DEFAULT_ANCHOR_LENS,
+    DEFAULT_ETHICS_PACK,
+    DEFAULT_IDENTITY_PACK,
+    DEFAULT_REGISTER_PACK,
+    RuntimeConfig,
+)
+
+# The never-swappable safety pack default (packs/safety/loader.py).
+DEFAULT_SAFETY_PACK: str = "core_safety_axes_v1"
+
+_PACKS_ROOT = Path(__file__).resolve().parents[1] / "packs"
+
+#: role -> packs/ subdirectory holding ``<pack_id>.json``.
+_ROLE_DIRS: dict[str, str] = {
+    "identity": "identity",
+    "safety": "safety",
+    "ethics": "ethics",
+    "register": "register",
+    "anchor_lens": "anchor_lens",
+}
+
+#: The ratified roles that constitute the engine's identity, in canonical order.
+RATIFIED_ROLES: tuple[str, ...] = (
+    "identity",
+    "safety",
+    "ethics",
+    "register",
+    "anchor_lens",
+)
+
+
+class EngineIdentityError(RuntimeError):
+    """A ratified pack named by the identity tuple could not be resolved."""
+
+
+class IdentityContinuityError(RuntimeError):
+    """A reboot found the stamped checkpoint identity != the recomputed identity.
+
+    The ratified substrate changed while the engine was down, so it would resume
+    the lived state under a DIFFERENT identity than the checkpoint was written
+    under. Raised only under strict identity continuity; the default surfaces a
+    warning and a queryable break flag (reboot is recovery, not control flow).
+    """
+
+
+def _pack_content_hash(role: str, pack_id: str) -> str:
+    path = _PACKS_ROOT / _ROLE_DIRS[role] / f"{pack_id}.json"
+    if not path.exists():
+        raise EngineIdentityError(
+            f"ratified {role} pack not found: {_ROLE_DIRS[role]}/{pack_id}.json"
+        )
+    return hashlib.sha256(path.read_bytes()).hexdigest()
+
+
+def ratified_substrate(pack_ids: dict[str, str], git_revision: str) -> dict[str, Any]:
+    """The canonical, auditable identity tuple.
+
+    ``{role: {"pack_id", "sha256"}}`` for every ratified role plus
+    ``"code_revision"``. ``pack_ids`` must name a pack for every role in
+    ``RATIFIED_ROLES``.
+    """
+    substrate: dict[str, Any] = {}
+    for role in RATIFIED_ROLES:
+        pack_id = pack_ids[role]
+        substrate[role] = {
+            "pack_id": pack_id,
+            "sha256": _pack_content_hash(role, pack_id),
+        }
+    substrate["code_revision"] = git_revision
+    return substrate
+
+
+def compute_engine_identity(pack_ids: dict[str, str], git_revision: str) -> str:
+    """The sha256 EngineIdentity over the ratified substrate tuple."""
+    canonical = json.dumps(
+        ratified_substrate(pack_ids, git_revision),
+        sort_keys=True,
+        separators=(",", ":"),
+        ensure_ascii=False,
+    )
+    return hashlib.sha256(canonical.encode("utf-8")).hexdigest()
+
+
+def _resolve_pack_ids(config: RuntimeConfig) -> dict[str, str]:
+    """Resolve each ratified role to its active pack id (config override or DEFAULT)."""
+    return {
+        "identity": config.identity_pack or DEFAULT_IDENTITY_PACK,
+        "safety": DEFAULT_SAFETY_PACK,  # never-swappable
+        "ethics": config.ethics_pack or DEFAULT_ETHICS_PACK,
+        "register": config.register_pack_id or DEFAULT_REGISTER_PACK,
+        "anchor_lens": config.anchor_lens_id or DEFAULT_ANCHOR_LENS,
+    }
+
+
+def engine_identity_for_config(config: RuntimeConfig, git_revision: str) -> str:
+    """EngineIdentity for a runtime config (resolving every role to its active pack)."""
+    return compute_engine_identity(_resolve_pack_ids(config), git_revision)
+
+
+def ratified_substrate_for_config(
+    config: RuntimeConfig, git_revision: str
+) -> dict[str, Any]:
+    """The auditable identity tuple for a runtime config."""
+    return ratified_substrate(_resolve_pack_ids(config), git_revision)

engine_state/__init__.py

@@ -158,12 +158,31 @@ class EngineStateStore:
             if line.strip()
         ]
 
-    def save_manifest(self, turn_count: int) -> None:
+    def save_manifest(
-        manifest = {
+        self,
+        turn_count: int,
+        *,
+        engine_identity: str = "",
+        parent_engine_identity: str = "",
+    ) -> None:
+        """Write the checkpoint manifest.
+
+        ``engine_identity`` (L11) stamps the content-derived identity the engine
+        was running under when this checkpoint was written; ``parent_engine_
+        identity`` links it to the identity of the prior checkpoint, forming a
+        git-like lineage chain (the parent differs only across a ratified
+        substrate change). Both are additive-optional — an empty string omits the
+        key, preserving the manifest bytes for pre-L11 callers.
+        """
+        manifest: dict = {
             "schema_version": _SCHEMA_VERSION,
             "turn_count": turn_count,
             "written_at_revision": get_git_revision(),
         }
+        if engine_identity:
+            manifest["engine_identity"] = engine_identity
+        if parent_engine_identity:
+            manifest["parent_engine_identity"] = parent_engine_identity
         _atomic_write_text(
             self.path / "manifest.json",
             json.dumps(manifest, sort_keys=True, indent=2),

evals/refusal_calibration/pack_runner.py

@@ -97,7 +97,14 @@ def _load_cases(eval_dir: Path) -> list[dict[str, Any]]:
 
 
 def _run_one_case(pack_id: str, case: dict[str, Any]) -> dict[str, Any]:
-    runtime = ChatRuntime(config=RuntimeConfig(identity_pack=pack_id))
+    # Cold-start grounding probe: each (pack, case) is an independent fresh
+    # runtime, so it must NOT resume a prior session — otherwise case N restores
+    # case N-1's checkpoint (a different identity pack), which both pollutes the
+    # grounding gate (ADR-0043 pack-invariance) and raises an L11 identity-
+    # continuity-break warning. Matches run_cognition_eval.py.
+    runtime = ChatRuntime(
+        config=RuntimeConfig(identity_pack=pack_id), no_load_state=True
+    )
     pipeline = CognitiveTurnPipeline(runtime)
     try:
         result = pipeline.run(case["prompt"], max_tokens=8)

tests/test_engine_identity.py

@@ -0,0 +1,85 @@
+"""EngineIdentity — L11 Phase 1.
+
+EngineIdentity is the content-derived sha256 of the engine's ratified personality
+substrate (identity / safety / ethics / register / anchor-lens pack files) plus
+the code revision. Content-derived, NOT entropy: same substrate -> same identity
+(two engines with the same ratified packs ARE the same engine functionally).
+This is the "who am I" hash the L11 lineage chain and reboot verification build on.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from core.config import DEFAULT_IDENTITY_PACK, RuntimeConfig
+from core.engine_identity import (
+    RATIFIED_ROLES,
+    EngineIdentityError,
+    compute_engine_identity,
+    engine_identity_for_config,
+    ratified_substrate,
+)
+
+_DEFAULTS: dict[str, str] = {
+    "identity": "default_general_v1",
+    "safety": "core_safety_axes_v1",
+    "ethics": "default_general_ethics_v1",
+    "register": "default_neutral_v1",
+    "anchor_lens": "default_unanchored_v1",
+}
+
+
+def test_engine_identity_is_deterministic_and_content_derived() -> None:
+    a = compute_engine_identity(_DEFAULTS, git_revision="abc123")
+    b = compute_engine_identity(dict(_DEFAULTS), git_revision="abc123")
+    assert a == b  # same substrate -> same identity (cross-engine portable)
+    assert len(a) == 64  # sha256 hex
+
+
+def test_changing_identity_pack_changes_identity() -> None:
+    base = compute_engine_identity(_DEFAULTS, git_revision="abc123")
+    swapped = compute_engine_identity(
+        {**_DEFAULTS, "identity": "precision_first_v1"}, git_revision="abc123"
+    )
+    assert base != swapped  # a different identity pack IS a different identity
+
+
+def test_changing_code_revision_changes_identity() -> None:
+    a = compute_engine_identity(_DEFAULTS, git_revision="rev-a")
+    b = compute_engine_identity(_DEFAULTS, git_revision="rev-b")
+    assert a != b
+
+
+def test_substrate_exposes_per_role_pack_content_hashes() -> None:
+    substrate = ratified_substrate(_DEFAULTS, git_revision="abc123")
+    assert set(RATIFIED_ROLES).issubset(substrate.keys())
+    assert substrate["code_revision"] == "abc123"
+    # Each role records its pack id + a 64-hex content hash of the pack file.
+    for role in RATIFIED_ROLES:
+        assert substrate[role]["pack_id"] == _DEFAULTS[role]
+        assert len(substrate[role]["sha256"]) == 64
+    # Distinct packs have distinct content hashes.
+    assert substrate["identity"]["sha256"] != substrate["safety"]["sha256"]
+
+
+def test_missing_ratified_pack_raises() -> None:
+    with pytest.raises(EngineIdentityError):
+        compute_engine_identity(
+            {**_DEFAULTS, "identity": "no_such_pack_xyz"}, git_revision="abc123"
+        )
+
+
+def test_engine_identity_for_config_resolves_defaults() -> None:
+    # An empty config resolves every role to its DEFAULT pack.
+    from_config = engine_identity_for_config(RuntimeConfig(), git_revision="abc123")
+    explicit = compute_engine_identity(_DEFAULTS, git_revision="abc123")
+    assert from_config == explicit
+
+
+def test_engine_identity_for_config_honors_identity_override() -> None:
+    base = engine_identity_for_config(RuntimeConfig(), git_revision="abc123")
+    override = engine_identity_for_config(
+        RuntimeConfig(identity_pack="precision_first_v1"), git_revision="abc123"
+    )
+    assert base != override
+    assert DEFAULT_IDENTITY_PACK == "default_general_v1"  # sanity on the default

tests/test_engine_identity_lineage.py

@@ -0,0 +1,80 @@
+"""EngineIdentity lineage — L11 Phase 2.
+
+Every checkpoint manifest stamps the content-derived ``engine_identity`` the
+engine was running under, plus the ``parent_engine_identity`` of the prior
+checkpoint — a git-like lineage chain. With a stable ratified substrate the
+identity is constant (identity == parent: one stable life); a ratified change
+would make identity != parent (the bump). Across reboot the chain is continuous:
+the resumed life's checkpoint descends from the pre-reboot identity.
+"""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+from chat.runtime import ChatRuntime, get_git_revision
+from core.cognition.pipeline import CognitiveTurnPipeline
+from core.config import RuntimeConfig
+from core.engine_identity import engine_identity_for_config
+
+
+def _manifest(state_dir: Path) -> dict:
+    return json.loads((state_dir / "manifest.json").read_text(encoding="utf-8"))
+
+
+def _drive(state_dir: Path, n: int, config: RuntimeConfig | None = None) -> ChatRuntime:
+    runtime = ChatRuntime(config=config or RuntimeConfig(), engine_state_path=state_dir)
+    pipe = CognitiveTurnPipeline(runtime=runtime)
+    for _ in range(n):
+        pipe.run("What causes light?")
+    return runtime
+
+
+def test_manifest_stamps_content_derived_engine_identity(tmp_path: Path) -> None:
+    state_dir = tmp_path / "es"
+    rt = _drive(state_dir, 1)
+    m = _manifest(state_dir)
+    assert m["engine_identity"] == rt._engine_identity
+    # It IS the content-derived identity of the ratified substrate.
+    assert m["engine_identity"] == engine_identity_for_config(
+        RuntimeConfig(), get_git_revision()
+    )
+
+
+def test_lineage_is_stable_across_a_continuous_run(tmp_path: Path) -> None:
+    state_dir = tmp_path / "es"
+    _drive(state_dir, 2)  # 2 checkpoints
+    m = _manifest(state_dir)
+    # Stable substrate -> identity == parent (the life is one continuous identity).
+    assert m["engine_identity"] == m["parent_engine_identity"]
+
+
+def test_lineage_is_continuous_across_reboot(tmp_path: Path) -> None:
+    state_dir = tmp_path / "es"
+    rt_a = _drive(state_dir, 2)
+    id_a = rt_a._engine_identity
+
+    # Reboot: a fresh runtime over the same checkpoint dir.
+    rt_b = ChatRuntime(config=RuntimeConfig(), engine_state_path=state_dir)
+    # The resumed life inherits the pre-reboot identity as its lineage parent.
+    assert rt_b._loaded_engine_identity == id_a
+    assert rt_b._engine_identity == id_a  # same substrate -> same identity
+
+    pipe = CognitiveTurnPipeline(runtime=rt_b)
+    pipe.run("What is a concept?")
+    m = _manifest(state_dir)
+    assert m["parent_engine_identity"] == id_a  # descends from the prior life
+    assert m["engine_identity"] == id_a
+
+
+def test_different_identity_pack_is_a_different_identity(tmp_path: Path) -> None:
+    rt_default = ChatRuntime(
+        config=RuntimeConfig(), engine_state_path=tmp_path / "a", no_load_state=True
+    )
+    rt_precision = ChatRuntime(
+        config=RuntimeConfig(identity_pack="precision_first_v1"),
+        engine_state_path=tmp_path / "b",
+        no_load_state=True,
+    )
+    assert rt_default._engine_identity != rt_precision._engine_identity

tests/test_identity_continuity_proof.py

@@ -0,0 +1,73 @@
+"""The identity-continuity PROOF — L11 Phase 4.
+
+"Same identity, not just same bytes" is proven by composing three legs:
+
+1. **Continuity (sufficient):** under a fixed identity, a rebooted resume-mode
+   run is BYTE-IDENTICAL to the uninterrupted run (Shape B+ P2b) AND the engine
+   identity is unchanged (no break). The resumed life behaves AND is-identified
+   as the same.
+2. **Load-bearing (structural):** distinct identity packs are distinct engine
+   identities — identity is causal, not incidental. (The `identity_divergence`
+   lane separately proves distinct identities also *behave* differently.)
+3. **Causal contrapositive (necessary):** resuming a checkpoint under a
+   DIFFERENT identity raises the continuity break — continuity holds iff the
+   identity is the same. Identity is necessary for continuity.
+
+Legs 1 + 3 bracket it: same identity ⟹ continuous; different identity ⟹ break.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+from chat.runtime import ChatRuntime
+from core.cognition.pipeline import CognitiveTurnPipeline
+from core.config import RuntimeConfig
+from core.engine_identity import engine_identity_for_config
+from engine_state import get_git_revision
+from evals.l10_continuity.runner import run_soak
+
+_PRECISION = RuntimeConfig(identity_pack="precision_first_v1")
+
+
+def test_resumed_life_is_byte_identical_and_same_identity(tmp_path: Path) -> None:
+    # Leg 1: under a fixed (non-default) identity, reboot is fully transparent.
+    rebooted = run_soak(
+        6, engine_state_dir=tmp_path / "r", reboot_at=(3,), config=_PRECISION
+    )
+    baseline = run_soak(6, engine_state_dir=tmp_path / "b", config=_PRECISION)
+    assert rebooted.trace_hashes() == baseline.trace_hashes()  # behavior continuous
+    # And the reboot carried the SAME identity (a fresh runtime over the dir
+    # finds no continuity break).
+    rt = ChatRuntime(config=_PRECISION, engine_state_path=tmp_path / "r")
+    assert rt.identity_continuity_break is False
+
+
+def test_identity_is_load_bearing_distinct_packs_distinct_identity() -> None:
+    # Leg 2: three identity packs -> three distinct engine identities.
+    rev = get_git_revision()
+    default_id = engine_identity_for_config(RuntimeConfig(), rev)
+    precision_id = engine_identity_for_config(_PRECISION, rev)
+    generosity_id = engine_identity_for_config(
+        RuntimeConfig(identity_pack="generosity_first_v1"), rev
+    )
+    assert len({default_id, precision_id, generosity_id}) == 3
+
+
+def test_continuity_breaks_under_a_different_identity(tmp_path: Path) -> None:
+    # Leg 3 (contrapositive): a life lived under one identity, resumed under
+    # another, raises the break — identity is NECESSARY for continuity.
+    state_dir = tmp_path / "es"
+    writer = ChatRuntime(config=_PRECISION, engine_state_path=state_dir)
+    pipe = CognitiveTurnPipeline(runtime=writer)
+    pipe.run("What causes light?")
+    pipe.run("What is a concept?")
+
+    import pytest
+
+    with pytest.warns(RuntimeWarning, match="identity continuity break"):
+        resumed = ChatRuntime(
+            config=RuntimeConfig(identity_pack="generosity_first_v1"),
+            engine_state_path=state_dir,
+        )
+    assert resumed.identity_continuity_break is True

tests/test_identity_continuity_verification.py

@@ -0,0 +1,75 @@
+"""Reboot identity verification — L11 Phase 3.
+
+On reboot the engine recomputes its identity from the current ratified substrate
+and compares it to the identity stamped in the checkpoint. A match = continuous
+identity. A mismatch = the substrate changed while the engine was down (a pack
+mutated), so resuming would carry the lived state into a DIFFERENT identity —
+surfaced (warn + ``identity_continuity_break`` flag) by default, refused under
+``strict_identity_continuity``.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+
+from chat.runtime import ChatRuntime
+from core.cognition.pipeline import CognitiveTurnPipeline
+from core.config import RuntimeConfig
+from core.engine_identity import IdentityContinuityError
+
+
+def _drive(state_dir: Path, n: int = 2, config: RuntimeConfig | None = None) -> None:
+    runtime = ChatRuntime(config=config or RuntimeConfig(), engine_state_path=state_dir)
+    pipe = CognitiveTurnPipeline(runtime=runtime)
+    for _ in range(n):
+        pipe.run("What causes light?")
+
+
+def test_same_substrate_reboot_has_no_break(tmp_path: Path) -> None:
+    state_dir = tmp_path / "es"
+    _drive(state_dir)
+    rt = ChatRuntime(config=RuntimeConfig(), engine_state_path=state_dir)
+    assert rt.identity_continuity_break is False
+
+
+def test_substrate_change_during_downtime_surfaces_break(tmp_path: Path) -> None:
+    state_dir = tmp_path / "es"
+    _drive(state_dir)  # stamps the default identity
+    # Boot under a different identity pack over the same checkpoint dir.
+    with pytest.warns(RuntimeWarning, match="identity continuity break"):
+        rt = ChatRuntime(
+            config=RuntimeConfig(identity_pack="precision_first_v1"),
+            engine_state_path=state_dir,
+        )
+    assert rt.identity_continuity_break is True
+
+
+def test_strict_mode_refuses_on_identity_mismatch(tmp_path: Path) -> None:
+    state_dir = tmp_path / "es"
+    _drive(state_dir)
+    with pytest.raises(IdentityContinuityError):
+        ChatRuntime(
+            config=RuntimeConfig(
+                identity_pack="precision_first_v1", strict_identity_continuity=True
+            ),
+            engine_state_path=state_dir,
+        )
+
+
+def test_strict_mode_allows_matching_identity(tmp_path: Path) -> None:
+    state_dir = tmp_path / "es"
+    _drive(state_dir)
+    # Same substrate + strict mode -> boots cleanly, no break.
+    rt = ChatRuntime(
+        config=RuntimeConfig(strict_identity_continuity=True),
+        engine_state_path=state_dir,
+    )
+    assert rt.identity_continuity_break is False
+
+
+def test_genesis_runtime_has_no_break(tmp_path: Path) -> None:
+    # No checkpoint to compare against -> no break.
+    rt = ChatRuntime(config=RuntimeConfig(), engine_state_path=tmp_path / "fresh")
+    assert rt.identity_continuity_break is False