diff --git a/alignment/graph.py b/alignment/graph.py index e1e478d9..59e60f2c 100644 --- a/alignment/graph.py +++ b/alignment/graph.py @@ -68,15 +68,21 @@ def _parse_edge(payload: dict) -> AlignmentEdge: ) -def load_alignment(pack_id: str) -> AlignmentGraph: +def load_alignment( + pack_id: str, *, data_root: Path | None = None +) -> AlignmentGraph: """ - Load AlignmentEdge records from language_packs/data//alignment.jsonl. + Load AlignmentEdge records from //alignment.jsonl. + + ``data_root`` defaults to the committed ``language_packs/data`` tree; pass + an alternate root (e.g. a test-fixture copy) to read packs from elsewhere + without forking the parser. Returns an empty AlignmentGraph if the file does not exist. This is intentional: operational_base packs (en_minimal_v1) do not currently carry cross-language alignment edges. """ - alignment_path = _DATA_DIR / pack_id / "alignment.jsonl" + alignment_path = (data_root or _DATA_DIR) / pack_id / "alignment.jsonl" if not alignment_path.exists(): return AlignmentGraph([]) diff --git a/morphology/registry.py b/morphology/registry.py index 508eb18a..b25e9a8e 100644 --- a/morphology/registry.py +++ b/morphology/registry.py @@ -66,15 +66,21 @@ def _parse_entry(payload: dict) -> MorphologyEntry: ) -def load_morphology(pack_id: str) -> MorphologyRegistry: +def load_morphology( + pack_id: str, *, data_root: Path | None = None +) -> MorphologyRegistry: """ - Load MorphologyEntry records from language_packs/data//morphology.jsonl. + Load MorphologyEntry records from //morphology.jsonl. + + ``data_root`` defaults to the committed ``language_packs/data`` tree; pass + an alternate root (e.g. a test-fixture copy) to read packs from elsewhere + without forking the parser. Packs without morphology data return an empty registry; packs that set LexicalEntry.morphology_id are validated by the compiler against this registry. """ - morphology_path = _DATA_DIR / pack_id / "morphology.jsonl" + morphology_path = (data_root or _DATA_DIR) / pack_id / "morphology.jsonl" if not morphology_path.exists(): return MorphologyRegistry([]) diff --git a/tests/test_workbench_api.py b/tests/test_workbench_api.py index 324906d8..05a996a5 100644 --- a/tests/test_workbench_api.py +++ b/tests/test_workbench_api.py @@ -156,6 +156,52 @@ def test_eval_lane_detail_returns_read_only_flag() -> None: assert response.payload["data"]["read_only"] is True +def test_logos_pack_routes_return_real_read_only_models() -> None: + listing = _request("GET", "/logos/packs") + + assert listing.status == 200 + assert [item["pack_id"] for item in listing.payload["data"]["items"]] == [ + "grc_logos_cognition_v1", + "grc_logos_micro_v1", + "he_core_cognition_v1", + "he_logos_micro_v1", + ] + + overview = _request("GET", "/logos/packs/he_logos_micro_v1") + assert overview.status == 200 + assert overview.payload["data"]["pack_id"] == "he_logos_micro_v1" + assert overview.payload["data"]["holonomy_case_count"] == 0 + + contents = _request("GET", "/logos/packs/he_logos_micro_v1/contents") + assert contents.status == 200 + assert contents.payload["data"]["holonomy_cases"] == [] + assert contents.payload["data"]["lexicon"][0]["entry_id"] == "he-001" + + safety = _request("GET", "/logos/packs/he_logos_micro_v1/safety") + assert safety.status == 200 + assert safety.payload["data"]["missing_holonomy_refs"] == "unknown" + assert safety.payload["data"]["verdict"] == "unknown" + + alignment = _request("GET", "/logos/packs/he_logos_micro_v1/alignment") + assert alignment.status == 200 + assert alignment.payload["data"]["items"][0]["edge_id"] + assert alignment.payload["data"]["items"][0]["invalid_target"] is False + + +def test_logos_routes_fail_closed_for_unsafe_or_non_logos_ids() -> None: + unsafe = _request("GET", "/logos/packs/../../pyproject.toml") + assert unsafe.status == 404 + assert unsafe.payload["error"]["code"] == "not_found" + + non_logos = _request("GET", "/logos/packs/en_core_relations_v3") + assert non_logos.status == 404 + assert non_logos.payload["error"]["code"] == "not_found" + + mutation = _request("POST", "/logos/packs/he_logos_micro_v1") + assert mutation.status == 404 + assert mutation.payload["error"]["code"] == "not_found" + + def test_eval_run_rejects_unsafe_lane() -> None: response = _request( "POST", @@ -252,6 +298,11 @@ def test_full_w026_route_table_preserves_teaching_and_pack_bytes() -> None: ("GET", "/proposals", None), ("GET", "/evals", None), ("GET", "/evals/contemplation_quality", None), + ("GET", "/logos/packs", None), + ("GET", "/logos/packs/he_logos_micro_v1", None), + ("GET", "/logos/packs/he_logos_micro_v1/contents", None), + ("GET", "/logos/packs/he_logos_micro_v1/safety", None), + ("GET", "/logos/packs/he_logos_micro_v1/alignment", None), ( "POST", "/evals/run", diff --git a/tests/test_workbench_logos.py b/tests/test_workbench_logos.py new file mode 100644 index 00000000..4abc28ec --- /dev/null +++ b/tests/test_workbench_logos.py @@ -0,0 +1,231 @@ +from __future__ import annotations + +import ast +import hashlib +import json +import shutil +from pathlib import Path +from typing import Any, Callable + +from workbench import logos +from workbench.schemas import LogosMorphologyLinkIssue, SafetyVerdict, to_data + + +REPO_ROOT = Path(__file__).resolve().parent.parent +DATA_ROOT = REPO_ROOT / "language_packs" / "data" +LOGOS_PACK_IDS = [ + "grc_logos_cognition_v1", + "grc_logos_micro_v1", + "he_core_cognition_v1", + "he_logos_micro_v1", +] + + +def _copy_language_pack_root(tmp_path: Path) -> Path: + root = tmp_path / "language_packs_data" + shutil.copytree(DATA_ROOT, root) + return root + + +def _rewrite_first_jsonl_object( + path: Path, mutator: Callable[[dict[str, Any]], None] +) -> None: + lines = path.read_text(encoding="utf-8").splitlines() + for idx, line in enumerate(lines): + if not line.strip(): + continue + payload = json.loads(line) + mutator(payload) + lines[idx] = json.dumps(payload, ensure_ascii=False, sort_keys=True) + break + path.write_text("\n".join(lines) + "\n", encoding="utf-8") + + +def _refresh_lexicon_checksum(pack_dir: Path) -> None: + manifest_path = pack_dir / "manifest.json" + manifest = json.loads(manifest_path.read_text(encoding="utf-8")) + manifest["checksum"] = hashlib.sha256( + (pack_dir / "lexicon.jsonl").read_bytes() + ).hexdigest() + manifest_path.write_text( + json.dumps(manifest, ensure_ascii=False, indent=2, sort_keys=True) + "\n", + encoding="utf-8", + ) + + +def test_list_logos_packs_filters_to_the_readonly_logos_universe() -> None: + packs = logos.list_logos_packs() + + assert [pack.pack_id for pack in packs] == LOGOS_PACK_IDS + assert "en_core_relations_v3" not in {pack.pack_id for pack in packs} + assert all(pack.holonomy_case_count == 0 for pack in packs) + + +def test_overview_counts_and_honest_absent_holonomy() -> None: + overview = logos.logos_pack_overview("he_logos_micro_v1") + + assert overview.lexicon_count == 9 + assert overview.gloss_count == 9 + assert overview.morphology_count == 9 + assert overview.alignment_edge_count == 11 + assert overview.holonomy_case_count == 0 + assert overview.safety_status is SafetyVerdict.UNKNOWN + + contents = logos.logos_pack_contents("he_logos_micro_v1") + payload = to_data(contents) + assert payload["holonomy_cases"] == [] + assert "success" not in payload + assert "proof" not in payload + + +def test_contents_project_lexicon_gloss_morphology_and_alignment_rows() -> None: + contents = logos.logos_pack_contents("he_logos_micro_v1") + + davar = next(row for row in contents.lexicon if row.entry_id == "he-001") + assert davar.lemma == "דבר" + assert davar.morphology_id == "he-morph-001" + assert davar.epistemic_status == "speculative" + + davar_gloss = next(row for row in contents.glosses if row.lemma == "דבר") + assert davar_gloss.gloss == "word, matter, or spoken thing" + assert davar_gloss.entry_ids == ["he-001", "he-008"] + + devarim = next( + row for row in contents.morphology if row.morphology_id == "he-morph-008" + ) + assert devarim.prefix_chain == [] + assert devarim.stem == "דבר" + assert devarim.suffix_chain == ["ים"] + + edge = contents.alignment_edges[0] + expected_edge_id = hashlib.sha256( + b"he-001|grc-001|cross_lang.logos.utterance" + ).hexdigest()[:16] + assert edge.edge_id == expected_edge_id + assert edge.invalid_target is False + assert edge.target_pack_id == "grc_logos_micro_v1" + + +def test_safety_reports_unknown_holonomy_without_calling_it_clear() -> None: + report = logos.logos_pack_safety("he_logos_micro_v1") + + assert report.checksum_status is SafetyVerdict.CLEAR + assert report.domain_contract_status is SafetyVerdict.CLEAR + assert report.missing_holonomy_refs is SafetyVerdict.UNKNOWN + assert report.verdict is SafetyVerdict.UNKNOWN + assert report.verdict is not SafetyVerdict.CLEAR + assert report.dangling_morphology_links == [] + assert report.invalid_alignment_targets == [] + + +def test_dangling_morphology_link_is_listed_without_checksum_noise( + tmp_path: Path, +) -> None: + data_root = _copy_language_pack_root(tmp_path) + pack_dir = data_root / "he_logos_micro_v1" + _rewrite_first_jsonl_object( + pack_dir / "lexicon.jsonl", + lambda row: row.__setitem__("morphology_id", "missing-morph-001"), + ) + _refresh_lexicon_checksum(pack_dir) + + report = logos.logos_pack_safety("he_logos_micro_v1", data_root=data_root) + + assert report.checksum_status is SafetyVerdict.CLEAR + assert report.verdict is SafetyVerdict.WARNING + assert report.dangling_morphology_links == [ + LogosMorphologyLinkIssue( + entry_id="he-001", + morphology_id="missing-morph-001", + ) + ] + + +def test_checksum_mismatch_is_failed_and_not_clear(tmp_path: Path) -> None: + data_root = _copy_language_pack_root(tmp_path) + pack_dir = data_root / "he_logos_micro_v1" + _rewrite_first_jsonl_object( + pack_dir / "lexicon.jsonl", + lambda row: row.__setitem__("surface", "דבר-corrupted"), + ) + + report = logos.logos_pack_safety("he_logos_micro_v1", data_root=data_root) + + assert report.checksum_status is SafetyVerdict.FAILED + assert report.verdict is SafetyVerdict.FAILED + assert report.verdict is not SafetyVerdict.CLEAR + assert any( + "lexicon_checksum:mismatch" in error for error in report.checksum_errors + ) + + +def test_invalid_alignment_target_is_listed_without_touching_engine_validators( + tmp_path: Path, +) -> None: + data_root = _copy_language_pack_root(tmp_path) + pack_dir = data_root / "he_logos_micro_v1" + _rewrite_first_jsonl_object( + pack_dir / "alignment.jsonl", + lambda row: row.__setitem__("target_id", "ghost-target"), + ) + + report = logos.logos_pack_safety("he_logos_micro_v1", data_root=data_root) + + assert report.checksum_status is SafetyVerdict.CLEAR + assert report.verdict is SafetyVerdict.WARNING + assert [ + (issue.source_id, issue.target_id, issue.relation) + for issue in report.invalid_alignment_targets + ] == [("he-001", "ghost-target", "cross_lang.logos.utterance")] + + +def test_collapse_anchor_target_validity_tracks_declared_lexicon_entries() -> None: + """A declared collapse anchor resolves; an undeclared one is dangling. + + Regression for the removed relation/prefix carve-out: ``en-collapse-*`` + targets are valid iff they are real declared lexicon entries (e.g. + ``en-collapse-love`` in ``en_collapse_anchors_v1``). An ``en-collapse-*`` + target declared nowhere must surface as an invalid alignment target, not be + silently passed by the safety reader. + """ + + rows = logos.logos_pack_alignment("grc_logos_cognition_v1") + by_target = { + row.target_id: row + for row in rows + if row.relation == "cross_lang.no_english_collapse" + } + + declared = by_target["en-collapse-love"] + assert declared.invalid_target is False + assert declared.target_resolved is True + assert declared.target_pack_id == "en_collapse_anchors_v1" + + undeclared = by_target["en-collapse-breath"] + assert undeclared.invalid_target is True + assert undeclared.target_resolved is False + assert undeclared.target_pack_id is None + + report = logos.logos_pack_safety("grc_logos_cognition_v1") + invalid_target_ids = { + issue.target_id for issue in report.invalid_alignment_targets + } + assert "en-collapse-breath" in invalid_target_ids + assert "en-collapse-love" not in invalid_target_ids + assert report.verdict is SafetyVerdict.WARNING + assert report.verdict is not SafetyVerdict.CLEAR + + +def test_no_algebra_imports_in_workbench_logos_module() -> None: + tree = ast.parse((REPO_ROOT / "workbench" / "logos.py").read_text(encoding="utf-8")) + + imported: list[str] = [] + for node in ast.walk(tree): + if isinstance(node, ast.Import): + imported.extend(alias.name for alias in node.names) + elif isinstance(node, ast.ImportFrom) and node.module: + imported.append(node.module) + + assert not any( + name == "algebra" or name.startswith("algebra.") for name in imported + ) diff --git a/tests/test_workbench_schemas.py b/tests/test_workbench_schemas.py index 23ee7ff6..c46fb00d 100644 --- a/tests/test_workbench_schemas.py +++ b/tests/test_workbench_schemas.py @@ -1,6 +1,6 @@ from __future__ import annotations -from workbench.schemas import RuntimeStatus, error, ok +from workbench.schemas import RuntimeStatus, SafetyVerdict, error, ok def test_ok_envelope_contains_generated_at_and_data() -> None: @@ -25,3 +25,16 @@ def test_error_envelope_contains_generated_at_and_code() -> None: assert payload["ok"] is False assert isinstance(payload["generated_at"], str) assert payload["error"] == {"code": "not_found", "message": "missing"} + + +def test_safety_verdict_values_and_json_projection() -> None: + assert {item.value for item in SafetyVerdict} == { + "clear", + "warning", + "failed", + "unknown", + } + + payload = ok({"verdict": SafetyVerdict.WARNING}) + + assert payload["data"] == {"verdict": "warning"} diff --git a/workbench-ui/schema-snapshot.json b/workbench-ui/schema-snapshot.json index edb71401..bffc27b4 100644 --- a/workbench-ui/schema-snapshot.json +++ b/workbench-ui/schema-snapshot.json @@ -237,6 +237,120 @@ "source_digest", "calibration_evidence_ref" ], + "LogosAlignmentRow": [ + "edge_id", + "source_id", + "target_id", + "relation", + "weight", + "evidence_ids", + "target_pack_id", + "target_resolved", + "invalid_target" + ], + "LogosAlignmentTargetIssue": [ + "edge_id", + "source_id", + "target_id", + "relation", + "target_pack_id" + ], + "LogosGlossRow": [ + "gloss_id", + "lemma", + "gloss", + "pos", + "entry_ids", + "provenance_ids", + "epistemic_status", + "raw" + ], + "LogosLexiconRow": [ + "entry_id", + "surface", + "lemma", + "language", + "part_of_speech", + "pos", + "morphology_id", + "morphology_tags", + "semantic_domains", + "provenance_ids", + "epistemic_status" + ], + "LogosMorphologyLinkIssue": [ + "entry_id", + "morphology_id" + ], + "LogosMorphologyRow": [ + "morphology_id", + "surface", + "lemma", + "language", + "root", + "prefix_chain", + "stem", + "inflection", + "suffix_chain" + ], + "LogosPackContents": [ + "schema_version", + "pack_id", + "manifest", + "lexicon", + "glosses", + "morphology", + "frames", + "compositions", + "alignment_edges", + "holonomy_cases" + ], + "LogosPackOverview": [ + "schema_version", + "normalization_policy", + "source_manifest", + "known_gaps" + ], + "LogosPackSummary": [ + "pack_id", + "language", + "role", + "script", + "version", + "determinism_class", + "gate_engaged", + "oov_policy", + "lexicon_count", + "gloss_count", + "morphology_count", + "frame_count", + "composition_count", + "alignment_edge_count", + "holonomy_case_count", + "safety_status", + "manifest_digest", + "manifest_path" + ], + "LogosSafetyReport": [ + "schema_version", + "pack_id", + "checksum_status", + "checksum_errors", + "domain_contract", + "domain_contract_status", + "oov_policy_ok", + "gate_policy_ok", + "path_safety_ok", + "dangling_morphology_links", + "invalid_alignment_targets", + "missing_holonomy_refs", + "epistemic_status_counts", + "speculative_entries", + "contested_entries", + "falsified_entries", + "known_gaps", + "verdict" + ], "MathProposalDetail": [ "wrong_zero_assertion", "proposed_change_payload", diff --git a/workbench-ui/src/design/doctrine/schemaDrift.test.ts b/workbench-ui/src/design/doctrine/schemaDrift.test.ts index f9eadb74..520d3a2d 100644 --- a/workbench-ui/src/design/doctrine/schemaDrift.test.ts +++ b/workbench-ui/src/design/doctrine/schemaDrift.test.ts @@ -19,10 +19,22 @@ import { describe, expect, it } from "vitest"; * only shrink. */ -// Empty: every engine schema is now mirrored. New unmirrored schemas may be -// allowlisted here as shrink-only debt (a class gaining a mirror while listed -// fails the gate). -const NOT_YET_MIRRORED = new Set([]); +// Shrink-only debt: backend schemas shipped ahead of their TS mirror. A class +// gaining a TS interface while still listed here fails the gate. +// CORE-Logos read models (LG-1): backend readers shipped without the /logos +// frontend route; LG-2 adds the TS interfaces and removes each entry here. +const NOT_YET_MIRRORED = new Set([ + "LogosPackSummary", + "LogosPackOverview", + "LogosPackContents", + "LogosLexiconRow", + "LogosGlossRow", + "LogosMorphologyRow", + "LogosAlignmentRow", + "LogosMorphologyLinkIssue", + "LogosAlignmentTargetIssue", + "LogosSafetyReport", +]); const UI_ROOT = join(__dirname, "..", "..", ".."); const snapshot: Record = JSON.parse( diff --git a/workbench/api.py b/workbench/api.py index a99f1ff9..70f5b505 100644 --- a/workbench/api.py +++ b/workbench/api.py @@ -18,7 +18,7 @@ from core.epistemic_state import ( epistemic_state_for_grounding_source, normative_detail_from_verdicts, ) -from workbench import calibration, readers +from workbench import calibration, logos, readers from workbench.journal import DEFAULT_JOURNAL_DIR, TurnJournal, TurnJournalEntry from workbench.evidence_bundle import build_evidence_bundle from workbench.tour import determinism_tour @@ -192,6 +192,10 @@ class WorkbenchApi: if method == "GET" and path.startswith("/math-proposals/"): proposal_id = unquote(path.removeprefix("/math-proposals/")) return ApiResponse(200, ok(readers.read_math_proposal(proposal_id))) + if method == "GET" and path == "/logos/packs": + return ApiResponse(200, ok({"items": logos.list_logos_packs()})) + if method == "GET" and path.startswith("/logos/packs/"): + return self._logos_read(path) if method == "GET" and path == "/packs": limit, offset = _pagination(query) return ApiResponse( @@ -431,6 +435,44 @@ class WorkbenchApi: return ApiResponse(200, ok(comparison)) return ApiResponse(404, error("not_found", f"route not found: {method} {path}")) + def _logos_read(self, path: str) -> ApiResponse: + tail = unquote(path.removeprefix("/logos/packs/")) + endpoint = "overview" + pack_id = tail + for suffix, name in ( + ("/contents", "contents"), + ("/safety", "safety"), + ("/alignment", "alignment"), + ): + if tail.endswith(suffix): + endpoint = name + pack_id = tail.removesuffix(suffix) + break + if not pack_id or "/" in pack_id: + return ApiResponse( + 404, error("not_found", f"logos pack not found: {pack_id or tail}") + ) + try: + if endpoint == "contents": + return ApiResponse(200, ok(logos.logos_pack_contents(pack_id))) + if endpoint == "safety": + return ApiResponse(200, ok(logos.logos_pack_safety(pack_id))) + if endpoint == "alignment": + return ApiResponse( + 200, + ok( + { + "pack_id": pack_id, + "items": logos.logos_pack_alignment(pack_id), + } + ), + ) + return ApiResponse(200, ok(logos.logos_pack_overview(pack_id))) + except (FileNotFoundError, ValueError): + return ApiResponse( + 404, error("not_found", f"logos pack not found: {pack_id}") + ) + def _math_ratify(self, proposal_id: str, body: bytes) -> ApiResponse: """Route ratification by change_kind; in-process execution with allowlist checks.""" category: str | None = None diff --git a/workbench/logos.py b/workbench/logos.py new file mode 100644 index 00000000..e79a9a45 --- /dev/null +++ b/workbench/logos.py @@ -0,0 +1,520 @@ +"""Read-only CORE-Logos Workbench projections. + +The engine owns pack truth. This module projects committed language-pack +artifacts into stable Workbench read models; it never mutates packs, compiles +new state, or repairs broken links. +""" + +from __future__ import annotations + +import hashlib +import json +import re +from pathlib import Path +from typing import Any + +from language_packs.schema import AlignmentEdge, LexicalEntry, MorphologyEntry +from workbench.schemas import ( + LogosAlignmentRow, + LogosAlignmentTargetIssue, + LogosGlossRow, + LogosLexiconRow, + LogosMorphologyLinkIssue, + LogosMorphologyRow, + LogosPackContents, + LogosPackOverview, + LogosPackSummary, + LogosSafetyReport, + SafetyVerdict, +) + +REPO_ROOT = Path(__file__).resolve().parents[1] +LANGUAGE_PACK_ROOT = REPO_ROOT / "language_packs" / "data" +READ_CHUNK_BYTES = 64 * 1024 +SAFE_LOGOS_PACK_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_-]{0,127}$") + +_LOGOS_DEPTH_ROLES = frozenset({"depth_root", "depth_relation"}) +_LOGOS_DOMAIN_ID = "hebrew_greek_textual_reasoning" +_OOV_POLICIES = frozenset( + {"fail_closed", "tagged_fallback", "propose_vocab_expansion"} +) + + +def list_logos_packs( + *, data_root: Path | None = None +) -> list[LogosPackSummary]: + """Return the deterministic CORE-Logos pack universe.""" + + root = _data_root(data_root) + summaries: list[LogosPackSummary] = [] + for manifest_path in sorted(root.glob("*/manifest.json")): + manifest = _read_json_object(manifest_path) + pack_dir = manifest_path.parent + if not _is_logos_pack(manifest, pack_dir): + continue + overview = logos_pack_overview(pack_dir.name, data_root=root) + summaries.append( + LogosPackSummary( + pack_id=overview.pack_id, + language=overview.language, + role=overview.role, + script=overview.script, + version=overview.version, + determinism_class=overview.determinism_class, + gate_engaged=overview.gate_engaged, + oov_policy=overview.oov_policy, + lexicon_count=overview.lexicon_count, + gloss_count=overview.gloss_count, + morphology_count=overview.morphology_count, + frame_count=overview.frame_count, + composition_count=overview.composition_count, + alignment_edge_count=overview.alignment_edge_count, + holonomy_case_count=overview.holonomy_case_count, + safety_status=overview.safety_status, + manifest_digest=overview.manifest_digest, + manifest_path=overview.manifest_path, + ) + ) + return sorted(summaries, key=lambda item: item.pack_id) + + +def logos_pack_overview( + pack_id: str, *, data_root: Path | None = None +) -> LogosPackOverview: + root = _data_root(data_root) + pack_dir = _require_logos_pack_dir(pack_id, root) + manifest_path = pack_dir / "manifest.json" + manifest = _read_json_object(manifest_path) + lexicon = _load_lexicon(pack_dir) + morphology = _load_morphology(pack_id, root) + alignment = _load_alignment(pack_id, root) + safety = logos_pack_safety(pack_id, data_root=root) + return LogosPackOverview( + pack_id=str(manifest.get("pack_id") or pack_id), + language=_optional_str(manifest.get("language")), + role=_optional_str(manifest.get("role")), + script=_optional_str(manifest.get("script")), + version=_optional_str(manifest.get("version")), + determinism_class=_optional_str(manifest.get("determinism_class")), + gate_engaged=bool(manifest.get("gate_engaged", False)), + oov_policy=_optional_str(manifest.get("oov_policy")), + lexicon_count=len(lexicon), + gloss_count=len(_read_jsonl_objects(pack_dir / "glosses.jsonl")), + morphology_count=len(morphology), + frame_count=len(_read_jsonl_objects(pack_dir / "frames.jsonl")), + composition_count=len(_read_jsonl_objects(pack_dir / "compositions.jsonl")), + alignment_edge_count=len(alignment), + holonomy_case_count=0, + safety_status=safety.verdict, + manifest_digest=_sha256_file(manifest_path), + manifest_path=_display_path(manifest_path), + normalization_policy=_optional_str(manifest.get("normalization_policy")), + source_manifest=_optional_str(manifest.get("source_manifest")), + known_gaps=_string_list(manifest.get("known_gaps")), + ) + + +def logos_pack_contents( + pack_id: str, *, data_root: Path | None = None +) -> LogosPackContents: + root = _data_root(data_root) + pack_dir = _require_logos_pack_dir(pack_id, root) + manifest = _read_json_object(pack_dir / "manifest.json") + lexicon = _load_lexicon(pack_dir) + lexicon_rows = [_lexicon_row(entry) for entry in lexicon] + entry_ids_by_lemma = _entry_ids_by_lemma(lexicon_rows) + return LogosPackContents( + schema_version="logos_pack_contents_v1", + pack_id=str(manifest.get("pack_id") or pack_id), + manifest=manifest, + lexicon=lexicon_rows, + glosses=_gloss_rows(pack_dir, entry_ids_by_lemma), + morphology=[ + _morphology_row(entry) + for entry in _load_morphology(pack_id, root) + ], + frames=_read_jsonl_objects(pack_dir / "frames.jsonl"), + compositions=_read_jsonl_objects(pack_dir / "compositions.jsonl"), + alignment_edges=logos_pack_alignment(pack_id, data_root=root), + holonomy_cases=[], + ) + + +def logos_pack_alignment( + pack_id: str, *, data_root: Path | None = None +) -> list[LogosAlignmentRow]: + root = _data_root(data_root) + _require_logos_pack_dir(pack_id, root) # validate id / logos membership + target_index = _entry_id_index(root) + return [ + _alignment_row(edge, target_index) + for edge in _load_alignment(pack_id, root) + ] + + +def logos_pack_safety( + pack_id: str, *, data_root: Path | None = None +) -> LogosSafetyReport: + root = _data_root(data_root) + pack_dir = _require_logos_pack_dir(pack_id, root) + manifest = _read_json_object(pack_dir / "manifest.json") + lexicon = _load_lexicon(pack_dir) + morphology = _load_morphology(pack_id, root) + alignment_rows = logos_pack_alignment(pack_id, data_root=root) + + checksum_status, checksum_errors = _checksum_status(pack_id, pack_dir, root) + domain_contract = _domain_contract_status(pack_id, root) + domain_status = ( + SafetyVerdict.CLEAR + if bool(domain_contract.get("valid", False)) + else SafetyVerdict.FAILED + ) + morphology_ids = {entry.morphology_id for entry in morphology} + dangling = [ + LogosMorphologyLinkIssue( + entry_id=entry.entry_id, + morphology_id=entry.morphology_id or "", + ) + for entry in lexicon + if entry.morphology_id and entry.morphology_id not in morphology_ids + ] + invalid_targets = [ + LogosAlignmentTargetIssue( + edge_id=row.edge_id, + source_id=row.source_id, + target_id=row.target_id, + relation=row.relation, + target_pack_id=row.target_pack_id, + ) + for row in alignment_rows + if row.invalid_target + ] + epistemic_counts: dict[str, int] = {} + speculative: list[str] = [] + contested: list[str] = [] + falsified: list[str] = [] + for entry in lexicon: + status = entry.epistemic_status or "speculative" + epistemic_counts[status] = epistemic_counts.get(status, 0) + 1 + if status == "speculative": + speculative.append(entry.entry_id) + elif status == "contested": + contested.append(entry.entry_id) + elif status == "falsified": + falsified.append(entry.entry_id) + + oov_policy = _optional_str(manifest.get("oov_policy")) + role = _optional_str(manifest.get("role")) + oov_policy_ok = oov_policy in _OOV_POLICIES + gate_policy_ok = not ( + role in _LOGOS_DEPTH_ROLES + and bool(manifest.get("gate_engaged", False)) + and oov_policy != "fail_closed" + ) + warning_present = bool(dangling or invalid_targets or contested or falsified) + known_gaps = _string_list(manifest.get("known_gaps")) + if checksum_status is SafetyVerdict.FAILED or domain_status is SafetyVerdict.FAILED: + verdict = SafetyVerdict.FAILED + elif not (oov_policy_ok and gate_policy_ok): + verdict = SafetyVerdict.FAILED + elif warning_present or known_gaps: + verdict = SafetyVerdict.WARNING + else: + verdict = SafetyVerdict.UNKNOWN + + return LogosSafetyReport( + schema_version="logos_safety_report_v1", + pack_id=str(manifest.get("pack_id") or pack_id), + checksum_status=checksum_status, + checksum_errors=checksum_errors, + domain_contract=domain_contract, + domain_contract_status=domain_status, + oov_policy_ok=oov_policy_ok, + gate_policy_ok=gate_policy_ok, + path_safety_ok=True, + dangling_morphology_links=dangling, + invalid_alignment_targets=invalid_targets, + missing_holonomy_refs=SafetyVerdict.UNKNOWN, + epistemic_status_counts=dict(sorted(epistemic_counts.items())), + speculative_entries=sorted(speculative), + contested_entries=sorted(contested), + falsified_entries=sorted(falsified), + known_gaps=known_gaps, + verdict=verdict, + ) + + +def _data_root(data_root: Path | None) -> Path: + return (data_root or LANGUAGE_PACK_ROOT).resolve() + + +def _validate_pack_id(pack_id: str) -> str: + if not SAFE_LOGOS_PACK_ID_RE.fullmatch(pack_id) or ".." in pack_id: + raise ValueError("pack id contains unsafe characters") + return pack_id + + +def _require_logos_pack_dir(pack_id: str, root: Path) -> Path: + safe_id = _validate_pack_id(pack_id) + candidate = (root / safe_id).resolve() + if candidate != root and root not in candidate.parents: + raise ValueError("pack id resolves outside language pack root") + manifest_path = candidate / "manifest.json" + if not manifest_path.is_file(): + raise FileNotFoundError(pack_id) + manifest = _read_json_object(manifest_path) + if not _is_logos_pack(manifest, candidate): + raise FileNotFoundError(pack_id) + return candidate + + +def _is_logos_pack(manifest: dict[str, Any], pack_dir: Path) -> bool: + if (pack_dir / "alignment.jsonl").is_file(): + return True + role = _optional_str(manifest.get("role")) + if role not in _LOGOS_DEPTH_ROLES: + return False + pack_id = str(manifest.get("pack_id") or pack_dir.name) + # The committed tree also contains a generic English relation seed with + # role=depth_relation. Role-only admission is constrained to the Logos + # substrate so the Studio universe stays the four-pack wave contract. + return ( + manifest.get("domain_id") == _LOGOS_DOMAIN_ID + or "logos" in pack_id + or pack_id.startswith(("he_", "grc_")) + ) + + +def _optional_str(value: Any) -> str | None: + return None if value is None else str(value) + + +def _string_list(value: Any) -> list[str]: + if not isinstance(value, list): + return [] + return [item for item in (str(v) for v in value) if item] + + +def _sha256_file(path: Path) -> str: + hasher = hashlib.sha256() + with path.open("rb") as fh: + for chunk in iter(lambda: fh.read(READ_CHUNK_BYTES), b""): + hasher.update(chunk) + return "sha256:" + hasher.hexdigest() + + +def _sha256_hex(path: Path) -> str: + hasher = hashlib.sha256() + with path.open("rb") as fh: + for chunk in iter(lambda: fh.read(READ_CHUNK_BYTES), b""): + hasher.update(chunk) + return hasher.hexdigest() + + +def _display_path(path: Path) -> str: + try: + return path.resolve().relative_to(REPO_ROOT.resolve()).as_posix() + except ValueError: + return path.resolve().as_posix() + + +def _read_json_object(path: Path) -> dict[str, Any]: + payload = json.loads(path.read_text(encoding="utf-8")) + if not isinstance(payload, dict): + raise ValueError(f"expected JSON object in {_display_path(path)}") + return payload + + +def _read_jsonl_records(path: Path) -> list[tuple[int, dict[str, Any]]]: + if not path.exists(): + return [] + records: list[tuple[int, dict[str, Any]]] = [] + for line_no, raw_line in enumerate( + path.read_text(encoding="utf-8").splitlines(), start=1 + ): + line = raw_line.strip() + if not line: + continue + payload = json.loads(line) + if not isinstance(payload, dict): + raise ValueError(f"expected JSON object at {_display_path(path)}:{line_no}") + records.append((line_no, payload)) + return records + + +def _read_jsonl_objects(path: Path) -> list[dict[str, Any]]: + return [payload for _, payload in _read_jsonl_records(path)] + + +def _load_lexicon(pack_dir: Path) -> list[LexicalEntry]: + from language_packs.compiler import _parse_entry + + return [ + _parse_entry(payload) + for _, payload in _read_jsonl_records(pack_dir / "lexicon.jsonl") + ] + + +def _load_morphology(pack_id: str, root: Path) -> tuple[MorphologyEntry, ...]: + from morphology.registry import load_morphology + + return tuple(load_morphology(pack_id, data_root=root).entries) + + +def _load_alignment(pack_id: str, root: Path) -> tuple[AlignmentEdge, ...]: + from alignment.graph import load_alignment + + return tuple(load_alignment(pack_id, data_root=root).edges) + + +def _lexicon_row(entry: LexicalEntry) -> LogosLexiconRow: + return LogosLexiconRow( + entry_id=entry.entry_id, + surface=entry.surface, + lemma=entry.lemma, + language=entry.language, + part_of_speech=entry.part_of_speech, + pos=entry.pos, + morphology_id=entry.morphology_id, + morphology_tags=list(entry.morphology_tags), + semantic_domains=list(entry.semantic_domains), + provenance_ids=list(entry.provenance_ids), + epistemic_status=entry.epistemic_status, + ) + + +def _entry_ids_by_lemma(rows: list[LogosLexiconRow]) -> dict[str, list[str]]: + out: dict[str, list[str]] = {} + for row in rows: + out.setdefault(row.lemma, []).append(row.entry_id) + return {lemma: sorted(ids) for lemma, ids in sorted(out.items())} + + +def _gloss_rows( + pack_dir: Path, entry_ids_by_lemma: dict[str, list[str]] +) -> list[LogosGlossRow]: + rows: list[LogosGlossRow] = [] + for line_no, payload in _read_jsonl_records(pack_dir / "glosses.jsonl"): + lemma = str(payload.get("lemma") or "") + gloss = str(payload.get("gloss") or "") + pos = _optional_str(payload.get("pos")) + gloss_id = hashlib.sha256( + f"{pack_dir.name}|{line_no}|{lemma}|{pos or ''}|{gloss}".encode("utf-8") + ).hexdigest()[:16] + rows.append( + LogosGlossRow( + gloss_id=gloss_id, + lemma=lemma, + gloss=gloss, + pos=pos, + entry_ids=entry_ids_by_lemma.get(lemma, []), + provenance_ids=_string_list(payload.get("provenance_ids")), + epistemic_status=_optional_str( + payload.get("epistemic_status", payload.get("status")) + ), + raw=dict(payload), + ) + ) + return rows + + +def _morphology_row(entry: MorphologyEntry) -> LogosMorphologyRow: + return LogosMorphologyRow( + morphology_id=entry.morphology_id, + surface=entry.surface, + lemma=entry.lemma, + language=entry.language, + root=entry.root, + prefix_chain=list(entry.prefix_chain), + stem=entry.stem, + inflection={str(k): str(v) for k, v in entry.inflection.items()}, + suffix_chain=list(entry.suffix_chain), + ) + + +def _edge_id(edge: AlignmentEdge) -> str: + return hashlib.sha256( + f"{edge.source_id}|{edge.target_id}|{edge.relation}".encode("utf-8") + ).hexdigest()[:16] + + +def _entry_id_index(root: Path) -> dict[str, str]: + index: dict[str, str] = {} + for lexicon_path in sorted(root.glob("*/lexicon.jsonl")): + for _, payload in _read_jsonl_records(lexicon_path): + entry_id = payload.get("entry_id") + if isinstance(entry_id, str) and entry_id: + index[entry_id] = lexicon_path.parent.name + return index + + +def _alignment_row( + edge: AlignmentEdge, target_index: dict[str, str] +) -> LogosAlignmentRow: + target_pack_id = target_index.get(edge.target_id) + target_resolved = target_pack_id is not None + # A target is valid iff it resolves to a real declared lexicon entry in some + # pack (collapse anchors like ``en-collapse-love`` are declared entries in + # ``en_collapse_anchors_v1``). An ``en-collapse-*`` target that is declared + # nowhere is a genuine dangling reference and is reported as such — no + # relation/prefix carve-out, so the safety reader cannot silently pass an + # undeclared anchor. + return LogosAlignmentRow( + edge_id=_edge_id(edge), + source_id=edge.source_id, + target_id=edge.target_id, + relation=edge.relation, + weight=edge.weight, + evidence_ids=list(edge.evidence_ids), + target_pack_id=target_pack_id, + target_resolved=target_resolved, + invalid_target=not target_resolved, + ) + + +def _checksum_status( + pack_id: str, pack_dir: Path, root: Path +) -> tuple[SafetyVerdict, list[str]]: + manifest = _read_json_object(pack_dir / "manifest.json") + errors: list[str] = [] + lexicon_path = pack_dir / "lexicon.jsonl" + declared = manifest.get("checksum") + if not isinstance(declared, str) or not declared: + errors.append("manifest.checksum:missing") + elif not lexicon_path.is_file(): + errors.append("lexicon.jsonl:missing") + else: + actual = _sha256_hex(lexicon_path) + if actual != declared: + errors.append(f"lexicon_checksum:mismatch:{actual}!={declared}") + + for filename, key in ( + ("glosses.jsonl", "glosses_checksum"), + ("frames.jsonl", "frame_checksum"), + ("compositions.jsonl", "composition_checksum"), + ): + expected = manifest.get(key) + if expected is None: + continue + path = pack_dir / filename + if not path.is_file(): + errors.append(f"{filename}:missing_for:{key}") + continue + actual = _sha256_hex(path) + if actual != expected: + errors.append(f"{key}:mismatch:{actual}!={expected}") + + if not errors and root == LANGUAGE_PACK_ROOT.resolve(): + try: + from language_packs.compiler import load_pack + + load_pack(pack_id) + except ValueError as exc: + errors.append(f"compiler:{exc}") + return (SafetyVerdict.FAILED, errors) if errors else (SafetyVerdict.CLEAR, []) + + +def _domain_contract_status(pack_id: str, root: Path) -> dict[str, Any]: + from language_packs.domain_contract import validate_domain_contract_pack + + return validate_domain_contract_pack(pack_id, data_root=root).as_dict() diff --git a/workbench/schemas.py b/workbench/schemas.py index 22f09da7..d00f7e81 100644 --- a/workbench/schemas.py +++ b/workbench/schemas.py @@ -4,6 +4,7 @@ from __future__ import annotations from dataclasses import asdict, dataclass, field from datetime import datetime, timezone +from enum import Enum from typing import Any, Literal @@ -60,10 +61,12 @@ def utc_now() -> str: def to_data(value: Any) -> Any: + if isinstance(value, Enum): + return value.value if hasattr(value, "as_dict") and callable(value.as_dict): return value.as_dict() if hasattr(value, "__dataclass_fields__"): - return asdict(value) + return to_data(asdict(value)) if isinstance(value, dict): return {str(k): to_data(v) for k, v in value.items()} if isinstance(value, (list, tuple)): @@ -615,6 +618,147 @@ class PackDetail(PackSummary): manifest: dict[str, Any] = field(default_factory=dict) +class SafetyVerdict(str, Enum): + CLEAR = "clear" + WARNING = "warning" + FAILED = "failed" + UNKNOWN = "unknown" + + +@dataclass(frozen=True, slots=True) +class LogosPackSummary: + pack_id: str + language: str | None + role: str | None + script: str | None + version: str | None + determinism_class: str | None + gate_engaged: bool + oov_policy: str | None + lexicon_count: int + gloss_count: int + morphology_count: int + frame_count: int + composition_count: int + alignment_edge_count: int + holonomy_case_count: int + safety_status: SafetyVerdict + manifest_digest: str + manifest_path: str + + +@dataclass(frozen=True, slots=True) +class LogosPackOverview(LogosPackSummary): + schema_version: Literal["logos_pack_overview_v1"] = "logos_pack_overview_v1" + normalization_policy: str | None = None + source_manifest: str | None = None + known_gaps: list[str] = field(default_factory=list) + + +@dataclass(frozen=True, slots=True) +class LogosLexiconRow: + entry_id: str + surface: str + lemma: str + language: str + part_of_speech: str | None + pos: str | None + morphology_id: str | None + morphology_tags: list[str] + semantic_domains: list[str] + provenance_ids: list[str] + epistemic_status: str + + +@dataclass(frozen=True, slots=True) +class LogosGlossRow: + gloss_id: str + lemma: str + gloss: str + pos: str | None + entry_ids: list[str] + provenance_ids: list[str] + epistemic_status: str | None + raw: dict[str, Any] = field(default_factory=dict) + + +@dataclass(frozen=True, slots=True) +class LogosMorphologyRow: + morphology_id: str + surface: str + lemma: str + language: str + root: str | None + prefix_chain: list[str] + stem: str | None + inflection: dict[str, str] + suffix_chain: list[str] + + +@dataclass(frozen=True, slots=True) +class LogosAlignmentRow: + edge_id: str + source_id: str + target_id: str + relation: str + weight: float + evidence_ids: list[str] + target_pack_id: str | None + target_resolved: bool + invalid_target: bool + + +@dataclass(frozen=True, slots=True) +class LogosPackContents: + schema_version: Literal["logos_pack_contents_v1"] + pack_id: str + manifest: dict[str, Any] + lexicon: list[LogosLexiconRow] = field(default_factory=list) + glosses: list[LogosGlossRow] = field(default_factory=list) + morphology: list[LogosMorphologyRow] = field(default_factory=list) + frames: list[dict[str, Any]] = field(default_factory=list) + compositions: list[dict[str, Any]] = field(default_factory=list) + alignment_edges: list[LogosAlignmentRow] = field(default_factory=list) + holonomy_cases: list[dict[str, Any]] = field(default_factory=list) + + +@dataclass(frozen=True, slots=True) +class LogosMorphologyLinkIssue: + entry_id: str + morphology_id: str + + +@dataclass(frozen=True, slots=True) +class LogosAlignmentTargetIssue: + edge_id: str + source_id: str + target_id: str + relation: str + target_pack_id: str | None + + +@dataclass(frozen=True, slots=True) +class LogosSafetyReport: + schema_version: Literal["logos_safety_report_v1"] + pack_id: str + checksum_status: SafetyVerdict + checksum_errors: list[str] + domain_contract: dict[str, Any] + domain_contract_status: SafetyVerdict + oov_policy_ok: bool + gate_policy_ok: bool + path_safety_ok: bool + dangling_morphology_links: list[LogosMorphologyLinkIssue] + invalid_alignment_targets: list[LogosAlignmentTargetIssue] + missing_holonomy_refs: SafetyVerdict + epistemic_status_counts: dict[str, int] + speculative_entries: list[str] + contested_entries: list[str] + falsified_entries: list[str] + known_gaps: list[str] + verdict: SafetyVerdict + + AuditSource = Literal[ "engine_state_manifest", "math_proposal_log",