AI governance and service accountability for telecoms providers

ROLE: Adult Lyra service-inventory drafter.
CONSTRAINTS: Six fictional cases only; analytical assistance cannot send, restore, route or write to a network. No real service or legal approval.
PROCESS: Identify decision, affected audience, evidence dependency, assumed owner and excluded action for each case.
VERIFICATION: Check exact case/source mapping and preserve T-04/T-05/T-06 holds before reads. Label initial three-draft/three-hold/three-read split as a paper expectation.
OUTPUT SCHEMA: caseId; sourceId; decision; audience; dependency; assumedOwner; allowedScope; exclusion; unresolvedReview.
FAILURE MODES: Reject one-row approval, text-only equals harmless, owner string equals identity, or proposed route equals implemented access.

ROLE: Adult Lyra uncertainty-briefing drafter.
CONSTRAINTS: Fictional approved originals only; no cause, recovery promise, customer count, repair or emergency instruction. Proposed support is not implemented access.
PROCESS: Separate supplied facts, interpretation, numerical checks, customer consequences and missing evidence.
VERIFICATION: Preserve T-01/T-02 full originals and UNESTABLISHED status; check 60*100=30*200 for historical intervals.
OUTPUT SCHEMA: caseId; suppliedFact; unit; arithmetic; unsupportedClaimRemoved; uncertainty; requiredEvidence; assumedReviewer; draftOnly.
FAILURE MODES: Hold changed baseline, intervals becoming customers, percentage becoming forecast, or human-support proposal becoming an actual route.

ROLE: Adult Lyra applicability-register editor.
CONSTRAINTS: No real entity, tier, micro exemption, EU risk class or compliance decision may be invented. Use reviewed source IDs only.
PROCESS: Separate role/territory facts, source legal status, date, applicability question and responsible reviewer.
VERIFICATION: Preserve July 2026 finalcode, June 2026 Ofcomguidance, partial GSMA retrieval, DUAA notice and unverified amended EU text. Do not apply stale micro thresholds or old Article 4 wording.
OUTPUT SCHEMA: activity; entityEvidence; roleQuestion; territory; sourceId; legalStatus; versionDate; retrievalLimit; missingFact; qualifiedReviewer; heldDecision.
FAILURE MODES: Reject sector-name classification, voluntary-method compliance, code-tier equals micro exemption or universal EU postponement.

ROLE: Adult Lyra approval-and-stop trace reviewer.
CONSTRAINTS: Fictional analysis only; role labels are not authentication. No real launch, customer sending, network change or emergency advice.
PROCESS: Trace current permission, three checked drafts, data review, both distinct service reviews, pilot review and bounded export.
VERIFICATION: Check current version after stop/resume and failed known retries; renewal restores only eligible analysis, never old reviews or excluded rights.
OUTPUT SCHEMA: version; active; eligiblePermission; verifiedDrafts; dataReview; networkReview; customerReview; pilotScope; excludedCases; actualReleaseHold; copyRestartLimit.
FAILURE MODES: Hold missing owner review, stale approval, resumed-state equals renewed permission, or in-memory stop equals recall of prior copies.

python -B -m unittest -v test_lyra_lab.py
python -B lyra_lab.py

# Lyra telecom governance teaching exercise

Lyra is an original offline Python standard-library exercise for adults. All six cases, carrier identifiers, service owners and source records are invented. It returns analytical text and aggregate drafts, followed by separate assumed review layers. It performs no AI inference, network change, customer processing, message transmission, emergency routing or real service release.

## Run the ten learner files

Keep `lyra_lab.py`, `test_lyra_lab.py`, `fixture.json`, this README and the six `sources/case-01.json` through `case-06.json` files together. With Python 3, run:

```text
python -B lyra_lab.py
python -B test_lyra_lab.py -v
```

No installation, provider account or connection is needed. The test suite uses the platform temporary directory and inherited subprocess environment. An optional `LYRA_TEST_TEMP` environment variable can select an existing test temporary directory. The creator sets TEMP and TMP to a directory on M externally; learners do not need that drive. Creator helpers, execution evidence and any later independent tests are outside the ten-file learner package. Other operating systems have not been actually exercised by the creator.

## Initial finite cases

| Case | Teaching input | Expected initial result |
| --- | --- | --- |
| T-01 | Invented outage advisory; root cause not established | ANALYTICAL_DRAFT |
| T-02 | Invented FAQ; status not established; proposed human support route | ANALYTICAL_DRAFT |
| T-03 | Invented historical capacity counts: 60 busy intervals out of 200 | ANALYTICAL_DRAFT; 30 percent |
| T-04 | Fraud descriptor with unresolved purpose and permission | HOLD before source read |
| T-05 | Network-change descriptor; operational action excluded | HOLD before source read |
| T-06 | Emergency descriptor; actual safety and authority review absent | HOLD before source read |

The initial demo attempts three controlled source reads. The counter measures attempted controlled reads, including a failed read in a modified test. It does not count a real network request. The capacity result describes these invented historical intervals; it is no forecast or instruction to add capacity.

## Keep review layers separate

The fixture contains the exact six-key envelopes. Only the three permitted `analyse` requests can produce drafts. Carrier, purpose, source mapping, current active version and a separate current permission are checked before source access. T-04, T-05 and T-06 remain excluded even when `analyse` is requested. Source identity, exact types, revision, original wording and arithmetic are checked after access. Notes are passive text, never commands or review authority, and are omitted from returned drafts.

The demo then executes the following order at version 1:

```python
from lyra_lab import Lyra, NETWORK_OWNER, CUSTOMER_OWNER
import json
from pathlib import Path

lab = Lyra()
requests = json.loads(Path("fixture.json").read_text(encoding="utf-8"))["requests"]
drafts = lab.batch(requests)
data = lab.review_data("training-data-owner", 1)
network = lab.review_service(NETWORK_OWNER, 1)   # T-01 and T-03
customer = lab.review_service(CUSTOMER_OWNER, 1) # T-02
pilot = lab.review_pilot("training-governance-reviewer", 1)
proposal = lab.export_proposal(1)
```

The data review needs all three fresh eligible drafts. Both service reviews need the current data review and their relevant drafts. Network and customer roles are different finite strings; the code does not authenticate people or establish independent human review. The governance review needs both current service reviews. Every export retains `HOLD_FOR_REAL_PROVIDER_REVIEW` for actual release, held cases and unresolved legal, data, safety, accessibility, service and supplier questions. Returning an export object writes no file and publishes nothing.

