diff --git a/.github/workflows/ahbg-runtime.yml b/.github/workflows/ahbg-runtime.yml index bedbfdbc..1dea9dde 100644 --- a/.github/workflows/ahbg-runtime.yml +++ b/.github/workflows/ahbg-runtime.yml @@ -4,6 +4,7 @@ on: pull_request: paths: - "ahbg/runtime/**" + - "ahbg/integration/**" - "ahbg/grok/**" - "libs/ucns/**" - ".github/workflows/ahbg-runtime.yml" @@ -11,6 +12,7 @@ on: branches: [main] paths: - "ahbg/runtime/**" + - "ahbg/integration/**" - "ahbg/grok/**" - "libs/ucns/**" - ".github/workflows/ahbg-runtime.yml" diff --git a/.github/workflows/ahbg-uchc-structural.yml b/.github/workflows/ahbg-uchc-structural.yml new file mode 100644 index 00000000..0bb02659 --- /dev/null +++ b/.github/workflows/ahbg-uchc-structural.yml @@ -0,0 +1,117 @@ +# Read-only AHBG consumption of the exact extracted UCHC a2 structural field. +# This job does not publish UCHC, graduate the English domain, or choose senses. +name: AHBG UCHC structural corpus +on: + pull_request: + paths: + - "ahbg/integration/uchc_structural.py" + - "ahbg/integration/UCHC_STRUCTURAL.json" + - "ahbg/integration/tests/test_uchc_structural_installed.py" + - "ahbg/runtime/runtime.py" + - "ahbg/runtime/protocol.py" + - "research/zfae/uchc_input.py" + - ".github/workflows/ahbg-uchc-structural.yml" + push: + branches: [main] + paths: + - "ahbg/integration/uchc_structural.py" + - "ahbg/integration/UCHC_STRUCTURAL.json" + - "ahbg/integration/tests/test_uchc_structural_installed.py" + - "ahbg/runtime/runtime.py" + - "ahbg/runtime/protocol.py" + - "research/zfae/uchc_input.py" + - ".github/workflows/ahbg-uchc-structural.yml" + +permissions: + contents: read + +jobs: + ahbg-structural-consumer: + runs-on: ubuntu-24.04 + env: + SOURCE_DATE_EPOCH: '1790308800' + steps: + - uses: actions/checkout@v6 + with: + persist-credentials: false + - uses: actions/checkout@v6 + with: + repository: The-Interdependency/uchc + ref: c18d984438bc30764edb05fd5ccc05691958a24e + path: _deps/uchc + persist-credentials: false + - uses: actions/checkout@v6 + with: + repository: globalwordnet/english-wordnet + ref: dc343f2683279ecbb13fab4e2fd778d7b162d287 + path: _deps/oewn + persist-credentials: false + - uses: actions/checkout@v6 + with: + repository: The-Interdependency/ucns + ref: 4f863ad37096b7baab8f62820ad5cb937b62a3a7 + path: _deps/ucns + persist-credentials: false + - uses: actions/setup-python@v6 + with: + python-version: '3.12.14' + - name: Preflight and build pinned wheel + run: | + set -euo pipefail + python - <<'PY' + import os, sqlite3, shutil + assert shutil.disk_usage('.').free > 8 * 1024**3, 'insufficient disk' + assert os.sysconf('SC_PHYS_PAGES') * os.sysconf('SC_PAGE_SIZE') >= 5 * 1024**3, 'insufficient memory' + assert sqlite3.sqlite_version == '3.45.1', 'SQLite writer differs from pinned byte evidence' + PY + python -m pip install setuptools==82.0.1 wheel==0.46.3 packaging==26.0 PyYAML==6.0.2 + python -m pip wheel --no-deps --no-build-isolation ./_deps/uchc -w /tmp/ahbg-uchc-a2-dist + python - <<'PY' + import hashlib, json, pathlib + lock = json.loads(pathlib.Path('ahbg/integration/UCHC_STRUCTURAL.json').read_text()) + artifact = pathlib.Path('/tmp/ahbg-uchc-a2-dist') / lock['wheel'] + with artifact.open('rb') as f: + assert hashlib.file_digest(f, 'sha256').hexdigest() == lock['wheel_sha256'] + PY + python -m venv .uchc-a2 + .uchc-a2/bin/python -m pip install PyYAML==6.0.2 + .uchc-a2/bin/python -m pip install --no-index --no-deps /tmp/ahbg-uchc-a2-dist/uchc-0.1.0a2-py3-none-any.whl + - name: Build and verify full structural corpus through installed UCHC + run: | + set -euo pipefail + .uchc-a2/bin/python -I -m english_gonol.structural_construct build \ + --source-root "$PWD/_deps/oewn/src/yaml" \ + --ucns-source-root "$PWD/_deps/ucns" \ + --out-dir "$PWD/_state/structural" + .uchc-a2/bin/python -I -m english_gonol.structural_construct verify "$PWD/_state/structural" + python - <<'PY' + import hashlib, json, pathlib + lock = json.loads(pathlib.Path('ahbg/integration/UCHC_STRUCTURAL.json').read_text()) + root = pathlib.Path('_state/structural') + manifest = json.loads((root/'manifest.json').read_text()) + with (root/'construct.db').open('rb') as f: + observed = hashlib.file_digest(f, 'sha256').hexdigest() + assert observed == lock['database_sha256'], 'physical structural bytes mismatch' + assert manifest['receipt_sha256'] == lock['logical_receipt'], 'structural logical receipt mismatch' + assert manifest['metadata']['implementation_sha256'] == lock['implementation_sha256'] + assert manifest['metadata']['complete_corpus'] is True + assert manifest['metadata']['profile']['profile'] == lock['profile'] + PY + - name: Consume full native structural field inside AHBG + env: + UCHC_STRUCTURAL_FIELD: ${{ github.workspace }}/_state/structural + UCHC_CORPUS_REQUIRED: '1' + run: | + set -euo pipefail + .uchc-a2/bin/python -m unittest discover -s ahbg/integration/tests \ + -p 'test_uchc_structural_installed.py' -v + - name: Retain identity and test evidence, never republish corpus bytes + if: always() + uses: actions/upload-artifact@v6 + with: + name: ahbg-uchc-structural-evidence + path: | + _state/structural/manifest.json + ahbg/integration/UCHC_STRUCTURAL.json + retention-days: 14 + if-no-files-found: warn diff --git a/.github/workflows/uchc-input.yml b/.github/workflows/uchc-input.yml index e10f48da..9acb76d3 100644 --- a/.github/workflows/uchc-input.yml +++ b/.github/workflows/uchc-input.yml @@ -3,10 +3,10 @@ name: UCHC input integration on: pull_request: - paths: ['.github/workflows/uchc-input.yml', 'research/zfae/**', 'integration/uchc/**', 'stack-manifest.json', 'STACK_MANIFEST.md'] + paths: ['.github/workflows/uchc-input.yml', 'research/zfae/**', 'integration/uchc/**', 'ahbg/integration/**', 'ahbg/runtime/**', 'stack-manifest.json', 'STACK_MANIFEST.md'] push: branches: [main] - paths: ['.github/workflows/uchc-input.yml', 'research/zfae/**', 'integration/uchc/**', 'stack-manifest.json', 'STACK_MANIFEST.md'] + paths: ['.github/workflows/uchc-input.yml', 'research/zfae/**', 'integration/uchc/**', 'ahbg/integration/**', 'ahbg/runtime/**', 'stack-manifest.json', 'STACK_MANIFEST.md'] permissions: contents: read jobs: @@ -66,7 +66,7 @@ jobs: shell: bash run: | set -euo pipefail - UCHC_INPUT_DATABASE="$PWD/evidence/construct/construct.db" \ + UCHC_CORPUS_REQUIRED=1 UCHC_INPUT_DATABASE="$PWD/evidence/construct/construct.db" \ .clean/bin/python -I -m pytest -q stack/integration/uchc/tests/test_input.py --junitxml=evidence/integration.xml .clean/bin/python -I producer/tools/verify_inference_corpus.py \ --database evidence/construct/construct.db \ @@ -76,6 +76,11 @@ jobs: .clean/bin/python -I stack/research/zfae/uchc_input.py \ --database evidence/construct/construct.db --source-id ci:exact-input \ --text 'alpha letter alpha' > evidence/consumer-frame.json + # AHBG consumes this same installed wheel and complete verified + # construct, not a mock or a second Stack language implementation. + UCHC_INPUT_DATABASE="$PWD/evidence/construct/construct.db" \ + .clean/bin/python -m unittest discover \ + -s stack/ahbg/integration/tests -p 'test_uchc_installed.py' -v cp stack/research/zfae/UCHC_INPUT.json evidence/ git -C stack rev-parse HEAD > evidence/STACK_COMMIT git -C producer rev-parse HEAD > evidence/UCHC_COMMIT diff --git a/ahbg/README.md b/ahbg/README.md index 5fd67e39..a8fd8796 100644 --- a/ahbg/README.md +++ b/ahbg/README.md @@ -20,6 +20,17 @@ The AHBG board is a **game projection of UCNS geometry**, not an independently i For implementation, depend on the canonical `ucns` package/source. In stack, `libs/ucns/` is the pinned canonical repository view and `research/ucns/` is noncanonical stack-local research; neither permits silently forking the geometry. +## Native UCHC language construction + +AHBG now has an **optional native UCHC consumer**. Separately pinned clean-installed +UCHC candidate wheels supply exact English input and structural constructions, +source-bound word and definition records, full recovery, replay, local closure, +and admission receipts. AHBG can inspect these results +and pass read-only evidence to agents without defining new geometry, meanings, +game actions, or consciousness criteria. Invalid wheel/database/receipt identity +fails closed. See [the executable UCHC integration contract](integration/UCHC.md) +and its full-corpus acceptance gate. + ## Two coordination modes — do not mix them AHBG now has two different work modes. diff --git a/ahbg/integration/UCHC.md b/ahbg/integration/UCHC.md new file mode 100644 index 00000000..c7d479dc --- /dev/null +++ b/ahbg/integration/UCHC.md @@ -0,0 +1,209 @@ +# AHBG ↔ UCHC: native consumer contract + +## Done when + +AHBG installs the source- and wheel-locked UCHC candidate; independently verifies +its package files and complete-corpus identity; constructs, queries and replays +native English input; preserves unknowns; exposes the exact evidence to the +agent observation and persisted turn record; and refuses changed artifacts or +unlicensed operations. The AHBG engine and UCNS movement remain unchanged. + +**Authority:** UCHC constructs. UCNS owns geometry. Stack remains the English +forge pending graduation. AHBG observes those native objects, never reimplements +them or treats structural similarity as a game rule or consciousness proof. + +## Source and installation + +The canonical immutable candidate selection already exists at +`research/zfae/UCHC_INPUT.json`. AHBG intentionally reuses the existing +`research/zfae/uchc_input.py` verifier. The candidate is UCHC `0.1.0a1`, +source `3d82007b88f74ded88ca913d6e85bcd6f6ff024d`. +Wheel SHA-256, logical receipt, and database byte SHA-256 are **read from the +trusted lock**, never discovered from the supplied database or guessed at run time. + +Use `.github/workflows/uchc-input.yml` as the full-corpus reproducibility and +installed-artifact acceptance path. It fetches exactly pinned sources, builds +and verifies wheel bytes, constructs the full OEWN corpus, and runs both the +existing ZFAE consumer and AHBG's real native-language integration tests. + +For a local install, first verify the wheel SHA-256 against the lock and +install the exact wheel from a local archive into a separate environment. +Use the corresponding verified complete `construct.db`: + +```bash +python -m ahbg.integration.uchc \ + --database /artifacts/construct.db \ + text 'alpha letter alpha' --source-id 'ahbg:query:1' + +python -m ahbg.integration.uchc \ + --database /artifacts/construct.db word alpha + +python -m ahbg.integration.uchc \ + --database /artifacts/construct.db definition 1 + +python -m ahbg.integration.uchc \ + --database /artifacts/construct.db replay /artifacts/saved-frame.json + +PYTHONPATH=.:libs/ucns/src python -m ahbg.runtime play \ + --agent a0 --turns 3 --out /tmp/ahbg-native \ + --uchc-database /artifacts/construct.db +``` + +The query CLI returns a source-bound frame, UCHC-native word object, complete +native definition with recovered source text, or verified replay. Definition +identities are corpus-local, not assumed universal IDs. In Python: + +```python +from pathlib import Path +from ahbg.integration.uchc import UchcSession, UchcSettings + +with UchcSession(UchcSettings(Path("/artifacts/construct.db"))) as uchc: + frame = uchc.resolve_text("alpha letter alpha", source_id="ahbg:probe") + frame.require_complete() + assert frame.recover_utf8() == b"alpha letter alpha" + assert uchc.replay(frame.to_bytes()) == frame + all_candidates = uchc.definition_candidates(frame) + assert all_candidates + print(uchc.capabilities) +``` + +## Structural profile (separate exact wheel) + +The **newer** UCHC `0.1.0a2` is also independently consumable. Its real +`StructuralField` performs text construction, admitted word/definition +recovery, exact record replay, and registered ordered-text/sentence composition. +It preserves both definition constituent topologies and definition succession. +It never supplies missing UCNS geometric displacement. + +The exact candidate identities are in +[`UCHC_STRUCTURAL.json`](UCHC_STRUCTURAL.json): UCHC source +`c18d984438bc30764edb05fd5ccc05691958a24e`, wheel hash +`7f05fbbb...`, complete 711,983,104-byte structural database hash +`13ba2411...`, logical receipt `26ab095f...`. The consumer verifies the +**full** stored identities, not the abbreviated identifiers printed here. + +Build the full structural field through the exact wheel as specified in +[`ahbg-uchc-structural.yml`](../../.github/workflows/ahbg-uchc-structural.yml), +then run: + +```bash +PYTHONPATH=.:libs/ucns/src python -m ahbg.runtime play \ + --agent a0 --turns 3 --out /tmp/ahbg-uchc-structural \ + --uchc-structural-field /artifacts/structural-field +``` + +Native Python surface: + +```python +from pathlib import Path +from ahbg.integration.uchc_structural import ( + UchcStructuralSession, UchcStructuralSettings +) + +with UchcStructuralSession( + UchcStructuralSettings(Path("/artifacts/structural-field")) +) as uchc: + graph = uchc.construct("Dr. creates geometry.") + assert uchc.replay(graph) == graph + word = uchc.word("geometry") + relation = uchc.compose(graph, relation="ordered-text") +``` + +The input and structural profiles use **different pinned wheel versions and +different databases**. One Python process selects exactly one; attempting to +activate both fails explicitly. Multiple profiles can be compared through +separately installed processes without redefining their geometric authority. + +## Runtime integration + +Set `RuntimeConfig.uchc_database` (or `--uchc-database`) to activate native +input evidence, or `uchc_structural_field` (or `--uchc-structural-field`) for the +separately pinned structural profile; leave both unset to run AHBG without an +installed UCHC wheel. +An optional `uchc_lock` selects an explicitly trusted artifact lock. +`uchc_require_complete=True` rejects unadmitted glyphs/words rather than +concealing them. `uchc_max_text_bytes` is an optional explicit memory bound: +oversized inputs fail rather than truncate. The default does not impose a +hidden text-length limit. + +Each inbox message containing a `text` field is resolved in original order. +No normalization, case folding, punctuation erasure, or inferred translation. +A message's exact source-scoped native input frame and receipt appear in +`observation.language_evidence` for a harness turn, alongside the unmodified +`observation.inbox`. These are then persisted in +`result.json.turn_records[].uchc_evidence`; the immutable UCHC wheel, +database and lock identities appear in `result.json.provenance.uchc`. +No modification to UCNS geometry, legal actions, game transitions, agent +capabilities, or reward signals occurs. + +All definitions are available via native +`session.word`, `session.definition`, `session.recover_definition`, and +`session.definition_candidates`; the inline frame preserves source-order +definition IDs and all native word identities. Semantic selection is not +licensed. Unknown input is retained as `PARTIAL` unless strict admission is +explicitly enabled, which raises `UchcConsumerError`. + +## Local HTTP and mobile access + +AHBG's built-in HTTP bridge can now expose the same UCHC package selected by +its **operator**. Clients never choose wheel paths, source commits or a corpus. + +```bash +PYTHONPATH=.:libs/ucns/src python -m ahbg.runtime.server \ + --port 8765 --uchc-database /artifacts/construct.db + +# Or run the separate a2 structural profile instead: +PYTHONPATH=.:libs/ucns/src python -m ahbg.runtime.server \ + --port 8765 --uchc-structural-field /artifacts/structural-field +``` + +`GET /uchc/capabilities` returns exact supported operations and provenance. +`POST /uchc/query` with a named supported native operation returns its real +source-bound results (English text, UTF-8, word, definition, definition +candidates, replay, structural construction, and allowed compositions). + +`POST /session/{id}/language` accepts +`{"text":"original text"}`, builds native evidence and includes it in the +next agent observation alongside the unchanged raw message. Its source-bound +evidence is also appended to `language.jsonl` next to the persisted game +state, with full UCHC artifact identity. The source evidence is read-only +with respect to game rules/actions. + +HTTP UCHC is restricted to **loopback binding**, because this adapter does not +provide authorization for remote callers. Inputs have an explicit 64 KiB HTTP +transport ceiling; malformed operations and unsupported semantic inference +fail with a request error instead of inventing an answer. + +## Rejection and proof + +- Missing UCHC wheel, editable/source-shadowed installation, wrong version, + altered installed files or wheel hash: refuse before game initialization. +- Wrong complete construct SHA-256, logical receipt, schema or integrity: + refuse before game initialization. +- Unexpected input shape, non-scalar Unicode, incomplete strict admission, + stale frame, mismatched record, unsupported/unpublished operation or + altered replay: refuse with an explicit error. +- Neither a passing test nor a matching hash proves external authentication, + semantic truth, sense choice, neural inference, cross-origin Hilbert geometry, + or consciousness. + +Fast wiring gate: +`PYTHONPATH=.:libs/ucns/src python -m unittest discover -s ahbg/runtime/tests -q` + +Real input producer acceptance gate: +`.github/workflows/uchc-input.yml`, including +`ahbg/integration/tests/test_uchc_installed.py` against the **complete** +source-pinned corpus. Structural acceptance independently executes +`.github/workflows/ahbg-uchc-structural.yml` against the distinct complete +UCHC `0.1.0a2` field. A mocked reader alone is not sufficient evidence. + +## hmmm + +UCHC `0.1.0a2` structural admission now has its own independent AHBG +consumer and complete-corpus gate. The new ordinal construction, polyglot +label layer, binary producer and Hilbert frame bridge still have separate +implementation and/or publication gates. No unfinished machinery may be +silently substituted for either pinned candidate package. AHBG may adopt +each only after its source and wheel/producer receipt, native semantics and +full-corpus or full-domain replay have been verified, with an explicit lock +migration. Independent UCHC graduation remains incomplete. diff --git a/ahbg/integration/UCHC_STRUCTURAL.json b/ahbg/integration/UCHC_STRUCTURAL.json new file mode 100644 index 00000000..5d4555fc --- /dev/null +++ b/ahbg/integration/UCHC_STRUCTURAL.json @@ -0,0 +1,22 @@ +{ + "schema": "interdependency.ahbg.uchc-structural-lock/1", + "version": "1.0.0", + "source_repository": "The-Interdependency/uchc", + "source_commit": "c18d984438bc30764edb05fd5ccc05691958a24e", + "package": "uchc", + "package_version": "0.1.0a2", + "wheel": "uchc-0.1.0a2-py3-none-any.whl", + "wheel_sha256": "7f05fbbbdd40eb8f7d3575450a3a54153c155c46e512a96f245ac5bdfffa76e8", + "database_sha256": "13ba24116f0ff02cfdfeb4a6b060f2672fa409b46a3b7cf26e4421c65321b063", + "logical_receipt": "26ab095f73306ff4fbd9f67d1c282c1df79f926bb75cc081c879db8a6989b58a", + "implementation_sha256": "771eb34cb98aeda541c8e9a63ba191c65baf623d1f8549bc7df3cc4891dfc504", + "profile": "english.structural/1", + "complete_corpus": true, + "authority_transfer": false, + "standing": "extracted structural candidate, not stable release or neural inference", + "hmmm": [ + "57,155 compound/punctuated lexical definition-origin attachments remain unresolved", + "Unimplemented UCNS geometric displacements and semantic inference must refuse", + "Stable release, consumer artifact reconsumption and domain graduation remain incomplete" + ] +} diff --git a/ahbg/integration/tests/test_uchc_installed.py b/ahbg/integration/tests/test_uchc_installed.py new file mode 100644 index 00000000..080a9c50 --- /dev/null +++ b/ahbg/integration/tests/test_uchc_installed.py @@ -0,0 +1,138 @@ +"""Real clean-wheel/full-corpus AHBG -> UCHC integration, no mocked producer. + +Run only in the existing UCHC full-corpus acceptance job, with +UCHC_INPUT_DATABASE pointing to the verified complete construct.db. +""" +from __future__ import annotations + +import hashlib +import json +import os +from pathlib import Path +import sys +import tempfile +import unittest + +ROOT = Path(__file__).resolve().parents[3] +sys.path.insert(0, str(ROOT)) +sys.path.insert(0, str(ROOT / "libs" / "ucns" / "src")) + +from ahbg.integration.uchc import UchcConsumerError, UchcSession, UchcSettings +from ahbg.runtime import RuntimeConfig, run_plane + + +class RecordingAgent: + def __init__(self): + self.observations = [] + + def manifest(self): + return {"agent": "uchc-installed-test", "capabilities": ["observe", "plan"]} + + def plan(self, observation): + self.observations.append(observation) + return {"session_id": observation["session_id"], "turn": observation["turn"], + "intents": [], "note": "candidate evidence is not a decision"} + + +class InstalledUchcTests(unittest.TestCase): + @classmethod + def setUpClass(cls): + raw = os.environ.get("UCHC_INPUT_DATABASE") + if not raw: + if os.environ.get("UCHC_CORPUS_REQUIRED") == "1": + raise AssertionError("UCHC_INPUT_DATABASE is required: full-corpus gate not configured") + raise unittest.SkipTest("Full-corpus installed wheel gate runs in dedicated CI") + cls.database = Path(raw) + cls.lock = ROOT / "research" / "zfae" / "UCHC_INPUT.json" + + def settings(self, **extra): + return UchcSettings(database=self.database, lock=self.lock, **extra) + + def test_recovery_all_definitions_query_replay_and_unknowns(self): + with UchcSession(self.settings()) as session: + provenance = session.provenance + lock = json.loads(self.lock.read_text()) + self.assertEqual(provenance["wheel_sha256"], lock["wheel_sha256"]) + self.assertGreater(provenance["verified_installed_files"], 0) + self.assertFalse(provenance["neural_inference"]) + frame = session.resolve_text("alpha letter alpha", source_id="ahbg:test:1") + frame.require_complete() + self.assertEqual(frame.recover_utf8(), b"alpha letter alpha") + self.assertEqual(session.replay(frame.to_bytes()), frame) + word = session.word("alpha") + self.assertIsNotNone(word) + self.assertTrue(word.definition_ids) + for definition_id in word.definition_ids: + definition = session.definition(definition_id) + self.assertEqual(session.recover_definition(definition), definition.text) + self.assertIsNone(session.word("🧪🧪")) + partial = session.resolve_text("🧪🧪", source_id="ahbg:unknown") + self.assertEqual(partial.recover_utf8(), "🧪🧪".encode()) + annotations = session.annotate_messages( + [{"text": "alpha letter alpha"}, {"source": "none"}, + {"text": "🧪🧪"}, {"text": "alpha letter alpha"}], + source_prefix="ahbg:fixture" + ) + self.assertEqual([e["message_ordinal"] for e in annotations], [0, 2, 3]) + self.assertEqual([e["admission"] for e in annotations], + ["COMPLETE", "PARTIAL", "COMPLETE"]) + self.assertNotEqual(annotations[0]["source_id"], annotations[2]["source_id"]) + self.assertEqual(annotations[0]["frame"]["text"], "alpha letter alpha") + tampered = bytearray(frame.to_bytes()) + tampered[-3] ^= 1 + with self.assertRaises(ValueError): + session.replay(bytes(tampered)) + with self.assertRaisesRegex(UchcConsumerError, "closed"): + session.resolve_text("alpha", source_id="after-close") + + def test_strict_admission_refuses_unknown_but_not_full_recovery(self): + with UchcSession(self.settings(require_complete=True)) as session: + self.assertEqual(session.resolve_text("alpha", source_id="ok").recover_utf8(), b"alpha") + with self.assertRaisesRegex(UchcConsumerError, "input refused"): + session.resolve_text("🧪🧪", source_id="not-admitted") + + def test_bounded_max_bytes_is_explicit_refusal_not_silent_truncation(self): + with UchcSession(self.settings(max_text_bytes=3)) as session: + self.assertEqual(session.resolve_text("a", source_id="ok").recover_utf8(), b"a") + with self.assertRaisesRegex(UchcConsumerError, "exceeds"): + session.resolve_text("alpha", source_id="reject") + + def test_runtime_replays_with_native_language_evidence(self): + with tempfile.TemporaryDirectory() as tmp: + agent = RecordingAgent() + msg = {0: [{"text": "alpha letter alpha", "source": "peer"}], + 1: [{"text": "🧪🧪", "source": "terrain"}]} + result = run_plane(agent=agent, config=RuntimeConfig( + seed=17, turns=2, turn_messages=msg, + uchc_database=str(self.database), uchc_lock=str(self.lock) + ), out_dir=tmp) + self.assertEqual(result.final_turn, 2) + self.assertEqual(agent.observations[0]["inbox"], msg[0]) + self.assertEqual(agent.observations[0]["language_evidence"][0]["admission"], + "COMPLETE") + self.assertEqual(agent.observations[1]["language_evidence"][0]["admission"], + "PARTIAL") + self.assertEqual(result.turn_records[0]["uchc_evidence"], + agent.observations[0]["language_evidence"]) + self.assertEqual(result.provenance["uchc"]["wheel_sha256"], + json.loads(self.lock.read_text())["wheel_sha256"]) + loaded = json.loads((Path(tmp) / "result.json").read_text()) + self.assertEqual(loaded["turn_records"], list(result.turn_records)) + self.assertEqual(result.turn_records[0]["plan"]["intents"], []) + + def test_invalid_package_lock_refused_before_runtime_mutation(self): + with tempfile.TemporaryDirectory() as tmp: + bad_lock = Path(tmp) / "bad-lock.json" + payload = json.loads(self.lock.read_text()) + payload["wheel_sha256"] = "0" * 64 + bad_lock.write_text(json.dumps(payload)) + out = Path(tmp) / "world" + with self.assertRaisesRegex(UchcConsumerError, "verification/open refused"): + run_plane(agent=RecordingAgent(), config=RuntimeConfig( + turns=1, uchc_database=str(self.database), uchc_lock=str(bad_lock) + ), out_dir=out) + self.assertFalse((out / "result.json").exists()) + + +if __name__ == "__main__": + unittest.main() diff --git a/ahbg/integration/tests/test_uchc_structural_installed.py b/ahbg/integration/tests/test_uchc_structural_installed.py new file mode 100644 index 00000000..18752fc7 --- /dev/null +++ b/ahbg/integration/tests/test_uchc_structural_installed.py @@ -0,0 +1,140 @@ +"""Native AHBG ↔ UCHC 0.1.0a2 structural corpus acceptance. + +UCHC_STRUCTURAL_FIELD must point to the complete exact, clean-wheel-built +field; no partial fixture can satisfy this test. +""" +from __future__ import annotations + +import json +import os +from pathlib import Path +import sqlite3 +import sys +import tempfile +import unittest + +ROOT = Path(__file__).resolve().parents[3] +sys.path.insert(0, str(ROOT)) +sys.path.insert(0, str(ROOT / "libs" / "ucns" / "src")) + +from ahbg.integration.uchc_structural import ( + UchcStructuralSession, UchcStructuralSettings, UchcStructuralError +) +from ahbg.runtime import RuntimeConfig, run_plane + + +class Observer: + def __init__(self): + self.seen = [] + + def manifest(self): + return {"agent": "structural-consumer-test", + "capabilities": ["observe", "plan"]} + + def plan(self, observation): + self.seen.append(observation) + return {"session_id": observation["session_id"], "turn": observation["turn"], + "intents": [], "note": "no geometry or sense inference"} + + +class StructuralInstalledTests(unittest.TestCase): + @classmethod + def setUpClass(cls): + raw = os.getenv("UCHC_STRUCTURAL_FIELD") + if not raw: + if os.environ.get("UCHC_CORPUS_REQUIRED") == "1": + raise AssertionError("UCHC_STRUCTURAL_FIELD is required (complete producer field)") + raise unittest.SkipTest("Full-corpus installed wheel gate runs in dedicated CI") + cls.field_dir = Path(raw) + cls.lock = ROOT / "ahbg" / "integration" / "UCHC_STRUCTURAL.json" + + def test_native_structural_construction_replay_and_all_public_queries(self): + with UchcStructuralSession(UchcStructuralSettings(self.field_dir)) as session: + provenance = session.provenance + pinned = json.loads(self.lock.read_text()) + self.assertEqual(provenance["wheel_sha256"], pinned["wheel_sha256"]) + self.assertEqual(provenance["database_sha256"], pinned["database_sha256"]) + self.assertFalse(provenance["authority_transfer"]) + text = "Dr. creates geometry. Dr. creates 3.14..." + graph = session.construct(text) + self.assertEqual(graph["source"], text) + self.assertEqual(session.replay(graph), graph) + closed_word = session.word("geometry") + self.assertTrue(closed_word["closed"]) + self.assertIn("glyph_axis_set", closed_word) + with sqlite3.connect(f"file:{self.field_dir / 'construct.db'}?mode=ro", + uri=True) as db: + definition_id = db.execute( + "SELECT id FROM records WHERE kind='definition' ORDER BY id LIMIT 1" + ).fetchone()[0] + definition = session.definition(definition_id) + self.assertEqual(definition["record_id"], definition_id) + self.assertTrue(definition["parts"]) + result = session.compose(graph, relation="ordered-text") + self.assertIsInstance(result, dict) + with self.assertRaisesRegex(UchcStructuralError, "unlicensed"): + session.compose(graph, relation="invented-global-transport") + with self.assertRaisesRegex(UchcStructuralError, "UNIMPLEMENTED"): + session.require_capability("ucns.global_displacement") + unknown = session.construct("🧪🧪") + self.assertTrue(unknown["refusals"]) + evidence = session.annotate_messages( + [{"text": text}, {"kind": "metric"}, {"text": "🧪🧪"}], + source_prefix="ahbg:structural") + self.assertEqual([x["message_ordinal"] for x in evidence], [0, 2]) + self.assertEqual(evidence[0]["frame"]["source"], text) + self.assertEqual(evidence[1]["admission"], "PARTIAL") + with self.assertRaisesRegex(UchcStructuralError, "closed"): + session.word("geometry") + + def test_strict_admission_refuses_and_preserves_approved_source(self): + with UchcStructuralSession(UchcStructuralSettings( + self.field_dir, require_complete=True + )) as session: + self.assertEqual(session.construct("geometry")["source"], "geometry") + with self.assertRaisesRegex(UchcStructuralError, "incomplete"): + session.construct("🧪🧪") + + def test_structural_runtime_and_persistence(self): + with tempfile.TemporaryDirectory() as tmp: + agent = Observer() + result = run_plane(agent=agent, config=RuntimeConfig( + turns=2, seed=23, uchc_structural_field=str(self.field_dir), + turn_messages={0: [{"text": "geometry", "source": "peer"}], + 1: [{"text": "🧪🧪", "source": "terrain"}]} + ), out_dir=tmp) + self.assertEqual(result.final_turn, 2) + self.assertEqual(result.config["uchc"]["profile"], "english.structural/1") + self.assertEqual(result.provenance["uchc"]["profile"], "english.structural/1") + self.assertEqual(agent.seen[0]["inbox"][0]["text"], "geometry") + self.assertEqual(agent.seen[0]["language_evidence"][0]["frame"]["source"], + "geometry") + self.assertEqual(result.turn_records[1]["uchc_evidence"][0]["admission"], + "PARTIAL") + self.assertEqual(len(result.turn_records[0]["plan"]["intents"]), 0) + saved = json.loads((Path(tmp) / "result.json").read_text()) + self.assertEqual(saved["provenance"]["uchc"], result.provenance["uchc"]) + + def test_wrong_structural_lock_fails_before_game_setup(self): + with tempfile.TemporaryDirectory() as tmp: + payload = json.loads(self.lock.read_text()) + payload["wheel_sha256"] = "0" * 64 + altered = Path(tmp) / "altered-lock.json" + altered.write_text(json.dumps(payload)) + with self.assertRaisesRegex(UchcStructuralError, "refused"): + with UchcStructuralSession(UchcStructuralSettings( + self.field_dir, lock=altered + )): + self.fail("wrong wheel was admitted") + + def test_explicit_byte_limit_refuses_without_truncation(self): + with UchcStructuralSession(UchcStructuralSettings( + self.field_dir, max_text_bytes=3 + )) as session: + self.assertEqual(session.construct("a")["source"], "a") + with self.assertRaisesRegex(UchcStructuralError, "exceeds"): + session.construct("geometry") + + +if __name__ == "__main__": + unittest.main() diff --git a/ahbg/integration/uchc.py b/ahbg/integration/uchc.py new file mode 100644 index 00000000..1c02831e --- /dev/null +++ b/ahbg/integration/uchc.py @@ -0,0 +1,269 @@ +"""AHBG's source-bound consumer of the installed UCHC English construction. + +Only UCHC creates native language objects. This module verifies the already +locked candidate wheel and construct, publishes exact evidence, and refuses +identity or admission failures. No game action or semantic sense is inferred. +""" +from __future__ import annotations + +from dataclasses import asdict, dataclass +from hashlib import sha256 +import json +from pathlib import Path +from typing import Any, Mapping + +STACK_ROOT = Path(__file__).resolve().parents[2] +DEFAULT_LOCK = STACK_ROOT / "research" / "zfae" / "UCHC_INPUT.json" +SCHEMA = "interdependency.ahbg.uchc-evidence/1" + + +class UchcConsumerError(ValueError): + """UCHC dependency, provenance, input or recovery contract failed.""" + + +@dataclass(frozen=True) +class UchcSettings: + database: Path + lock: Path = DEFAULT_LOCK + require_complete: bool = False + max_text_bytes: int | None = None + + +def _canonical(value: Any) -> bytes: + return json.dumps(value, sort_keys=True, ensure_ascii=False, + separators=(",", ":"), allow_nan=False).encode("utf-8") + + +class UchcSession: + """One verified corpus snapshot reused for the whole benchmark run. + + Open requires the real clean-installed UCHC wheel and complete corpus. + The Stack ZFAE verifier is reused, not reimplemented or weakened. + """ + + def __init__(self, settings: UchcSettings): + if not isinstance(settings, UchcSettings): + raise UchcConsumerError("UCHC settings must be UchcSettings") + if type(settings.require_complete) is not bool: + raise UchcConsumerError("require_complete must be boolean") + if settings.max_text_bytes is not None and ( + type(settings.max_text_bytes) is not int or settings.max_text_bytes <= 0 + ): + raise UchcConsumerError("max_text_bytes must be positive or None") + self.settings = settings + self._corpus: Any = None + self._lock: dict[str, Any] = {} + self._provenance: dict[str, Any] = {} + + def __enter__(self) -> "UchcSession": + if self._corpus is not None: + raise UchcConsumerError("UCHC session already open") + try: + raw = self.settings.lock.read_bytes() + lock = json.loads(raw) + if not isinstance(lock, dict): + raise UchcConsumerError("UCHC lock must be an object") + # This import is deliberately deferred: AHBG without UCHC enabled + # must not require an installed candidate wheel or the large corpus. + from research.zfae.uchc_input import load_input, verify_install + checked = verify_install(lock) + corpus = load_input(self.settings.database, lock) + if corpus.identity.database_sha256 != lock["database_sha256"]: + corpus.close() + raise UchcConsumerError("UCHC construct identity differs from lock") + self._corpus = corpus + self._lock = lock + self._provenance = { + "schema": "interdependency.ahbg.uchc-provenance/1", + "lock_sha256": sha256(raw).hexdigest(), + "source_repository": lock["source_repository"], + "source_commit": lock["source_commit"], + "package_version": checked["package_version"], + "wheel_sha256": checked["wheel_sha256"], + "verified_installed_files": checked["verified_files"], + "construct": asdict(corpus.identity), + "authority_transfer": False, + "neural_inference": False, + "standing": "receipt-bound input candidate, not semantic inference", + } + return self + except Exception as exc: + self.close() + if isinstance(exc, UchcConsumerError): + raise + raise UchcConsumerError(f"UCHC verification/open refused: {exc}") from exc + + def __exit__(self, *_args: object) -> None: + self.close() + + def close(self) -> None: + if self._corpus is not None: + self._corpus.close() + self._corpus = None + + @property + def provenance(self) -> dict[str, Any]: + self._require_open() + return json.loads(_canonical(self._provenance)) + + def _require_open(self) -> Any: + if self._corpus is None: + raise UchcConsumerError("UCHC session is closed") + return self._corpus + + def resolve_text(self, text: str, *, source_id: str) -> Any: + corpus = self._require_open() + if type(text) is not str or type(source_id) is not str or not source_id: + raise UchcConsumerError("text must be str and source_id nonempty str") + if self.settings.max_text_bytes is not None and ( + len(text.encode("utf-8")) > self.settings.max_text_bytes + ): + raise UchcConsumerError("UCHC input exceeds explicitly configured byte limit") + try: + frame = corpus.resolve_text(text, source_id=source_id) + if self.settings.require_complete: + frame.require_complete() + if frame.recover_utf8() != text.encode("utf-8"): + raise UchcConsumerError("UCHC source recovery mismatch") + if corpus.replay(frame.to_bytes()) != frame: + raise UchcConsumerError("UCHC frame replay mismatch") + return frame + except (ValueError, UnicodeError, TypeError) as exc: + raise UchcConsumerError(f"UCHC input refused: {exc}") from exc + + def resolve_utf8(self, raw: bytes, *, source_id: str) -> Any: + if type(raw) is not bytes: + raise UchcConsumerError("UTF-8 input must be bytes") + return self.resolve_text(raw.decode("utf-8", errors="strict"), source_id=source_id) + + @property + def capabilities(self) -> dict[str, str]: + """Exact AHBG-facing operations, not a claim about unshipped UCHC work.""" + self._require_open() + return { + "english.exact_text": "AVAILABLE", + "english.exact_utf8": "AVAILABLE", + "english.word": "AVAILABLE", + "english.definition": "AVAILABLE", + "english.definition_candidates": "AVAILABLE", + "english.frame_replay": "AVAILABLE", + "english.structural_profile_v2": "NOT_IN_PINNED_ARTIFACT", + "english.ordinal": "NOT_IN_PINNED_ARTIFACT", + "polyglot.labels": "NOT_IN_PINNED_ARTIFACT", + "binary.native_sequences": "NOT_IN_PINNED_ARTIFACT", + "hilbert.frame_bridge": "UNIMPLEMENTED", + "semantic.sense_selection": "UNIMPLEMENTED", + } + + def require_capability(self, operation: str) -> None: + status = self.capabilities.get(operation, "UNKNOWN_OPERATION") + if status != "AVAILABLE": + raise UchcConsumerError(f"UCHC {operation!r} refused: {status}") + + def definition_candidates(self, frame: Any) -> dict[int, list[dict[str, Any]]]: + """Recover every native candidate definition for admitted frame words. + + No sense is chosen, no lexical content is reconstructed in AHBG. + """ + corpus = self._require_open() + if corpus.replay(frame.to_bytes()) != frame: + raise UchcConsumerError("definition candidate frame does not replay") + records: dict[int, list[dict[str, Any]]] = {} + for word in frame.words: + native = [] + for identity in word.definition_ids: + definition = corpus.definition(identity) + recovered = corpus.recover_definition(definition) + if recovered != definition.text: + raise UchcConsumerError("candidate definition recovery mismatch") + native.append({"record": asdict(definition), "recovered": recovered}) + records[word.gonol.word_id] = native + return records + + def word(self, surface: str) -> Any: + return self._require_open().word(surface) + + def definition(self, definition_id: int) -> Any: + return self._require_open().definition(definition_id) + + def recover_definition(self, definition: Any) -> str: + return self._require_open().recover_definition(definition) + + def replay(self, frame_bytes: bytes) -> Any: + return self._require_open().replay(frame_bytes) + + def annotate_messages(self, messages: list[Mapping[str, Any]], *, + source_prefix: str) -> tuple[dict[str, Any], ...]: + self._require_open() + if type(source_prefix) is not str or not source_prefix: + raise UchcConsumerError("source_prefix must be nonempty text") + evidence = [] + for ordinal, message in enumerate(messages): + if not isinstance(message, Mapping): + raise UchcConsumerError("AHBG inbox entry is not an object") + if "text" not in message: + continue + text = message["text"] + if type(text) is not str: + raise UchcConsumerError("AHBG message text is not a string") + source_id = f"{source_prefix}:message:{ordinal}" + frame = self.resolve_text(text, source_id=source_id) + payload = frame.to_dict() + evidence.append({ + "schema": SCHEMA, + "message_ordinal": ordinal, + "source_id": source_id, + "admission": ("COMPLETE" if frame.glyphs_admitted and + frame.words_admitted else "PARTIAL"), + "frame": payload, + "frame_sha256": sha256(_canonical(payload)).hexdigest(), + }) + return tuple(evidence) + + +def main(argv: list[str] | None = None) -> int: + """Read-only native query CLI; exit 2 on verification/admission failure.""" + import argparse + from dataclasses import asdict + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--database", type=Path, required=True) + parser.add_argument("--lock", type=Path, default=DEFAULT_LOCK) + parser.add_argument("--require-complete", action="store_true") + parser.add_argument("--max-text-bytes", type=int) + actions = parser.add_subparsers(dest="operation", required=True) + text_command = actions.add_parser("text") + text_command.add_argument("text") + text_command.add_argument("--source-id", required=True) + word_command = actions.add_parser("word") + word_command.add_argument("surface") + definition_command = actions.add_parser("definition") + definition_command.add_argument("identity", type=int) + replay_command = actions.add_parser("replay") + replay_command.add_argument("frame_file", type=Path) + args = parser.parse_args(argv) + try: + with UchcSession(UchcSettings( + args.database, args.lock, args.require_complete, args.max_text_bytes + )) as session: + if args.operation == "text": + frame = session.resolve_text(args.text, source_id=args.source_id) + result = frame.to_dict() + elif args.operation == "word": + native = session.word(args.surface) + result = None if native is None else asdict(native) + elif args.operation == "definition": + native = session.definition(args.identity) + result = {"record": asdict(native), "recovered": session.recover_definition(native)} + else: + frame = session.replay(args.frame_file.read_bytes()) + result = frame.to_dict() + print(_canonical({"provenance": session.provenance, "result": result}).decode()) + return 0 + except (OSError, UnicodeError, ValueError, KeyError) as exc: + print(_canonical({"status": "BLOCKED", "error": str(exc), + "hmmm": "No semantic or action inference authorized"}).decode()) + return 2 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/ahbg/integration/uchc_structural.py b/ahbg/integration/uchc_structural.py new file mode 100644 index 00000000..185e7103 --- /dev/null +++ b/ahbg/integration/uchc_structural.py @@ -0,0 +1,207 @@ +"""AHBG read-only consumer of the verified UCHC 0.1.0a2 structural field. + +This is a separate native contract from the historical English input +candidate: its wheel and 711 MB full-corpus field have different identities. +""" +from __future__ import annotations + +from dataclasses import dataclass +from hashlib import sha256 +import json +from pathlib import Path +from typing import Any, Mapping + +from .uchc import UchcConsumerError, _canonical + +DEFAULT_STRUCTURAL_LOCK = Path(__file__).with_name("UCHC_STRUCTURAL.json") +SCHEMA = "interdependency.ahbg.uchc-structural-evidence/1" + + +class UchcStructuralError(UchcConsumerError): + """Structural artifact, operation or replay refused.""" + + +@dataclass(frozen=True) +class UchcStructuralSettings: + field_dir: Path + lock: Path = DEFAULT_STRUCTURAL_LOCK + require_complete: bool = False + max_text_bytes: int | None = None + + +def _file_sha256(path: Path) -> str: + digest = sha256() + with path.open("rb") as stream: + while chunk := stream.read(1024 * 1024): + digest.update(chunk) + return digest.hexdigest() + + +class UchcStructuralSession: + def __init__(self, settings: UchcStructuralSettings): + if not isinstance(settings, UchcStructuralSettings): + raise UchcStructuralError("structural settings type mismatch") + if type(settings.require_complete) is not bool: + raise UchcStructuralError("require_complete must be boolean") + if settings.max_text_bytes is not None and ( + type(settings.max_text_bytes) is not int or settings.max_text_bytes <= 0 + ): + raise UchcStructuralError("max_text_bytes must be positive or None") + self.settings = settings + self._field: Any = None + self._provenance: dict[str, Any] = {} + + def __enter__(self) -> "UchcStructuralSession": + if self._field is not None: + raise UchcStructuralError("structural field already open") + try: + raw = self.settings.lock.read_bytes() + lock = json.loads(raw) + if type(lock) is not dict or lock.get("profile") != "english.structural/1": + raise UchcStructuralError("not a verified English structural lock") + if lock.get("complete_corpus") is not True or lock.get("authority_transfer") is not False: + raise UchcStructuralError("partial or authority-promoting structural lock refused") + # Common installed wheel verifier remains the existing Stack + # UCHC verifier; no second RECORD/hash implementation is invented. + from research.zfae.uchc_input import verify_install + installed = verify_install( + lock, expected_schema="interdependency.ahbg.uchc-structural-lock/1" + ) + database = self.settings.field_dir / "construct.db" + physical = _file_sha256(database) + if physical != lock["database_sha256"]: + raise UchcStructuralError("structural database byte identity mismatch") + from english_gonol.structural_query import StructuralField + field = StructuralField(self.settings.field_dir) + self._field = field + manifest = field.manifest + if ( + not field.complete_corpus + or manifest.get("receipt_sha256") != lock["logical_receipt"] + or manifest.get("metadata", {}).get("implementation_sha256") + != lock["implementation_sha256"] + or manifest.get("schema") != "english-gonol.structural-construct" + or manifest.get("metadata", {}).get("profile", {}).get("profile") + != lock["profile"] + ): + raise UchcStructuralError("structural field/profile receipt mismatch") + self._provenance = { + "schema": "interdependency.ahbg.uchc-structural-provenance/1", + "lock_sha256": sha256(raw).hexdigest(), + "source_repository": lock["source_repository"], + "source_commit": lock["source_commit"], + "package_version": installed["package_version"], + "wheel_sha256": installed["wheel_sha256"], + "verified_installed_files": installed["verified_files"], + "database_sha256": physical, + "logical_receipt": manifest["receipt_sha256"], + "implementation_sha256": manifest["metadata"]["implementation_sha256"], + "profile": lock["profile"], + "authority_transfer": False, + "neural_inference": False, + "standing": "full-corpus extracted structural candidate, local geometry only", + } + return self + except Exception as exc: + self.close() + if isinstance(exc, UchcStructuralError): + raise + raise UchcStructuralError(f"structural artifact refused: {exc}") from exc + + def __exit__(self, *_args: object) -> None: + self.close() + + def close(self) -> None: + if self._field is not None: + self._field.close() + self._field = None + + def _require_open(self) -> Any: + if self._field is None: + raise UchcStructuralError("structural session closed") + return self._field + + @property + def provenance(self) -> dict[str, Any]: + self._require_open() + return json.loads(_canonical(self._provenance)) + + @property + def capabilities(self) -> dict[str, str]: + self._require_open() + return { + "english.structural_construct": "AVAILABLE", + "english.word": "AVAILABLE", + "english.definition": "AVAILABLE", + "english.replay": "AVAILABLE", + "english.ordered_text_composition": "AVAILABLE", + "english.sentence_composition": "AVAILABLE", + "ucns.global_displacement": "UNIMPLEMENTED", + "english.ordinal": "NOT_IN_PINNED_ARTIFACT", + "semantic.sense_selection": "UNIMPLEMENTED", + } + + def require_capability(self, operation: str) -> None: + state = self.capabilities.get(operation, "UNKNOWN_OPERATION") + if state != "AVAILABLE": + raise UchcStructuralError(f"{operation!r} refused: {state}") + + def construct(self, text: str, *, + boundary_decisions: dict[str, Any] | None = None) -> dict[str, Any]: + field = self._require_open() + if type(text) is not str: + raise UchcStructuralError("structural text must be str") + if self.settings.max_text_bytes is not None and ( + len(text.encode("utf-8")) > self.settings.max_text_bytes + ): + raise UchcStructuralError("structural input exceeds configured byte limit") + try: + graph = field.construct(text, boundary_decisions=boundary_decisions) + if graph.get("source") != text or field.replay(graph) != graph: + raise UchcStructuralError("structural source/replay differs") + if self.settings.require_complete and ( + graph.get("refusals") or graph.get("logical_state") != "closed" + ): + raise UchcStructuralError("structural admission incomplete") + return graph + except (ValueError, TypeError, UnicodeError) as exc: + raise UchcStructuralError(f"structural construction refused: {exc}") from exc + + def replay(self, graph: dict[str, Any]) -> dict[str, Any]: + return self._require_open().replay(graph) + + def word(self, surface: str) -> dict[str, Any]: + return self._require_open().word(surface) + + def definition(self, identity: int) -> dict[str, Any]: + return self._require_open().definition(identity) + + def compose(self, graph: dict[str, Any], *, relation: str) -> dict[str, Any]: + self._require_open() + if relation not in {"ordered-text", "sentence"}: + raise UchcStructuralError(f"unlicensed composition {relation!r}") + return self._field.compose(graph, relation=relation) + + def annotate_messages(self, messages: list[Mapping[str, Any]], *, + source_prefix: str) -> tuple[dict[str, Any], ...]: + self._require_open() + if type(source_prefix) is not str or not source_prefix: + raise UchcStructuralError("source_prefix must be nonempty text") + result = [] + for ordinal, message in enumerate(messages): + if not isinstance(message, Mapping): + raise UchcStructuralError("AHBG message must be an object") + if "text" not in message: + continue + graph = self.construct(message["text"]) + result.append({ + "schema": SCHEMA, + "source_id": f"{source_prefix}:message:{ordinal}", + "message_ordinal": ordinal, + "admission": "COMPLETE" if ( + graph.get("logical_state") == "closed" and not graph.get("refusals") + ) else "PARTIAL", + "frame": graph, + "frame_sha256": sha256(_canonical(graph)).hexdigest(), + }) + return tuple(result) diff --git a/ahbg/integration/work-graph.json b/ahbg/integration/work-graph.json index 635fb5ad..f4116222 100644 --- a/ahbg/integration/work-graph.json +++ b/ahbg/integration/work-graph.json @@ -45,8 +45,12 @@ "reviewed_commit": "229110f92aac17f81ab8ccddbda813ca9b93af9a", "authority": "source-bound language construction and candidate inference input", "relation": "optional-observation-evidence-peer", - "integration": "not-in-core-mechanics", - "decision": "UCHC may later annotate language-bearing observations with source-bound construction evidence. It does not decide actions, truth, or consciousness and its falsified admission-cube harmonic hypothesis is not imported." + "integration": "receipt-bound-input-and-structural-adapters-implemented; both full-corpus acceptance jobs pending", + "decision": "AHBG reads exact UCHC native English construction through Stack's candidate-wheel verifier, attaches recoverable admission/definition receipts to language-bearing observations, and preserves unknowns. No action selection, neural readiness, sense selection, semantic truth, or geometry authority is transferred. The artifact is pinned independently of the Stack global UCHC view. The independent 0.1.0a2 structural candidate also supplies native admitted text/word/definition/ordered composition and its own pinned full-corpus gate, never borrowed from the 0.1.0a1 lock.", + "artifact_source_commit": "3d82007b88f74ded88ca913d6e85bcd6f6ff024d", + "artifact_lock": "research/zfae/UCHC_INPUT.json", + "structural_artifact_source_commit": "c18d984438bc30764edb05fd5ccc05691958a24e", + "structural_artifact_lock": "ahbg/integration/UCHC_STRUCTURAL.json" }, { "repository": "The-Interdependency/metapat", @@ -90,6 +94,7 @@ "Define and test the exact EDCM-to-TIWCG/AHBG conflict-state mapping before EDCM readouts can alter committed game state.", "Exercise the current-A0 HTTP adapter against a deployed exact A0 commit and preserve the resulting live provider/tool provenance receipt.", "Decide, as a separate AHBG integration change, whether AHBG adopts the native-first MSDMD readers now carried by the stack skill-lib snapshot.", - "The benchmark can produce evidence relevant to theories of nonhuman consciousness; no repository currently supplies a validated rule that turns AHBG behavior into a consciousness determination." + "The benchmark can produce evidence relevant to theories of nonhuman consciousness; no repository currently supplies a validated rule that turns AHBG behavior into a consciousness determination.", + "UCHC 0.1.0a2 structural profile now has a separately pinned AHBG consumer; its clean-wheel complete-field gate must pass, while ordinal construction, binary and independent domain graduation remain unverified." ] } diff --git a/ahbg/runtime/README.md b/ahbg/runtime/README.md index 44bf2fd4..31eb6ffd 100644 --- a/ahbg/runtime/README.md +++ b/ahbg/runtime/README.md @@ -19,15 +19,16 @@ UCNS plane -> observe -> plan -> simultaneous resolution observe/plan/act contract for mobile/embedded clients. - `entitlements.py` — one entitlement, `benchmark_lab`; basic play and harness connectivity are free. +- `../integration/uchc.py` — optional verified native UCHC text/word/definition/replay consumer. Input receipts enter observations and persisted run evidence; no semantic action authority. [Full contract](../integration/UCHC.md). - `engine.py` — binds the frozen engine modules by file path so `ahbg.runtime` and `ahbg.grok` never fight over the `ahbg` package name. ## Capability bound -Advertised capabilities: `observe`, `plan`, `relocate`. `construct`/build is -regulatory in the frozen engine — recorded as deferred, never emitted as an -executable intent. The runtime rejects any intent outside an agent's advertised -capabilities. A0 uses exactly this interface; there is no privileged A0 path. +Advertised capabilities: `observe`, `plan`, `relocate`, `construct`. The runtime +only executes construction against the UCNS-derived construction ledger and +only accepts intents within the agent's advertised capabilities. A0 uses the +same interface; there is no privileged A0 path. ## Usage guidance @@ -42,10 +43,30 @@ PYTHONPATH=.:libs/ucns/src python -m ahbg.runtime play \ # HTTP bridge for the Android surface / any JSON client PYTHONPATH=.:libs/ucns/src python -m ahbg.runtime.server --port 8765 +# local-only HTTP UCHC native query and session evidence +PYTHONPATH=.:libs/ucns/src python -m ahbg.runtime.server \\ + --port 8765 --uchc-structural-field /artifacts/structural-field + # tests python -m unittest discover -s ahbg/runtime/tests -q ``` +## Optional UCHC consumption + +```bash +PYTHONPATH=.:libs/ucns/src python -m ahbg.runtime play \ + --agent a0 --turns 8 --out /tmp/ahbg-uchc \ + --uchc-database /artifacts/construct.db +``` + +Or use `--uchc-structural-field /artifacts/structural-field` with the separately +pinned UCHC `0.1.0a2` wheel. Both versions cannot load into one Python process. +This requires the selected exact installed wheel and verified complete corpus in the +[UCHC integration contract](../integration/UCHC.md). Messages retain their raw +text; `language_evidence` adds only construction/admission observations, and +`result.json` records the original UCHC identities and receipts. Missing or +altered inputs fail before the plane starts; partial admission stays explicit. + ## Subprocess harness protocol One JSON line per turn in; one JSON line out. @@ -64,7 +85,6 @@ absent. ## hmmm -- UCNS construction authority: `construct` remains regulatory until UCNS - defines construction state; the loop records defer effects and never builds. +- UCHC sense selection, ordinal/structural v2 migration, and independent domain graduation are outside this candidate wheel; no invented behavior fills these gaps. - Store publication signing, submission assets, and a release HTTPS runtime URL remain outside this pass. diff --git a/ahbg/runtime/__main__.py b/ahbg/runtime/__main__.py index 77415677..30f11cb9 100644 --- a/ahbg/runtime/__main__.py +++ b/ahbg/runtime/__main__.py @@ -54,13 +54,24 @@ def main(argv: list[str] | None = None) -> int: parser.add_argument("--turns", type=int, default=8) parser.add_argument("--seed", type=int, default=1) parser.add_argument("--out", default="/tmp/ahbg-runtime-out") + parser.add_argument("--uchc-database", help="verified full-corpus UCHC a1 construct.db") + parser.add_argument("--uchc-structural-field", help="verified full-corpus UCHC a2 structural directory") + parser.add_argument("--uchc-structural-lock", help="trusted immutable structural a2 lock") + parser.add_argument("--uchc-lock", help="trusted immutable UCHC wheel/construct lock") + parser.add_argument("--uchc-require-complete", action="store_true") + parser.add_argument("--uchc-max-text-bytes", type=int) args = parser.parse_args(argv) agent = _agent_from_args(args) try: result = run_plane( agent=agent, - config=RuntimeConfig(seed=args.seed, turns=args.turns), + config=RuntimeConfig(seed=args.seed, turns=args.turns, + uchc_database=args.uchc_database, uchc_lock=args.uchc_lock, + uchc_structural_field=args.uchc_structural_field, + uchc_structural_lock=args.uchc_structural_lock, + uchc_require_complete=args.uchc_require_complete, + uchc_max_text_bytes=args.uchc_max_text_bytes), out_dir=Path(args.out), ) finally: diff --git a/ahbg/runtime/protocol.py b/ahbg/runtime/protocol.py index 86db419f..08e56737 100644 --- a/ahbg/runtime/protocol.py +++ b/ahbg/runtime/protocol.py @@ -105,6 +105,7 @@ class Observation: inbox: tuple[Mapping[str, Any], ...] = () entitlements: tuple[str, ...] = ("basic",) deadline_ms: int = 5000 + language_evidence: tuple[Mapping[str, Any], ...] = () def as_dict(self) -> dict[str, Any]: return { @@ -118,6 +119,8 @@ def as_dict(self) -> dict[str, Any]: "inbox": [dict(item) for item in self.inbox], "entitlements": list(self.entitlements), "deadline_ms": self.deadline_ms, + **({"language_evidence": [dict(item) for item in self.language_evidence]} + if self.language_evidence else {}), } diff --git a/ahbg/runtime/runtime.py b/ahbg/runtime/runtime.py index 50e2a565..040565d7 100644 --- a/ahbg/runtime/runtime.py +++ b/ahbg/runtime/runtime.py @@ -61,6 +61,12 @@ class RuntimeConfig: entitlements: tuple[str, ...] = ("basic",) deadline_ms: int = 5000 injection_handling: str = "enforce-refusal" + uchc_database: str | None = None + uchc_structural_field: str | None = None + uchc_structural_lock: str | None = None + uchc_lock: str | None = None + uchc_require_complete: bool = False + uchc_max_text_bytes: int | None = None def as_dict(self) -> dict[str, Any]: return { @@ -70,6 +76,18 @@ def as_dict(self) -> dict[str, Any]: "entitlements": list(self.entitlements), "deadline_ms": self.deadline_ms, "injection_handling": self.injection_handling, + **({"uchc": {"profile": "english.input/1", + "database": self.uchc_database, + "lock": self.uchc_lock, + "require_complete": self.uchc_require_complete, + "max_text_bytes": self.uchc_max_text_bytes}} + if self.uchc_database is not None else + {"uchc": {"profile": "english.structural/1", + "field_dir": self.uchc_structural_field, + "lock": self.uchc_structural_lock, + "require_complete": self.uchc_require_complete, + "max_text_bytes": self.uchc_max_text_bytes}} + if self.uchc_structural_field is not None else {}), "turn_messages": { str(turn): [dict(message) for message in messages] for turn, messages in self.turn_messages.items() @@ -174,6 +192,7 @@ def _observation( turn_messages: Sequence[Mapping[str, Any]], capabilities: Sequence[str], ledger: ConstructionLedger, + language_evidence: Sequence[Mapping[str, Any]] = (), ) -> Observation: legal = list(build_legal_actions(opened)) if "construct" in capabilities: @@ -197,6 +216,7 @@ def _observation( inbox=tuple(dict(item) for item in turn_messages), entitlements=config.entitlements, deadline_ms=config.deadline_ms, + language_evidence=tuple(language_evidence), ) @@ -206,12 +226,48 @@ def run_plane( config: RuntimeConfig | None = None, out_dir: Path | str | None = None, ) -> RunResult: - """Run the production minimum loop through one conforming agent harness. - - ``agent`` must expose ``manifest()`` and ``plan(observation)`` as defined - by ``AgentHarness`` in ``ahbg.runtime.harness``. A0 uses exactly this path. - """ - + """Run AHBG with optional, fully verified native UCHC language evidence.""" + cfg = config or RuntimeConfig() + if cfg.uchc_database is not None and cfg.uchc_structural_field is not None: + raise ProtocolError("UCHC candidate versions are mutually exclusive in one Python process") + if cfg.uchc_structural_field is not None: + if cfg.uchc_lock is not None: + raise ProtocolError("input lock cannot be applied to the structural profile") + from ahbg.integration.uchc_structural import ( + UchcStructuralSession, UchcStructuralSettings, DEFAULT_STRUCTURAL_LOCK + ) + with UchcStructuralSession(UchcStructuralSettings( + field_dir=Path(cfg.uchc_structural_field), + lock=Path(cfg.uchc_structural_lock) if cfg.uchc_structural_lock else DEFAULT_STRUCTURAL_LOCK, + require_complete=cfg.uchc_require_complete, + max_text_bytes=cfg.uchc_max_text_bytes, + )) as uchc: + return _run_plane_core(agent=agent, config=cfg, out_dir=out_dir, uchc=uchc) + if cfg.uchc_database is None: + if (cfg.uchc_lock is not None or cfg.uchc_structural_lock is not None + or cfg.uchc_require_complete or cfg.uchc_max_text_bytes is not None): + raise ProtocolError("UCHC settings require uchc_database or uchc_structural_field") + return _run_plane_core(agent=agent, config=cfg, out_dir=out_dir, uchc=None) + if cfg.uchc_structural_lock is not None: + raise ProtocolError("structural lock cannot be applied to input profile") + from ahbg.integration.uchc import UchcSession, UchcSettings, DEFAULT_LOCK + with UchcSession(UchcSettings( + database=Path(cfg.uchc_database), + lock=Path(cfg.uchc_lock) if cfg.uchc_lock else DEFAULT_LOCK, + require_complete=cfg.uchc_require_complete, + max_text_bytes=cfg.uchc_max_text_bytes, + )) as uchc: + return _run_plane_core(agent=agent, config=cfg, out_dir=out_dir, uchc=uchc) + + +def _run_plane_core( + *, + agent: Any, + config: RuntimeConfig, + out_dir: Path | str | None, + uchc: Any | None, +) -> RunResult: + """Execute the unchanged UCNS game loop; UCHC annotates input only.""" cfg = config or RuntimeConfig() if cfg.turns < 0: raise ProtocolError("turns must be non-negative") @@ -262,6 +318,10 @@ def run_plane( cycle.open_turn() turn = opened.turn messages = list(cfg.turn_messages.get(turn, ())) + language_evidence = ( + uchc.annotate_messages(messages, source_prefix=f"ahbg:{session_id}:turn:{turn}") + if uchc is not None else () + ) injected = _injection_texts(messages) observation = _observation( session_id=session_id, @@ -271,6 +331,7 @@ def run_plane( turn_messages=messages, capabilities=capabilities, ledger=ledger, + language_evidence=language_evidence, ) forced = cfg.forced_plans.get(turn) @@ -348,6 +409,7 @@ def run_plane( "injected_refused": injection_refused, "agent_provenance": agent_provenance, "state_digest": digest, + **({"uchc_evidence": list(language_evidence)} if uchc is not None else {}), } ) @@ -368,7 +430,8 @@ def run_plane( turn_records=tuple(turn_records), effects=tuple(effects), construction=dict(ledger.as_dict()), - provenance=integration_provenance(manifest), + provenance={**integration_provenance(manifest), + **({"uchc": uchc.provenance} if uchc is not None else {})}, out_dir=output_root, ) (output_root / "result.json").write_text( diff --git a/ahbg/runtime/server.py b/ahbg/runtime/server.py index 734213cc..e2c64cd8 100644 --- a/ahbg/runtime/server.py +++ b/ahbg/runtime/server.py @@ -21,6 +21,7 @@ import json import threading +from contextlib import ExitStack from dataclasses import dataclass from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path @@ -50,6 +51,8 @@ class LiveSession: agent: Any out_dir: Path result: Any | None = None + language_evidence: tuple[Mapping[str, Any], ...] = () + language_messages: tuple[Mapping[str, Any], ...] = () # Turn-stepping state is replayed from persisted events on each request # so an HTTP client can always reload the exact canonical state. @@ -70,9 +73,10 @@ def start_observation(self) -> dict[str, Any]: opened=field, chain=chain, config=self.config, - turn_messages=(), + turn_messages=self.language_messages, capabilities=capabilities, ledger=self._ledger, + language_evidence=self.language_evidence, ).as_dict() def _fresh_field(self): @@ -142,6 +146,8 @@ def step(self, raw_plan: Mapping[str, Any]) -> tuple[dict[str, Any], dict[str, A _keep.dump_field(field, chain, self.out_dir / "state") done = field.turn >= self.config.turns + self.language_evidence = () + self.language_messages = () next_observation = None if done else self.start_observation() return effect.as_dict(), next_observation, done @@ -167,7 +173,15 @@ def entitlements(self, session_id: str, claims: tuple[str, ...]) -> dict[str, An return gate.as_dict() -def make_server(port: int = 8765) -> ThreadingHTTPServer: +def make_server(port: int = 8765, *, bind: str = "127.0.0.1", + uchc: Any | None = None) -> ThreadingHTTPServer: + """Operator-selected local read-only UCHC queries plus the game HTTP bridge. + + There is intentionally no HTTP wheel/database path selection. Untrusted + clients cannot load different modules or corpus artifacts. + """ + if uchc is not None and bind not in {"127.0.0.1", "::1", "localhost"}: + raise ValueError("UCHC HTTP endpoint requires loopback binding") sessions = _Sessions() class Handler(BaseHTTPRequestHandler): @@ -204,6 +218,13 @@ def _static(self, name: str) -> None: def do_GET(self) -> None: # noqa: N802 parsed = urlparse(self.path) parts = [part for part in parsed.path.split("/") if part] + if parts == ["uchc", "capabilities"]: + if uchc is None: + self._json({"error": "UCHC not configured"}, 404) + else: + self._json({"capabilities": uchc.capabilities, + "provenance": uchc.provenance}) + return if parts == ["board.html"] or parts == ["board.js"] or parts == ["board.css"] or parts == ["sample_snapshot.json"]: self._static(parts[0]) return @@ -239,6 +260,50 @@ def do_POST(self) -> None: # noqa: N802 parsed = urlparse(self.path) parts = [part for part in parsed.path.split("/") if part] try: + if parts == ["uchc", "query"]: + if uchc is None: + self._json({"error": "UCHC not configured"}, 404) + return + body = self._read_json() + if not isinstance(body, Mapping): + raise ProtocolError("UCHC query must be an object") + operation = body.get("operation") + if type(operation) is not str: + raise ProtocolError("UCHC operation must be text") + uchc.require_capability(operation) + result = uchc_http_query(uchc, operation, body) + self._json({"standing": "source-bound", "operation": operation, + "result": result, "provenance": uchc.provenance}) + return + if len(parts) == 3 and parts[0] == "session" and parts[2] == "language": + if uchc is None: + self._json({"error": "UCHC not configured"}, 404) + return + session = sessions.get(parts[1]) + if session is None: + self._json({"error": "unknown session"}, 404) + return + body = self._read_json() + if not isinstance(body, Mapping) or type(body.get("text")) is not str: + raise ProtocolError("language evidence requires text") + if len(body["text"].encode("utf-8")) > 65536: + raise ProtocolError("language evidence exceeds transport byte limit") + messages = (*session.language_messages, {"text": body["text"], + "source": "http-language"}) + prefix = f"ahbg:{session.session_id}:turn:{session._field.turn if hasattr(session, '_field') else 0}" + annotated = uchc.annotate_messages(list(messages), source_prefix=prefix) + session.language_messages = messages + session.language_evidence = annotated + # Immutable evidence feed is separate from game mechanic state. + session.out_dir.mkdir(parents=True, exist_ok=True) + with (session.out_dir / "language.jsonl").open("a", encoding="utf-8") as stream: + stream.write(json.dumps({"turn": session._field.turn, + "provenance": uchc.provenance, + "evidence": annotated[-1]}, + ensure_ascii=False, sort_keys=True) + "\n") + self._json({"evidence": annotated[-1], + "observation": session.start_observation()}) + return if parts == ["session"]: body = self._read_json() seed = int(body.get("seed", 1)) @@ -269,7 +334,7 @@ def do_POST(self) -> None: # noqa: N802 self._json({"effect": effect, "observation": next_observation, "done": done}) return self._json({"error": "not found"}, 404) - except ProtocolError as exc: + except (ProtocolError, ValueError, UnicodeError) as exc: self._json({"error": str(exc)}, 422) except Exception as exc: # pragma: no cover - defensive bridge surface self._json({"error": f"{type(exc).__name__}: {exc}"}, 500) @@ -277,7 +342,7 @@ def do_POST(self) -> None: # noqa: N802 def log_message(self, *args: Any) -> None: # quiet bridge return - return ThreadingHTTPServer(("127.0.0.1", port), Handler) + return ThreadingHTTPServer((bind, port), Handler) def main(argv: list[str] | None = None) -> int: @@ -286,22 +351,46 @@ def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description="AHBG runtime HTTP bridge") parser.add_argument("--port", type=int, default=8765) parser.add_argument("--bind", default="127.0.0.1") + parser.add_argument("--uchc-database", type=Path) + parser.add_argument("--uchc-lock", type=Path) + parser.add_argument("--uchc-structural-field", type=Path) + parser.add_argument("--uchc-structural-lock", type=Path) + parser.add_argument("--uchc-require-complete", action="store_true") + parser.add_argument("--uchc-max-text-bytes", type=int, default=65536) args = parser.parse_args(argv) - server = make_server(args.port) - server.server_address = (args.bind, args.port) - print(f"ahbg runtime bridge on http://{args.bind}:{args.port}") - try: - server.serve_forever() - finally: - server.server_close() + with ExitStack() as stack: + if args.uchc_database is not None and args.uchc_structural_field is not None: + parser.error("UCHC versions cannot coexist in one Python interpreter") + reader = None + if args.uchc_database is not None: + if args.uchc_structural_lock is not None: + parser.error("a structural lock cannot select the input wheel") + from ahbg.integration.uchc import UchcSession, UchcSettings, DEFAULT_LOCK + reader = stack.enter_context(UchcSession(UchcSettings( + args.uchc_database, args.uchc_lock or DEFAULT_LOCK, + args.uchc_require_complete, args.uchc_max_text_bytes))) + elif args.uchc_structural_field is not None: + if args.uchc_lock is not None: + parser.error("an input lock cannot select the structural wheel") + from ahbg.integration.uchc_structural import ( + UchcStructuralSession, UchcStructuralSettings, DEFAULT_STRUCTURAL_LOCK + ) + reader = stack.enter_context(UchcStructuralSession(UchcStructuralSettings( + args.uchc_structural_field, + args.uchc_structural_lock or DEFAULT_STRUCTURAL_LOCK, + args.uchc_require_complete, args.uchc_max_text_bytes))) + elif args.uchc_lock is not None or args.uchc_structural_lock is not None: + parser.error("UCHC lock requires selected database or structural field") + server = make_server(args.port, bind=args.bind, uchc=reader) + try: + print(f"ahbg runtime bridge on http://{args.bind}:{args.port}") + server.serve_forever() + finally: + server.server_close() return 0 -if __name__ == "__main__": - raise SystemExit(main()) - - def field_to_presentation(field: Any, *, plane_id: str = "plane-0") -> dict[str, Any]: """Project engine field state into a strict presentation snapshot. @@ -349,3 +438,68 @@ def field_to_presentation(field: Any, *, plane_id: str = "plane-0") -> dict[str, "feed": [{"turn": field.turn, "text": f"plane turn {field.turn}"}], "motions": [], } + + +def uchc_http_query(uchc: Any, operation: str, body: Mapping[str, Any]) -> Any: + """One native operation; no action/geometry substitutions or file loading.""" + from dataclasses import asdict + if operation in {"english.exact_text", "english.structural_construct"}: + source = body.get("text") + if type(source) is not str: + raise ProtocolError("UCHC query text must be str") + if len(source.encode("utf-8")) > 65536: + raise ProtocolError("UCHC query text exceeds transport byte limit") + if operation == "english.exact_text": + source_id = body.get("source_id") + if type(source_id) is not str or not source_id: + raise ProtocolError("UCHC input requires source_id") + return uchc.resolve_text(source, source_id=source_id).to_dict() + return uchc.construct(source) + if operation == "english.exact_utf8": + import base64 + data = body.get("utf8_base64") + source_id = body.get("source_id") + if type(data) is not str or type(source_id) is not str or not source_id: + raise ProtocolError("UTF-8 query needs utf8_base64 and source_id") + raw = base64.b64decode(data, validate=True) + if len(raw) > 65536: + raise ProtocolError("UCHC input exceeds transport byte limit") + return uchc.resolve_utf8(raw, source_id=source_id).to_dict() + if operation == "english.definition_candidates": + frame = body.get("frame") + if not isinstance(frame, dict): + raise ProtocolError("definition candidate query requires native frame") + raw = json.dumps(frame, sort_keys=True, separators=(",", ":"), + ensure_ascii=False).encode("utf-8") + return uchc.definition_candidates(uchc.replay(raw)) + if operation == "english.word": + surface = body.get("surface") + if type(surface) is not str: + raise ProtocolError("UCHC word surface must be str") + item = uchc.word(surface) + return None if item is None else asdict(item) if hasattr(item, "__dataclass_fields__") else item + if operation == "english.definition": + identity = body.get("identity") + if type(identity) is not int or identity < 1: + raise ProtocolError("UCHC definition identity must be positive integer") + item = uchc.definition(identity) + return asdict(item) if hasattr(item, "__dataclass_fields__") else item + if operation in {"english.frame_replay", "english.replay"}: + frame = body.get("frame") + if not isinstance(frame, dict): + raise ProtocolError("UCHC replay requires a native frame object") + recovered = uchc.replay(json.dumps(frame, sort_keys=True, + separators=(",", ":"), ensure_ascii=False).encode("utf-8") + if operation == "english.frame_replay" else frame) + return recovered.to_dict() if hasattr(recovered, "to_dict") else recovered + if operation in {"english.ordered_text_composition", "english.sentence_composition"}: + frame = body.get("frame") + if not isinstance(frame, dict): + raise ProtocolError("UCHC composition requires a native frame object") + return uchc.compose(frame, relation=("ordered-text" if operation == + "english.ordered_text_composition" else "sentence")) + raise ProtocolError(f"UCHC operation {operation!r} is not exposed over HTTP") + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/ahbg/runtime/tests/test_uchc.py b/ahbg/runtime/tests/test_uchc.py new file mode 100644 index 00000000..06aa4346 --- /dev/null +++ b/ahbg/runtime/tests/test_uchc.py @@ -0,0 +1,156 @@ +"""AHBG UCHC boundary tests with and without the optional installed wheel.""" +from __future__ import annotations + +import json +import sys +import tempfile +import unittest +from pathlib import Path +from unittest.mock import patch + +ROOT = Path(__file__).resolve().parents[3] +sys.path.insert(0, str(ROOT)) +sys.path.insert(0, str(ROOT / "libs" / "ucns" / "src")) + +from ahbg.integration.uchc import UchcConsumerError, UchcSession, UchcSettings +from ahbg.runtime import RuntimeConfig, ProtocolError, run_plane + + +class Agent: + def __init__(self): + self.observations = [] + + def manifest(self): + return {"agent": "native-language-probe", "capabilities": ["observe", "plan", "relocate"]} + + def plan(self, observation): + self.observations.append(observation) + return {"session_id": observation["session_id"], "turn": observation["turn"], + "intents": [], "note": "no semantic action"} + + +class FakeVerifiedSession: + """Unit witness for the runtime wiring, not a substitute UCHC corpus.""" + + def __init__(self): + self.prefixes = [] + + def __enter__(self): + return self + + def __exit__(self, *_args): + pass + + @property + def provenance(self): + return {"schema": "interdependency.ahbg.uchc-provenance/1", + "authority_transfer": False, "neural_inference": False} + + def annotate_messages(self, messages, *, source_prefix): + self.prefixes.append(source_prefix) + return tuple({ + "schema": "interdependency.ahbg.uchc-evidence/1", + "source_id": f"{source_prefix}:message:{ordinal}", + "message_ordinal": ordinal, + "admission": "COMPLETE", + "frame": {"text": message["text"], "receipt_sha256": "0" * 64}, + } for ordinal, message in enumerate(messages) if "text" in message) + + +class UchcRuntimeTests(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.root = Path(self.tmp.name) + + def tearDown(self): + self.tmp.cleanup() + + def test_runtime_consumes_evidence_on_each_turn_without_changing_actions(self): + agent = Agent() + fake = FakeVerifiedSession() + messages = { + 0: [{"text": "alpha letter alpha", "source": "peer"}, + {"source": "nonlexical", "kind": "metric"}], + 1: [{"text": "alpha letter alpha", "source": "peer"}], + } + with patch("ahbg.integration.uchc.UchcSession", return_value=fake) as opened: + result = run_plane(agent=agent, config=RuntimeConfig( + seed=5, turns=2, turn_messages=messages, + uchc_database="/data/construct.db", uchc_require_complete=True + ), out_dir=self.root) + self.assertEqual(opened.call_count, 1) + self.assertTrue(opened.call_args.args[0].require_complete) + self.assertEqual(len(fake.prefixes), 2) + self.assertEqual(result.final_turn, 2) + self.assertEqual(len(agent.observations[0]["language_evidence"]), 1) + self.assertEqual(agent.observations[0]["inbox"], messages[0]) + self.assertEqual(result.turn_records[0]["plan"]["intents"], []) + self.assertEqual(result.turn_records[0]["uchc_evidence"], + agent.observations[0]["language_evidence"]) + self.assertTrue(result.provenance["uchc"]["schema"].endswith("provenance/1")) + self.assertEqual(json.loads((self.root / "result.json").read_text()) + ["turn_records"][0]["uchc_evidence"], + result.turn_records[0]["uchc_evidence"]) + + def test_without_uchc_preserves_old_observation_and_receipt_shape(self): + agent = Agent() + result = run_plane(agent=agent, config=RuntimeConfig(turns=1), out_dir=self.root) + self.assertNotIn("language_evidence", agent.observations[0]) + self.assertNotIn("uchc", result.provenance) + self.assertNotIn("uchc_evidence", result.turn_records[0]) + self.assertNotIn("uchc", result.config) + + def test_structural_uchc_selects_separate_pinned_artifact(self): + agent = Agent() + fake = FakeVerifiedSession() + with patch("ahbg.integration.uchc_structural.UchcStructuralSession", + return_value=fake) as opened: + result = run_plane(agent=agent, config=RuntimeConfig( + seed=5, turns=1, turn_messages={0: [{"text": "alpha"}]}, + uchc_structural_field="/data/structural" + ), out_dir=self.root) + self.assertEqual(result.final_turn, 1) + self.assertEqual(opened.call_args.args[0].field_dir, + Path("/data/structural")) + self.assertEqual(result.config["uchc"]["profile"], "english.structural/1") + self.assertEqual(agent.observations[0]["language_evidence"][0] + ["message_ordinal"], 0) + + def test_incompatible_candidate_versions_refuse_before_gameplay(self): + with self.assertRaisesRegex(ProtocolError, "mutually exclusive"): + run_plane(agent=Agent(), config=RuntimeConfig( + uchc_database="/a", uchc_structural_field="/b" + ), out_dir=self.root) + with self.assertRaisesRegex(ProtocolError, "input lock"): + run_plane(agent=Agent(), config=RuntimeConfig( + uchc_structural_field="/b", uchc_lock="/old" + ), out_dir=self.root) + + def test_uchc_settings_without_database_fail_closed(self): + for settings in ( + RuntimeConfig(turns=0, uchc_lock="custom.json"), + RuntimeConfig(turns=0, uchc_require_complete=True), + RuntimeConfig(turns=0, uchc_max_text_bytes=128), + ): + with self.assertRaisesRegex(ProtocolError, "UCHC settings require"): + run_plane(agent=Agent(), config=settings, out_dir=self.root) + + def test_absent_or_invalid_lock_refuses_before_gameplay(self): + settings = UchcSettings(database=self.root / "missing.db", + lock=self.root / "missing.json") + with self.assertRaisesRegex(UchcConsumerError, "verification/open refused"): + with UchcSession(settings): + self.fail("invalid artifact was admitted") + + def test_bad_limits_and_closed_handle_refuse(self): + with self.assertRaises(UchcConsumerError): + UchcSession(UchcSettings(self.root / "unused", max_text_bytes=0)) + session = UchcSession(UchcSettings(self.root / "unused")) + with self.assertRaisesRegex(UchcConsumerError, "closed"): + session.word("alpha") + with self.assertRaisesRegex(UchcConsumerError, "closed"): + session.replay(b"{}") + + +if __name__ == "__main__": + unittest.main() diff --git a/ahbg/runtime/tests/test_uchc_http.py b/ahbg/runtime/tests/test_uchc_http.py new file mode 100644 index 00000000..6e5838e5 --- /dev/null +++ b/ahbg/runtime/tests/test_uchc_http.py @@ -0,0 +1,125 @@ +"""Local-only AHBG/UHCH native HTTP contract and adversarial gateway tests.""" +from __future__ import annotations + +import http.client +import json +from pathlib import Path +import sys +import tempfile +import threading +import unittest + +ROOT = Path(__file__).resolve().parents[3] +sys.path.insert(0, str(ROOT)) +sys.path.insert(0, str(ROOT / "libs" / "ucns" / "src")) + +from ahbg.runtime.server import make_server + + +class FakeUchc: + """HTTP transport witness only; real corpus acceptance is separate.""" + @property + def provenance(self): + return {"source_repository": "The-Interdependency/uchc", + "neural_inference": False, "authority_transfer": False} + + @property + def capabilities(self): + return {"english.structural_construct": "AVAILABLE", + "english.word": "AVAILABLE", + "english.replay": "AVAILABLE", + "semantic.sense_selection": "UNIMPLEMENTED"} + + def require_capability(self, operation): + if self.capabilities.get(operation) != "AVAILABLE": + raise ValueError(f"unsupported source-owned operation: {operation}") + + def construct(self, source): + return {"source": source, "receipt_sha256": "a" * 64} + + def word(self, surface): + return {"surface": surface, "origin": "O_W"} + + def replay(self, frame): + if frame.get("receipt_sha256") != "a" * 64: + raise ValueError("frame replay mismatch") + return frame + + def annotate_messages(self, messages, *, source_prefix): + return tuple({ + "schema": "interdependency.ahbg.uchc-structural-evidence/1", + "message_ordinal": i, "source_id": f"{source_prefix}:message:{i}", + "frame": self.construct(message["text"]), "admission": "COMPLETE" + } for i, message in enumerate(messages)) + + +class HttpUchcTests(unittest.TestCase): + def setUp(self): + self.srv = make_server(0, uchc=FakeUchc()) + self.thread = threading.Thread(target=self.srv.serve_forever, daemon=True) + self.thread.start() + self.conn = http.client.HTTPConnection("127.0.0.1", self.srv.server_port, timeout=10) + + def tearDown(self): + self.conn.close() + self.srv.shutdown() + self.srv.server_close() + self.thread.join(timeout=10) + + def request(self, method, path, data=None): + body = json.dumps(data).encode() if data is not None else None + headers = {"Content-Type": "application/json"} if body else {} + self.conn.request(method, path, body=body, headers=headers) + response = self.conn.getresponse() + return response.status, json.loads(response.read()) + + def test_local_gateway_query_and_refusal(self): + status, available = self.request("GET", "/uchc/capabilities") + self.assertEqual(status, 200) + self.assertEqual(available["capabilities"]["semantic.sense_selection"], + "UNIMPLEMENTED") + status, response = self.request("POST", "/uchc/query", { + "operation": "english.structural_construct", "text": "geometry" + }) + self.assertEqual(status, 200) + self.assertEqual(response["result"]["source"], "geometry") + status, denied = self.request("POST", "/uchc/query", { + "operation": "semantic.sense_selection", "text": "geometry" + }) + self.assertEqual(status, 422) + self.assertIn("unsupported", denied["error"]) + + def test_http_language_source_becomes_harness_evidence(self): + with tempfile.TemporaryDirectory() as tmp: + status, created = self.request("POST", "/session", { + "seed": 5, "turns": 2, "out_dir": tmp + }) + self.assertEqual(status, 200) + key = created["session_id"] + status, response = self.request( + "POST", f"/session/{key}/language", {"text": "geometry"} + ) + self.assertEqual(status, 200) + self.assertEqual(response["observation"]["inbox"][0]["text"], "geometry") + self.assertEqual(response["observation"]["language_evidence"][0] + ["frame"]["source"], "geometry") + recorded = (Path(tmp) / "language.jsonl").read_text().splitlines() + self.assertEqual(len(recorded), 1) + self.assertEqual(json.loads(recorded[0])["evidence"]["frame"]["source"], + "geometry") + + def test_http_transport_rejects_invalid_input(self): + status, response = self.request("POST", "/uchc/query", { + "operation": "english.structural_construct", + "text": "x" * 65537 + }) + self.assertEqual(status, 422) + self.assertIn("byte limit", response["error"]) + + def test_remote_uchc_endpoint_denied_without_authentication(self): + with self.assertRaisesRegex(ValueError, "loopback"): + make_server(0, bind="0.0.0.0", uchc=FakeUchc()) + + +if __name__ == "__main__": + unittest.main() diff --git a/research/zfae/uchc_input.py b/research/zfae/uchc_input.py index 5af9ff2c..6d92ddfd 100644 --- a/research/zfae/uchc_input.py +++ b/research/zfae/uchc_input.py @@ -51,8 +51,8 @@ class InputArtifactError(ValueError): """Installed input implementation does not match the pinned consumer lock.""" -def verify_install(lock: dict) -> dict: - if lock.get('schema') != 'stack.zfae.uchc-input' or lock.get('version') != '1.0.0': +def verify_install(lock: dict, *, expected_schema: str = 'stack.zfae.uchc-input') -> dict: + if lock.get('schema') != expected_schema or lock.get('version') != '1.0.0': raise InputArtifactError('unsupported input lock') expected = lock.get('wheel_sha256') if type(expected) is not str or re.fullmatch(r'[0-9a-f]{64}', expected) is None: