Merge pull request #737 from AssetOverflow/codex/lg-1-core-logos-readers

feat(workbench): add read-only core-logos readers
This commit is contained in:
Shay 2026-06-14 11:46:27 -07:00 committed by GitHub
commit bfd665bf08
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
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. Returns an empty AlignmentGraph if the file does not exist.
This is intentional: operational_base packs (en_minimal_v1) do not This is intentional: operational_base packs (en_minimal_v1) do not
currently carry cross-language alignment edges. 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(): if not alignment_path.exists():
return AlignmentGraph([]) 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 Packs without morphology data return an empty registry; packs that set
LexicalEntry.morphology_id are validated by the compiler against this LexicalEntry.morphology_id are validated by the compiler against this
registry. 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(): if not morphology_path.exists():
return MorphologyRegistry([]) 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 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: def test_eval_run_rejects_unsafe_lane() -> None:
response = _request( response = _request(
"POST", "POST",
@ -252,6 +298,11 @@ def test_full_w026_route_table_preserves_teaching_and_pack_bytes() -> None:
("GET", "/proposals", None), ("GET", "/proposals", None),
("GET", "/evals", None), ("GET", "/evals", None),
("GET", "/evals/contemplation_quality", 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", "POST",
"/evals/run", "/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 __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: 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 payload["ok"] is False
assert isinstance(payload["generated_at"], str) assert isinstance(payload["generated_at"], str)
assert payload["error"] == {"code": "not_found", "message": "missing"} 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", "source_digest",
"calibration_evidence_ref" "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": [ "MathProposalDetail": [
"wrong_zero_assertion", "wrong_zero_assertion",
"proposed_change_payload", "proposed_change_payload",

View file

@ -19,10 +19,22 @@ import { describe, expect, it } from "vitest";
* only shrink. * only shrink.
*/ */
// Empty: every engine schema is now mirrored. New unmirrored schemas may be // Shrink-only debt: backend schemas shipped ahead of their TS mirror. A class
// allowlisted here as shrink-only debt (a class gaining a mirror while listed // gaining a TS interface while still listed here fails the gate.
// fails the gate). // CORE-Logos read models (LG-1): backend readers shipped without the /logos
const NOT_YET_MIRRORED = new Set<string>([]); // 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 UI_ROOT = join(__dirname, "..", "..", "..");
const snapshot: Record<string, string[]> = JSON.parse( const snapshot: Record<string, string[]> = JSON.parse(

View file

@ -18,7 +18,7 @@ from core.epistemic_state import (
epistemic_state_for_grounding_source, epistemic_state_for_grounding_source,
normative_detail_from_verdicts, 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.journal import DEFAULT_JOURNAL_DIR, TurnJournal, TurnJournalEntry
from workbench.evidence_bundle import build_evidence_bundle from workbench.evidence_bundle import build_evidence_bundle
from workbench.tour import determinism_tour from workbench.tour import determinism_tour
@ -192,6 +192,10 @@ class WorkbenchApi:
if method == "GET" and path.startswith("/math-proposals/"): if method == "GET" and path.startswith("/math-proposals/"):
proposal_id = unquote(path.removeprefix("/math-proposals/")) proposal_id = unquote(path.removeprefix("/math-proposals/"))
return ApiResponse(200, ok(readers.read_math_proposal(proposal_id))) 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": if method == "GET" and path == "/packs":
limit, offset = _pagination(query) limit, offset = _pagination(query)
return ApiResponse( return ApiResponse(
@ -431,6 +435,44 @@ class WorkbenchApi:
return ApiResponse(200, ok(comparison)) return ApiResponse(200, ok(comparison))
return ApiResponse(404, error("not_found", f"route not found: {method} {path}")) 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: def _math_ratify(self, proposal_id: str, body: bytes) -> ApiResponse:
"""Route ratification by change_kind; in-process execution with allowlist checks.""" """Route ratification by change_kind; in-process execution with allowlist checks."""
category: str | None = None 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 dataclasses import asdict, dataclass, field
from datetime import datetime, timezone from datetime import datetime, timezone
from enum import Enum
from typing import Any, Literal from typing import Any, Literal
@ -60,10 +61,12 @@ def utc_now() -> str:
def to_data(value: Any) -> Any: def to_data(value: Any) -> Any:
if isinstance(value, Enum):
return value.value
if hasattr(value, "as_dict") and callable(value.as_dict): if hasattr(value, "as_dict") and callable(value.as_dict):
return value.as_dict() return value.as_dict()
if hasattr(value, "__dataclass_fields__"): if hasattr(value, "__dataclass_fields__"):
return asdict(value) return to_data(asdict(value))
if isinstance(value, dict): if isinstance(value, dict):
return {str(k): to_data(v) for k, v in value.items()} return {str(k): to_data(v) for k, v in value.items()}
if isinstance(value, (list, tuple)): if isinstance(value, (list, tuple)):
@ -615,6 +618,147 @@ class PackDetail(PackSummary):
manifest: dict[str, Any] = field(default_factory=dict) 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[ AuditSource = Literal[
"engine_state_manifest", "engine_state_manifest",
"math_proposal_log", "math_proposal_log",