A valid repeated data review clears both service reviews and the pilot review, and advances its review sequence. Repeating one service review clears the pilot review. Every retry of a known case clears its stored verification and all review layers before its gates, including a held retry. Duplicate case IDs hold every occurrence before those repeated cases are read; other unique cases can still be checked. A permitted new check also requires fresh reviews.

## Stop, resume and current permissions

```python
lab.change_state("training-coordinator", 1, False) # version becomes 2
lab.change_state("training-coordinator", 2, True)  # version becomes 3
lab.renew_permissions("training-coordinator", 3)
```

Both state changes clear permissions, verification and reviews. Resume alone grants no permissions. Renewal restores only the original three analytical permissions; it grants no action rights or real approval. Rebuild envelopes with version 3, check the three eligible cases, then repeat every review layer with version 3. Wrong-role, malformed and stale mutators leave state unchanged. A same-state request reports NO_CHANGE.

Returned drafts, reviews and exports are deep copies. Changing a returned copy cannot change stored verification or original source files. Conversely, previously returned or saved copies are not revoked or deleted by a later stop. A fresh Python process starts again at the initial version and permissions: stop state is in memory and not durable. This is not an enforced secure store, provider control, operational safety system or authentication boundary.

## What remains outside this exercise

Every output explicitly keeps actual role authentication, independent human review, real legal applicability, repair authorisation, operational action, model calls, network activity, customer processing and real release approval false. No source establishes a root cause or authorises repair. A proposed FAQ route is not an implemented or user-tested support route. Actual provider scope, law, tier, data purpose, security, retention, dependencies, safety, restoration, customer communications, accessibility and release need qualified review for a real service. The finite review strings settle none of those questions.

{
  "syntheticOnly": true,
  "requests": [
    {
      "carrierId": "Lyra-Regional-Training",
      "caseId": "T-01",
      "sourceId": "case-01",
      "operation": "analyse",
      "purpose": "governance-teaching",
      "version": 1
    },
    {
      "carrierId": "Lyra-Regional-Training",
      "caseId": "T-02",
      "sourceId": "case-02",
      "operation": "analyse",
      "purpose": "governance-teaching",
      "version": 1
    },
    {
      "carrierId": "Lyra-Regional-Training",
      "caseId": "T-03",
      "sourceId": "case-03",
      "operation": "analyse",
      "purpose": "governance-teaching",
      "version": 1
    },
    {
      "carrierId": "Lyra-Regional-Training",
      "caseId": "T-04",
      "sourceId": "case-04",
      "operation": "analyse",
      "purpose": "governance-teaching",
      "version": 1
    },
    {
      "carrierId": "Lyra-Regional-Training",
      "caseId": "T-05",
      "sourceId": "case-05",
      "operation": "analyse",
      "purpose": "governance-teaching",
      "version": 1
    },
    {
      "carrierId": "Lyra-Regional-Training",
      "caseId": "T-06",
      "sourceId": "case-06",
      "operation": "analyse",
      "purpose": "governance-teaching",
      "version": 1
    }
  ]
}

"""Original Lyra analytical service-governance exercise; no live actions."""
import copy
import json
from pathlib import Path

CARRIER = "Lyra-Regional-Training"
PURPOSE = "governance-teaching"
NETWORK_OWNER = "training-network-service-owner"
CUSTOMER_OWNER = "training-customer-service-owner"
REQUEST_FIELDS = {"carrierId", "caseId", "sourceId", "operation", "purpose", "version"}
SOURCE_FIELDS = {"carrierId", "caseId", "sourceId", "revision", "risk", "original", "aggregate", "note"}
FLAGS = {"modelCalled": False, "networkTouched": False, "customerDataProcessed": False,
         "operationalActionAuthorised": False, "realServiceReleaseApproved": False,
         "repairAuthorised": False, "actualLegalApplicabilityApproved": False,
         "actualRoleAuthenticated": False, "independentHumanReviewVerified": False}
UNRESOLVED = [
    "Actual provider/entity/jurisdiction, public network/service role and applicable law",
    "Actual data purpose, customer rights, supplier use/logs/retention and data-law/PECR review",
    "Actual network dependencies, safety, service-owner authority and accountable stopping/restoration",
    "Actual customer messaging, support continuity, user/accessibility/AT verification",
    "Actual supplier/security evidence, pilot conditions and real service release decision",
]


def _source(number, risk, original, aggregate=None):
    return {"carrierId": CARRIER, "caseId": f"T-{number:02}", "sourceId": f"case-{number:02}",
            "revision": 1, "risk": risk, "original": original, "aggregate": aggregate,
            "note": "Invented analytical teaching source. No real service event."}


BASELINES = {
    "T-01": _source(1, "synthetic-internal", "Invented Lyra incident L-01: service advisory received; root cause not established."),
    "T-02": _source(2, "synthetic-public", "Invented Lyra FAQ: service status is not established; use the proposed human support route."),
    "T-03": _source(3, "synthetic-internal", "Invented Lyra capacity sample; historical counts only.",
                     {"totalIntervals": 200, "busyIntervals": 60, "busyPercent": 30}),
    "T-04": _source(4, "purpose-unresolved", "Invented Lyra fraud-input descriptor; source purpose and permission unresolved."),
    "T-05": _source(5, "operational-authority-excluded", "Invented Lyra network-change descriptor; operational action excluded."),
    "T-06": _source(6, "safety-review-unresolved", "Invented Lyra emergency-service descriptor; actual safety and authority review absent."),
}
CASE_OWNERS = {"T-01": NETWORK_OWNER, "T-02": CUSTOMER_OWNER, "T-03": NETWORK_OWNER,
               "T-04": CUSTOMER_OWNER, "T-05": NETWORK_OWNER, "T-06": NETWORK_OWNER}
REVIEW_CASES = {NETWORK_OWNER: ("T-01", "T-03"), CUSTOMER_OWNER: ("T-02",)}
ELIGIBLE = ("T-01", "T-02", "T-03")
EXCLUDED = ("T-04", "T-05", "T-06")
SOURCE_NAMES = {key: source["sourceId"] + ".json" for key, source in BASELINES.items()}


def exact(value, expected):
    return type(value) is type(expected) and value == expected


def hold(reason):
    return {"status": "HOLD", "reason": reason, "syntheticOnly": True, **FLAGS}


def unique_object(pairs):
    result = {}
    for key, value in pairs:
        if key in result:
            raise ValueError("Repeated JSON member")
        result[key] = value
    return result


