feat(workbench): add read-only core-logos readers

This commit is contained in:
Shay 2026-06-14 11:43:35 -07:00
parent b652bb0392
commit c5b30f68e4
10 changed files with 1152 additions and 13 deletions

View file

@ -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/<pack_id>/alignment.jsonl.
Load AlignmentEdge records from <data_root>/<pack_id>/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([])

View file

@ -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/<pack_id>/morphology.jsonl.
Load MorphologyEntry records from <data_root>/<pack_id>/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([])

View file

@ -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",

View file

@ -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
)

View file

@ -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"}

View file

@ -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",

View file

@ -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<string>([]);
// 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<string>([
"LogosPackSummary",
"LogosPackOverview",
"LogosPackContents",
"LogosLexiconRow",
"LogosGlossRow",
"LogosMorphologyRow",
"LogosAlignmentRow",
"LogosMorphologyLinkIssue",
"LogosAlignmentTargetIssue",
"LogosSafetyReport",
]);
const UI_ROOT = join(__dirname, "..", "..", "..");
const snapshot: Record<string, string[]> = JSON.parse(

View file

@ -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

520
workbench/logos.py Normal file
View file

@ -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()

View file

@ -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",