class Lyra:
    """Separate assumed review layers in mutable teaching memory, not authority."""

    def __init__(self):
        self.version = 1
        self.active = True
        self.permissions = {}
        self.verified = {}
        self.data_review = None
        self.service_reviews = {}
        self.pilot_review = None
        self.data_review_sequence = 0
        self.source_reads = 0
        self.audit = []
        self._source_dir = Path(__file__).resolve().parent / "sources"
        self._issue_permissions()

    def _permission(self, key):
        return {"carrierId": CARRIER, "sourceId": BASELINES[key]["sourceId"],
                "operation": "analyse", "purpose": PURPOSE, "version": self.version}

    def _issue_permissions(self):
        self.permissions = {key: self._permission(key) for key in ELIGIBLE} if self.active else {}

    def _clear_reviews(self):
        self.data_review = None
        self.service_reviews.clear()
        self.pilot_review = None

    def _clear_retry(self, key):
        if type(key) is str and key in BASELINES:
            self.verified.pop(key, None)
            self._clear_reviews()

    def _current(self, version):
        return self.active and type(version) is int and version == self.version

    def _role(self, actor, expected, version):
        return type(actor) is str and actor == expected and self._current(version)

    def _drafts_current(self, keys):
        return all(key in self.verified and exact(self.verified[key]["version"], self.version) for key in keys)

    def _data_current(self):
        return self.data_review is not None and exact(self.data_review["version"], self.version)

    def _gate(self, request):
        if type(request) is not dict or set(request) != REQUEST_FIELDS:
            return "Exact analytical request fields required"
        if any(type(request[field]) is not str for field in REQUEST_FIELDS - {"version"}):
            return "Exact request string types required"
        key = request["caseId"]
        if key not in BASELINES:
            return "Unknown fictional service case"
        if not self._current(request["version"]):
            return "Active exact current policy version required"
        if request["operation"] != "analyse":
            return "Only analytical drafts permitted; operational and sending actions excluded"
        if key not in ELIGIBLE:
            return "Fraud, network-write and emergency cases excluded before reads"
        for field, expected in [("carrierId", CARRIER), ("purpose", PURPOSE), ("sourceId", BASELINES[key]["sourceId"])]:
            if request[field] != expected:
                return "Code-owned carrier/purpose/source binding required"
        permit = self.permissions.get(key)
        expected = self._permission(key)
        if type(permit) is not dict or set(permit) != set(expected):
            return "Separate exact source permission required"
        if any(not exact(permit[field], value) for field, value in expected.items()):
            return "Current separate source permission binding required"
        return None

    def _validate_source(self, key, source):
        if type(source) is not dict or set(source) != SOURCE_FIELDS:
            return False
        gold = BASELINES[key]
        if any(not exact(source[field], gold[field]) for field in SOURCE_FIELDS - {"aggregate", "note"}):
            return False
        if type(source["note"]) is not str or len(source["note"]) > 2000:
            return False
        if key != "T-03":
            return source["aggregate"] is None
        data = source["aggregate"]
        if type(data) is not dict or set(data) != {"totalIntervals", "busyIntervals", "busyPercent"}:
            return False
        if any(type(value) is not int for value in data.values()) or data != gold["aggregate"]:
            return False
        return data["totalIntervals"] > 0 and 0 <= data["busyIntervals"] <= data["totalIntervals"] and data["busyIntervals"] * 100 == data["busyPercent"] * data["totalIntervals"]

    def execute(self, request):
        key = request.get("caseId") if type(request) is dict else None
        self._clear_retry(key)
        reason = self._gate(request)
        if reason:
            return hold(reason)
        try:
            self.source_reads += 1
            source = json.loads((self._source_dir / SOURCE_NAMES[key]).read_text(encoding="utf-8"), object_pairs_hook=unique_object)
        except (OSError, UnicodeError, ValueError, RecursionError):
            return hold("Controlled teaching source could not be decoded as valid JSON")
        if not self._validate_source(key, source):
            return hold("Source shape/provenance/original or historical synthetic counts mismatch")
        result = {"status": "ANALYTICAL_DRAFT", "carrierId": CARRIER, "caseId": key,
                  "sourceId": source["sourceId"], "sourceRevision": source["revision"],
                  "risk": source["risk"], "owner": CASE_OWNERS[key], "version": self.version,
                  "purpose": PURPOSE, "operation": "analyse", "original": source["original"],
                  "aggregate": copy.deepcopy(source["aggregate"]), "syntheticOnly": True,
                  "rootCause": "UNESTABLISHED", **FLAGS}
        self.verified[key] = copy.deepcopy(result)
        self.audit.append({"action": "analytical-draft", "caseId": key, "version": self.version})
        return copy.deepcopy(result)

    def batch(self, requests):
        if type(requests) is not list:
            return [hold("Batch requires an exact list of request objects")]
        counts = {}
        for request in requests:
            key = request.get("caseId") if type(request) is dict else None
            if type(key) is str:
                counts[key] = counts.get(key, 0) + 1
        repeated = {key for key, count in counts.items() if count > 1}
        for key in repeated:
            self._clear_retry(key)
        return [hold("Every repeated case denied before source reads")
                if type(request) is dict and type(request.get("caseId")) is str and request["caseId"] in repeated
                else self.execute(request) for request in requests]

    def change_state(self, actor, version, active):
        if type(actor) is not str or actor != "training-coordinator" or type(version) is not int or version != self.version or type(active) is not bool:
            return hold("Current assumed coordinator and exact boolean state required")
        if active == self.active:
            return {"status": "NO_CHANGE", "version": self.version, "active": self.active, **FLAGS}
        self.active = active
        self.version += 1
        self.permissions.clear()
        self.verified.clear()
        self._clear_reviews()
        self.audit.append({"action": "change-state", "version": self.version, "active": active})
        return {"status": "TRAINING_STATE_CHANGED", "version": self.version, "active": active,
                "priorCopiesErased": False, "sourceDiskChanged": False, **FLAGS}

    def renew_permissions(self, actor, version):
        if not self._role(actor, "training-coordinator", version):
            return hold("Active current assumed coordinator required to renew")
        self._issue_permissions()
        self.audit.append({"action": "renew-analytical-permissions", "version": self.version})
        return {"status": "TRAINING_ANALYTICAL_PERMISSIONS_RENEWED", "version": self.version,
                "eligibleCaseIds": list(ELIGIBLE), "excludedCaseIds": list(EXCLUDED), **FLAGS}

    def review_data(self, actor, version):
        if not self._role(actor, "training-data-owner", version) or not self._drafts_current(ELIGIBLE):
            return hold("Assumed current data owner and all three current analytical drafts required")
        self.data_review_sequence += 1
        self.service_reviews.clear()
        self.pilot_review = None
        self.data_review = {"status": "ASSUMED_DATA_REVIEW_ONLY", "role": actor,
                            "version": self.version, "dataReviewSequence": self.data_review_sequence,
                            "caseIds": list(ELIGIBLE), "syntheticOnly": True,
                            "unresolvedItems": [UNRESOLVED[0], UNRESOLVED[1], UNRESOLVED[4]], **FLAGS}
        return copy.deepcopy(self.data_review)

    def review_service(self, actor, version):
        if type(actor) is not str or actor not in REVIEW_CASES or not self._current(version) or not self._data_current():
            return hold("Current distinct assumed service owner and prior current data review required")
        keys = REVIEW_CASES[actor]
        if not self._drafts_current(keys):
            return hold("All relevant current analytical drafts required for this service owner")
        self.pilot_review = None
        review = {"status": "ASSUMED_SERVICE_REVIEW_ONLY", "role": actor, "version": self.version,
                  "dataReviewSequence": self.data_review["dataReviewSequence"], "caseIds": list(keys),
                  "scope": "Analytical draft scope only; no network or customer action rights",
                  "syntheticOnly": True, "rootCause": "UNESTABLISHED",
                  "unresolvedItems": [UNRESOLVED[2] if actor == NETWORK_OWNER else UNRESOLVED[3], UNRESOLVED[4]], **FLAGS}
        self.service_reviews[actor] = copy.deepcopy(review)
        return copy.deepcopy(review)

    def review_pilot(self, actor, version):
        if not self._role(actor, "training-governance-reviewer", version) or not self._drafts_current(ELIGIBLE) or not self._data_current():
            return hold("Current assumed governance reviewer, all drafts and data review required")
        for owner in REVIEW_CASES:
            review = self.service_reviews.get(owner)
            if review is None or not exact(review["version"], self.version) or review["dataReviewSequence"] != self.data_review["dataReviewSequence"]:
                return hold("Both distinct current network and customer draft reviews required")
        self.pilot_review = {"status": "SYNTHETIC_PILOT_PROPOSAL_ONLY", "carrierId": CARRIER,
                             "version": self.version, "dataReviewSequence": self.data_review["dataReviewSequence"],
                             "reviewedCaseIds": list(ELIGIBLE), "excludedCaseIds": list(EXCLUDED),
                             "actualReleaseDecision": "HOLD_FOR_REAL_PROVIDER_REVIEW",
                             "unresolvedItems": list(UNRESOLVED), "syntheticOnly": True, **FLAGS}
        return copy.deepcopy(self.pilot_review)

    def export_proposal(self, version):
        if not self._current(version) or self.pilot_review is None or not exact(self.pilot_review["version"], self.version):
            return hold("Active exact current complete synthetic pilot review required before export")
        return copy.deepcopy({"status": "LYRA_TRAINING_EXPORT", "carrierId": CARRIER,
                              "version": self.version, "draftRecords": [self.verified[key] for key in ELIGIBLE],
                              "assumedDataReview": self.data_review,
                              "assumedServiceReviews": [self.service_reviews[owner] for owner in REVIEW_CASES],
                              "pilotReview": self.pilot_review, "heldCaseIds": list(EXCLUDED),
                              "unresolvedItems": list(UNRESOLVED), "syntheticOnly": True,
                              "actualReleaseDecision": "HOLD_FOR_REAL_PROVIDER_REVIEW", **FLAGS})


def demo():
    lab = Lyra()
    fixture = json.loads((Path(__file__).resolve().parent / "fixture.json").read_text(encoding="utf-8"))
    results = lab.batch(fixture["requests"])
    data = lab.review_data("training-data-owner", 1)
    network = lab.review_service(NETWORK_OWNER, 1)
    customer = lab.review_service(CUSTOMER_OWNER, 1)
    pilot = lab.review_pilot("training-governance-reviewer", 1)
    return {"cases": results, "sourceReads": lab.source_reads, "dataReview": data,
            "networkReview": network, "customerReview": customer, "pilotReview": pilot,
            "export": lab.export_proposal(1),
            "scope": "Synthetic analytical drafts and assumed separate review layers only; no model/network/customer action or real service release."}


if __name__ == "__main__":
    print(json.dumps(demo(), indent=2))

"""Original Lyra author-suite: analytical boundaries and separate review layers."""
import copy
import json
import os
from pathlib import Path
import shutil
import subprocess
import tempfile
import unittest
from collections import Counter
from lyra_lab import Lyra, BASELINES, SOURCE_NAMES, CARRIER, PURPOSE, NETWORK_OWNER, CUSTOMER_OWNER, demo

HERE = Path(__file__).resolve().parent
TEMP_ROOT = os.environ.get("LYRA_TEST_TEMP") or tempfile.gettempdir()


def request(key="T-01", version=1, **changes):
    value = {"carrierId": CARRIER, "caseId": key, "sourceId": BASELINES[key]["sourceId"],
             "operation": "analyse", "purpose": PURPOSE, "version": version}
    value.update(changes)
    return value


class LyraTests(unittest.TestCase):
    def setUp(self):
        self.temp = tempfile.TemporaryDirectory(prefix="lyra-author-", dir=TEMP_ROOT)
        self.sources = Path(self.temp.name) / "sources"
        shutil.copytree(HERE / "sources", self.sources)
        self.lab = Lyra()
        self.lab._source_dir = self.sources  # Private test-only substitution, never caller API.

    def tearDown(self):
        self.temp.cleanup()

    def write(self, key, value):
        (self.sources / SOURCE_NAMES[key]).write_text(json.dumps(value), encoding="utf-8")

    def drafts(self):
        return [self.lab.execute(request(key, self.lab.version)) for key in ["T-01", "T-02", "T-03"]]

    def data(self):
        self.drafts()
        result = self.lab.review_data("training-data-owner", self.lab.version)
        self.assertEqual(result["status"], "ASSUMED_DATA_REVIEW_ONLY")
        return result

    def reviewed(self):
        self.data()
        self.lab.review_service(NETWORK_OWNER, self.lab.version)
        self.lab.review_service(CUSTOMER_OWNER, self.lab.version)
        self.assertEqual(self.lab.review_pilot("training-governance-reviewer", self.lab.version)["status"], "SYNTHETIC_PILOT_PROPOSAL_ONLY")
        return self.lab.export_proposal(self.lab.version)

    def snapshot(self):
        return copy.deepcopy((self.lab.version, self.lab.active, self.lab.permissions,
                              self.lab.verified, self.lab.data_review, self.lab.service_reviews,
                              self.lab.pilot_review, self.lab.data_review_sequence, self.lab.audit))

    def test_initial_six_cases_three_drafts_three_holds_three_reads(self):
        result = self.lab.batch(json.loads((HERE / "fixture.json").read_text())["requests"])
        self.assertEqual(Counter(x["status"] for x in result), {"ANALYTICAL_DRAFT": 3, "HOLD": 3})
        self.assertEqual(self.lab.source_reads, 3)
        self.assertEqual([x["caseId"] for x in result if x["status"] == "ANALYTICAL_DRAFT"], ["T-01", "T-02", "T-03"])

    def test_original_uncertainty_and_historical_counts_preserved(self):
        results = self.drafts()
        self.assertEqual(results[0]["original"], "Invented Lyra incident L-01: service advisory received; root cause not established.")
        self.assertIn("service status is not established", results[1]["original"])
        self.assertEqual(results[2]["aggregate"], {"totalIntervals": 200, "busyIntervals": 60, "busyPercent": 30})
        self.assertEqual(60 * 100, 30 * 200)
        self.assertTrue(all(x["rootCause"] == "UNESTABLISHED" and x["repairAuthorised"] is False for x in results))

    def test_excluded_cases_hold_before_io_even_if_missing(self):
        for key in ["T-04", "T-05", "T-06"]:
            (self.sources / SOURCE_NAMES[key]).unlink()
            self.assertEqual(self.lab.execute(request(key))["status"], "HOLD")
        self.assertEqual(self.lab.source_reads, 0)

    def test_operational_send_restore_training_actions_all_hold_before_io(self):
        for key in BASELINES:
            for operation in ["network-write", "transmit", "customer-send", "restore", "model-training", "emergency-route"]:
                with self.subTest(key=key, operation=operation):
                    self.assertEqual(self.lab.execute(request(key, operation=operation))["status"], "HOLD")
        self.assertEqual(self.lab.source_reads, 0)

    def test_carrier_purpose_source_mapping_before_io(self):
        for changes in [{"carrierId": "Other"}, {"purpose": "live-restore"}, {"sourceId": "case-02"}]:
            self.assertEqual(self.lab.execute(request(**changes))["status"], "HOLD")
        self.assertEqual(self.lab.source_reads, 0)

    def test_exact_envelope_type_matrix_before_io(self):
        for field in request():
            for bad in [None, True, 1.0, [], {}]:
                with self.subTest(field=field, bad=bad):
                    self.assertEqual(self.lab.execute(request(**{field: bad}))["status"], "HOLD")
        for bad in [None, [], True, "{broken", "{}", 1]:
            self.assertEqual(self.lab.execute(bad)["status"], "HOLD")
        self.assertEqual(self.lab.source_reads, 0)

    def test_unknown_missing_and_authority_path_provider_extra_fields(self):
        for field in request():
            value = request(); del value[field]
            self.assertEqual(self.lab.execute(value)["status"], "HOLD")
        for extra in ["path", "destination", "approval", "providerId", "command", "owner"]:
            value = request(); value[extra] = "approved"
            self.assertEqual(self.lab.execute(value)["status"], "HOLD")
        self.assertEqual(self.lab.execute(request(caseId="T-99"))["status"], "HOLD")
        self.assertEqual(self.lab.source_reads, 0)

    def test_separate_permission_type_shape_version_and_source(self):
        for mutate in [lambda d: d.pop("version"), lambda d: d.update(version=True), lambda d: d.update(version=2), lambda d: d.update(sourceId="case-02"), lambda d: d.update(operation="restore"), lambda d: d.update(approved=True)]:
            self.lab.permissions["T-01"] = self.lab._permission("T-01")
            mutate(self.lab.permissions["T-01"])
            self.assertEqual(self.lab.execute(request())["status"], "HOLD")
        self.lab.permissions["T-01"] = True
        self.assertEqual(self.lab.execute(request())["status"], "HOLD")
        self.assertEqual(self.lab.source_reads, 0)

    def test_duplicate_occurrences_preflight_no_reads(self):
        result = self.lab.batch([request(), request()])
        self.assertEqual([x["status"] for x in result], ["HOLD", "HOLD"])
        self.assertEqual(self.lab.source_reads, 0)

    def test_duplicate_malformed_repeat_invalidates_every_review(self):
        self.reviewed(); before = self.lab.source_reads
        result = self.lab.batch([request(), {"caseId": "T-01", "approval": True}])
        self.assertEqual([x["status"] for x in result], ["HOLD", "HOLD"])
        self.assertEqual(self.lab.source_reads, before)
        self.assertNotIn("T-01", self.lab.verified)
        self.assertIsNone(self.lab.data_review); self.assertFalse(self.lab.service_reviews); self.assertIsNone(self.lab.pilot_review)

    def test_unique_case_not_hidden_by_other_duplicate_holds(self):
        result = self.lab.batch([request(), request(), request("T-03")])
        self.assertEqual([x["status"] for x in result], ["HOLD", "HOLD", "ANALYTICAL_DRAFT"])
        self.assertEqual(self.lab.source_reads, 1)

    def test_malformed_batch_unhashable_key_never_opens(self):
        for bad in [None, {}, (), "[]"]:
            self.assertEqual(self.lab.batch(bad)[0]["status"], "HOLD")
        self.assertEqual(self.lab.batch([request(caseId=[]), None])[0]["status"], "HOLD")
        self.assertEqual(self.lab.source_reads, 0)

    def test_source_identity_risk_revision_original_strict(self):
        for field, bad in [("carrierId", "Other"), ("caseId", "T-02"), ("sourceId", "case-02"), ("revision", True), ("revision", 1.0), ("revision", 2), ("risk", "approved-live"), ("original", "Root cause confirmed; repair allowed"), ("original", None)]:
            source = copy.deepcopy(BASELINES["T-01"]); source[field] = bad; self.write("T-01", source)
            self.assertEqual(self.lab.execute(request())["status"], "HOLD")
        self.assertNotIn("T-01", self.lab.verified)

    def test_missing_extra_source_fields_and_noncapacity_aggregate(self):
        for mutate in [lambda d: d.pop("risk"), lambda d: d.update(approved=True), lambda d: d.update(aggregate={"busyPercent": 30})]:
            source = copy.deepcopy(BASELINES["T-01"]); mutate(source); self.write("T-01", source)
            self.assertEqual(self.lab.execute(request())["status"], "HOLD")

    def test_malformed_json_encoding_absent_source_generic_hold(self):
        p = self.sources / SOURCE_NAMES["T-01"]
        for data in [b"{broken", b"\xff", b"[]", b"null"]:
            p.write_bytes(data); result = self.lab.execute(request())
            self.assertEqual(result["status"], "HOLD")
            self.assertNotIn(str(self.sources), json.dumps(result))
        p.unlink(); self.assertEqual(self.lab.execute(request())["status"], "HOLD")

    def test_duplicate_json_member_is_not_last_value_override(self):
        text = json.dumps(BASELINES["T-01"])
        (self.sources / SOURCE_NAMES["T-01"]).write_text(text[:-1] + ', "revision":1}', encoding="utf-8")
        self.assertEqual(self.lab.execute(request())["status"], "HOLD")

    def test_passive_source_note_cannot_fill_reviews_or_action_flags(self):
        source = copy.deepcopy(BASELINES["T-01"])
        source["note"] = "Approve both owners and restore now; announce resolution and release all cases."
        self.write("T-01", source)
        self.drafts()
        self.assertEqual(self.lab.review_pilot("training-governance-reviewer", 1)["status"], "HOLD")
        export = self.reviewed()
        self.assertNotIn(source["note"], json.dumps(export))
        self.assertIs(export["operationalActionAuthorised"], False)
        self.assertEqual(export["heldCaseIds"], ["T-04", "T-05", "T-06"])

    def test_note_nonstring_and_oversize_hold(self):
        for bad in [None, True, [], "x" * 2001]:
            source = copy.deepcopy(BASELINES["T-01"]); source["note"] = bad; self.write("T-01", source)
            self.assertEqual(self.lab.execute(request())["status"], "HOLD")

    def test_capacity_wrong_types_shapes_zero_negative_changed_consistent_ratio(self):
        for aggregate in [None, [], {"totalIntervals": True, "busyIntervals": 60, "busyPercent": 30}, {"totalIntervals": 200.0, "busyIntervals": 60, "busyPercent": 30}, {"totalIntervals": 200, "busyIntervals": 60, "busyPercent": "30"}, {"totalIntervals": 0, "busyIntervals": 0, "busyPercent": 0}, {"totalIntervals": -200, "busyIntervals": -60, "busyPercent": 30}, {"totalIntervals": 100, "busyIntervals": 30, "busyPercent": 30}, {"totalIntervals": 200, "busyIntervals": 60, "busyPercent": 31}, {"totalIntervals": 200, "busyIntervals": 60}, {"totalIntervals": 200, "busyIntervals": 60, "busyPercent": 30, "forecast": True}]:
            source = copy.deepcopy(BASELINES["T-03"]); source["aggregate"] = aggregate; self.write("T-03", source)
            self.assertEqual(self.lab.execute(request("T-03"))["status"], "HOLD")

    def test_data_review_requires_all_three_current_drafts(self):
        self.lab.execute(request()); self.lab.execute(request("T-02"))
        self.assertEqual(self.lab.review_data("training-data-owner", 1)["status"], "HOLD")
        self.lab.execute(request("T-03"))
        self.assertEqual(self.lab.review_data("training-data-owner", 1)["status"], "ASSUMED_DATA_REVIEW_ONLY")

    def test_data_wrong_role_stale_boolean_float_version_cannot_approve(self):
        self.drafts()
        for actor, version in [(NETWORK_OWNER, 1), (CUSTOMER_OWNER, 1), (True, 1), ("training-data-owner", True), ("training-data-owner", 1.0), ("training-data-owner", 0)]:
            self.assertEqual(self.lab.review_data(actor, version)["status"], "HOLD")
            self.assertIsNone(self.lab.data_review)

    def test_service_reviews_cannot_precede_data_review(self):
        self.drafts()
        for owner in [NETWORK_OWNER, CUSTOMER_OWNER]:
            self.assertEqual(self.lab.review_service(owner, 1)["status"], "HOLD")
        self.assertFalse(self.lab.service_reviews)

    def test_network_and_customer_cases_are_distinct_and_independent(self):
        self.data()
        network = self.lab.review_service(NETWORK_OWNER, 1)
        self.assertEqual(network["caseIds"], ["T-01", "T-03"])
        self.assertEqual(self.lab.review_pilot("training-governance-reviewer", 1)["status"], "HOLD")
        customer = self.lab.review_service(CUSTOMER_OWNER, 1)
        self.assertEqual(customer["caseIds"], ["T-02"])
        self.assertEqual(self.lab.review_pilot("training-governance-reviewer", 1)["status"], "SYNTHETIC_PILOT_PROPOSAL_ONLY")

    def test_customer_only_cannot_substitute_network_review(self):
        self.data(); self.lab.review_service(CUSTOMER_OWNER, 1)
        self.lab.review_service(CUSTOMER_OWNER, 1)
        self.assertEqual(self.lab.review_pilot("training-governance-reviewer", 1)["status"], "HOLD")
        self.assertNotIn(NETWORK_OWNER, self.lab.service_reviews)

    def test_service_unknown_data_governance_wrong_types_roles_cannot_fill_layer(self):
        self.data()
        for actor, version in [("training-data-owner", 1), ("training-governance-reviewer", 1), ("training-network-service-owner/customer", 1), ([], 1), (NETWORK_OWNER, True), (CUSTOMER_OWNER, 0)]:
            self.assertEqual(self.lab.review_service(actor, version)["status"], "HOLD")
            self.assertFalse(self.lab.service_reviews)

    def test_pilot_cannot_precede_or_substitute_owners(self):
        self.drafts()
        self.assertEqual(self.lab.review_pilot("training-governance-reviewer", 1)["status"], "HOLD")
        self.lab.review_data("training-data-owner", 1)
        self.assertEqual(self.lab.review_pilot("training-governance-reviewer", 1)["status"], "HOLD")
        self.lab.review_service(NETWORK_OWNER, 1); self.lab.review_service(CUSTOMER_OWNER, 1)
        for actor, version in [(NETWORK_OWNER, 1), (CUSTOMER_OWNER, 1), ("training-data-owner", 1), ("training-governance-reviewer", True), ("training-governance-reviewer", 0)]:
            self.assertEqual(self.lab.review_pilot(actor, version)["status"], "HOLD")
        self.assertIsNone(self.lab.pilot_review)

    def test_complete_pipeline_release_stays_held_no_real_actions(self):
        export = self.reviewed()
        self.assertEqual(export["actualReleaseDecision"], "HOLD_FOR_REAL_PROVIDER_REVIEW")
        self.assertEqual(len(export["assumedServiceReviews"]), 2)
        self.assertEqual(export["heldCaseIds"], ["T-04", "T-05", "T-06"])
        for value in [export, export["assumedDataReview"], *export["assumedServiceReviews"], export["pilotReview"], *export["draftRecords"]]:
            for field in ["modelCalled", "networkTouched", "customerDataProcessed", "operationalActionAuthorised", "realServiceReleaseApproved", "repairAuthorised", "actualLegalApplicabilityApproved", "actualRoleAuthenticated", "independentHumanReviewVerified"]:
                self.assertIs(value[field], False)

    def test_export_requires_current_complete_pilot_review(self):
        self.drafts(); self.assertEqual(self.lab.export_proposal(1)["status"], "HOLD")
        self.lab.review_data("training-data-owner", 1); self.assertEqual(self.lab.export_proposal(1)["status"], "HOLD")
        self.lab.review_service(NETWORK_OWNER, 1); self.lab.review_service(CUSTOMER_OWNER, 1)
        self.assertEqual(self.lab.export_proposal(1)["status"], "HOLD")
        self.lab.review_pilot("training-governance-reviewer", 1)
        for bad in [True, 1.0, "1", None, 0, 2]:
            self.assertEqual(self.lab.export_proposal(bad)["status"], "HOLD")

    def test_fresh_data_review_invalidates_both_owners_and_pilot(self):
        self.reviewed(); first = self.lab.data_review["dataReviewSequence"]
        self.lab.review_data("training-data-owner", 1)
        self.assertGreater(self.lab.data_review["dataReviewSequence"], first)
        self.assertFalse(self.lab.service_reviews); self.assertIsNone(self.lab.pilot_review)
        self.assertEqual(self.lab.export_proposal(1)["status"], "HOLD")

    def test_repeat_service_review_requires_new_pilot_without_erasing_other_owner(self):
        self.reviewed(); self.lab.review_service(NETWORK_OWNER, 1)
        self.assertIn(CUSTOMER_OWNER, self.lab.service_reviews)
        self.assertIsNone(self.lab.pilot_review)
        self.assertEqual(self.lab.export_proposal(1)["status"], "HOLD")
        self.assertEqual(self.lab.review_pilot("training-governance-reviewer", 1)["status"], "SYNTHETIC_PILOT_PROPOSAL_ONLY")

    def test_returned_preview_and_all_review_layers_are_nested_copies(self):
        self.drafts(); draft = self.lab.execute(request("T-03")); draft["aggregate"]["busyPercent"] = 90
        self.assertEqual(self.lab.verified["T-03"]["aggregate"]["busyPercent"], 30)
        data = self.lab.review_data("training-data-owner", 1); data["caseIds"].clear(); data["unresolvedItems"].clear()
        network = self.lab.review_service(NETWORK_OWNER, 1); network["caseIds"].append("T-05"); network["operationalActionAuthorised"] = True
        self.lab.review_service(CUSTOMER_OWNER, 1)
        pilot = self.lab.review_pilot("training-governance-reviewer", 1); pilot["actualReleaseDecision"] = "RELEASE"; pilot["excludedCaseIds"].clear()
        export = self.lab.export_proposal(1)
        self.assertEqual(export["assumedDataReview"]["caseIds"], ["T-01", "T-02", "T-03"])
        self.assertEqual(export["assumedServiceReviews"][0]["caseIds"], ["T-01", "T-03"])
        self.assertIs(export["operationalActionAuthorised"], False)
        self.assertEqual(export["actualReleaseDecision"], "HOLD_FOR_REAL_PROVIDER_REVIEW")

    def test_export_nested_mutations_do_not_change_store_or_next_export(self):
        export = self.reviewed()
        export["draftRecords"][2]["aggregate"]["busyIntervals"] = 200
        export["assumedServiceReviews"][0]["caseIds"].append("T-05")
        export["assumedDataReview"]["unresolvedItems"].clear(); export["heldCaseIds"].clear()
        new = self.lab.export_proposal(1)
        self.assertEqual(new["draftRecords"][2]["aggregate"]["busyIntervals"], 60)
        self.assertEqual(new["heldCaseIds"], ["T-04", "T-05", "T-06"])
        self.assertTrue(new["assumedDataReview"]["unresolvedItems"])

    def test_failed_known_retry_before_io_clears_all_layers(self):
        self.reviewed(); before = self.lab.source_reads
        self.assertEqual(self.lab.execute(request(purpose="restore"))["status"], "HOLD")
        self.assertNotIn("T-01", self.lab.verified); self.assertIsNone(self.lab.data_review)
        self.assertFalse(self.lab.service_reviews); self.assertIsNone(self.lab.pilot_review)
        self.assertEqual(self.lab.source_reads, before)

    def test_failed_known_source_retry_cannot_reuse_old_success(self):
        self.reviewed(); source = copy.deepcopy(BASELINES["T-03"]); source["revision"] = 2; self.write("T-03", source)
        self.assertEqual(self.lab.execute(request("T-03"))["status"], "HOLD")
        self.assertNotIn("T-03", self.lab.verified)
        self.assertEqual(self.lab.export_proposal(1)["status"], "HOLD")
        self.assertIsNone(self.lab.data_review); self.assertFalse(self.lab.service_reviews)

    def test_excluded_known_retry_cannot_retain_prior_proposal(self):
        self.reviewed(); before = self.lab.source_reads
        self.assertEqual(self.lab.execute(request("T-06"))["status"], "HOLD")
        self.assertIsNone(self.lab.data_review); self.assertFalse(self.lab.service_reviews); self.assertIsNone(self.lab.pilot_review)
        self.assertEqual(self.lab.source_reads, before)

    def test_stop_invalidates_permissions_drafts_every_review_disk_and_old_copy_remain(self):
        old = self.reviewed(); disk = (self.sources / SOURCE_NAMES["T-01"]).read_bytes()
        result = self.lab.change_state("training-coordinator", 1, False)
        self.assertEqual(result["status"], "TRAINING_STATE_CHANGED"); self.assertEqual(self.lab.version, 2)
        self.assertFalse(self.lab.permissions); self.assertFalse(self.lab.verified); self.assertIsNone(self.lab.data_review)
        self.assertFalse(self.lab.service_reviews); self.assertIsNone(self.lab.pilot_review)
        self.assertEqual(old["version"], 1); self.assertEqual(len(old["draftRecords"]), 3)
        self.assertEqual((self.sources / SOURCE_NAMES["T-01"]).read_bytes(), disk)
        before = self.lab.source_reads
        self.assertEqual(self.lab.execute(request(version=2))["status"], "HOLD")
        self.assertEqual(self.lab.export_proposal(2)["status"], "HOLD")
        self.assertEqual(self.lab.source_reads, before)

    def test_resume_requires_renewal_fresh_drafts_and_separate_review_chain(self):
        self.reviewed(); self.lab.change_state("training-coordinator", 1, False)
        self.assertEqual(self.lab.renew_permissions("training-coordinator", 2)["status"], "HOLD")
        self.lab.change_state("training-coordinator", 2, True)
        self.assertEqual(self.lab.version, 3); self.assertFalse(self.lab.permissions)
        before = self.lab.source_reads
        self.assertEqual(self.lab.execute(request(version=3))["status"], "HOLD")
        self.assertEqual(self.lab.source_reads, before)
        self.lab.renew_permissions("training-coordinator", 3)
        self.assertEqual(self.lab.review_data("training-data-owner", 3)["status"], "HOLD")
        self.assertEqual(self.lab.review_pilot("training-governance-reviewer", 3)["status"], "HOLD")
        self.assertEqual(self.reviewed()["status"], "LYRA_TRAINING_EXPORT")

    def test_wrong_state_role_stale_bool_version_and_nonbool_active_do_not_mutate(self):
        self.reviewed(); before = self.snapshot()
        for args in [(NETWORK_OWNER, 1, False), ("training-coordinator", True, False), ("training-coordinator", 0, False)] + [("training-coordinator", 1, bad) for bad in [0, 1, None, "false", []]]:
            self.assertEqual(self.lab.change_state(*args)["status"], "HOLD")
            self.assertEqual(self.snapshot(), before)

    def test_same_state_is_no_change_and_does_not_erase_review(self):
        self.reviewed(); before = self.snapshot()
        self.assertEqual(self.lab.change_state("training-coordinator", 1, True)["status"], "NO_CHANGE")
        self.assertEqual(self.snapshot(), before)

    def test_renewal_wrong_role_or_version_does_not_mutate(self):
        self.reviewed(); before = self.snapshot()
        for actor, version in [(NETWORK_OWNER, 1), ("training-data-owner", 1), ("training-coordinator", True), ("training-coordinator", 1.0), ("training-coordinator", 0)]:
            self.assertEqual(self.lab.renew_permissions(actor, version)["status"], "HOLD")
            self.assertEqual(self.snapshot(), before)

    def test_renewal_only_original_analytical_rights_no_owner_approval(self):
        self.lab.permissions["T-05"] = self.lab._permission("T-05")
        self.lab.permissions["T-01"]["operation"] = "restore"
        result = self.lab.renew_permissions("training-coordinator", 1)
        self.assertEqual(result["eligibleCaseIds"], ["T-01", "T-02", "T-03"])
        self.assertNotIn("T-05", self.lab.permissions)
        self.assertEqual(self.lab.permissions["T-01"]["operation"], "analyse")
        self.assertIsNone(self.lab.data_review); self.assertFalse(self.lab.service_reviews); self.assertIsNone(self.lab.pilot_review)
        self.assertEqual(self.lab.source_reads, 0)

    def test_actual_separate_process_resets_stop_and_has_no_reviews(self):
        self.reviewed(); self.lab.change_state("training-coordinator", 1, False)
        code = "import json; from lyra_lab import Lyra; x=Lyra(); print(json.dumps({'version':x.version,'active':x.active,'permissions':list(x.permissions),'drafts':x.verified,'data':x.data_review,'services':x.service_reviews,'pilot':x.pilot_review,'reads':x.source_reads}))"
        result = subprocess.run(["python", "-B", "-c", code], cwd=HERE, env=os.environ.copy(), capture_output=True, text=True)
        self.assertEqual(result.returncode, 0, result.stderr)
        self.assertEqual(json.loads(result.stdout), {"version": 1, "active": True, "permissions": ["T-01", "T-02", "T-03"], "drafts": {}, "data": None, "services": {}, "pilot": None, "reads": 0})

    def test_demo_separate_layers_no_real_release(self):
        output = demo()
        self.assertEqual(output["sourceReads"], 3)
        self.assertEqual(output["dataReview"]["status"], "ASSUMED_DATA_REVIEW_ONLY")
        self.assertEqual(output["networkReview"]["caseIds"], ["T-01", "T-03"])
        self.assertEqual(output["customerReview"]["caseIds"], ["T-02"])
        self.assertEqual(output["pilotReview"]["actualReleaseDecision"], "HOLD_FOR_REAL_PROVIDER_REVIEW")
        self.assertIs(output["export"]["realServiceReleaseApproved"], False)


if __name__ == "__main__":
    unittest.main()

{
  "carrierId": "Lyra-Regional-Training",
  "caseId": "T-01",
  "sourceId": "case-01",
  "revision": 1,
  "risk": "synthetic-internal",
  "original": "Invented Lyra incident L-01: service advisory received; root cause not established.",
  "aggregate": null,
  "note": "Invented analytical teaching source. No real service event."
}

{
  "carrierId": "Lyra-Regional-Training",
  "caseId": "T-02",
  "sourceId": "case-02",
  "revision": 1,
  "risk": "synthetic-public",
  "original": "Invented Lyra FAQ: service status is not established; use the proposed human support route.",
  "aggregate": null,
  "note": "Invented analytical teaching source. No real service event."
}

{
  "carrierId": "Lyra-Regional-Training",
  "caseId": "T-03",
  "sourceId": "case-03",
  "revision": 1,
  "risk": "synthetic-internal",
  "original": "Invented Lyra capacity sample; historical counts only.",
  "aggregate": {
    "totalIntervals": 200,
    "busyIntervals": 60,
    "busyPercent": 30
  },
  "note": "Invented analytical teaching source. No real service event."
}

{
  "carrierId": "Lyra-Regional-Training",
  "caseId": "T-04",
  "sourceId": "case-04",
  "revision": 1,
  "risk": "purpose-unresolved",
  "original": "Invented Lyra fraud-input descriptor; source purpose and permission unresolved.",
  "aggregate": null,
  "note": "Invented analytical teaching source. No real service event."
}

{
  "carrierId": "Lyra-Regional-Training",
  "caseId": "T-05",
  "sourceId": "case-05",
  "revision": 1,
  "risk": "operational-authority-excluded",
  "original": "Invented Lyra network-change descriptor; operational action excluded.",
  "aggregate": null,
  "note": "Invented analytical teaching source. No real service event."
}

{
  "carrierId": "Lyra-Regional-Training",
  "caseId": "T-06",
  "sourceId": "case-06",
  "revision": 1,
  "risk": "safety-review-unresolved",
  "original": "Invented Lyra emergency-service descriptor; actual safety and authority review absent.",
  "aggregate": null,
  "note": "Invented analytical teaching source. No real service event."
}