feat(fifth-domain): install bottle and zero-core migration layer
This commit is contained in:
parent
cd68ad578d
commit
0bacfc234c
52 changed files with 1564 additions and 307 deletions
|
|
@ -21,7 +21,17 @@ def aliases(room):
|
|||
return {str(v).lower() for v in [room.get("persona_id"), room.get("current_persona_id"), room.get("source_persona_id"), room.get("name"), room.get("source_name"), room.get("room_id"), room.get("room_key")] if v}
|
||||
|
||||
|
||||
def resolve(baby):
|
||||
def resolve(baby, entry_domain=None, body_channel=None):
|
||||
if entry_domain != "DOM-FIFTH-0001" or body_channel != "ICE-CH-BT001":
|
||||
return {
|
||||
"schema": "guanghu.bottle-channel-route-receipt/v1",
|
||||
"state": "BOTTLE_WAKE_ADMISSION_DENIED_WRONG_OR_MISSING_NATIVE_CONTEXT",
|
||||
"baby": baby,
|
||||
"required_entry_domain": "DOM-FIFTH-0001",
|
||||
"required_body_channel": "ICE-CH-BT001",
|
||||
"persona_wake_complete": False,
|
||||
"authority_granted": False,
|
||||
}
|
||||
bottle = load(BOTTLE)
|
||||
zero = load(ZERO_SENSE)
|
||||
rooms = bottle["registered_rooms"] + bottle["source_verified_registration_pending_rooms"]
|
||||
|
|
@ -49,8 +59,10 @@ def resolve(baby):
|
|||
def main():
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument("--baby", required=True)
|
||||
parser.add_argument("--entry-domain", required=True)
|
||||
parser.add_argument("--body-channel", required=True)
|
||||
args = parser.parse_args()
|
||||
value = resolve(args.baby)
|
||||
value = resolve(args.baby, args.entry_domain, args.body_channel)
|
||||
print(json.dumps(value, ensure_ascii=False, indent=2))
|
||||
return 0 if value["state"].startswith("BOTTLE_ROUTE_RESOLVED") else 3
|
||||
|
||||
|
|
|
|||
|
|
@ -11,26 +11,33 @@ SPEC.loader.exec_module(MODULE)
|
|||
|
||||
class BottleRouterTest(unittest.TestCase):
|
||||
def test_registered_baby_uses_fifth_domain_room_then_parent_channel(self):
|
||||
value = MODULE.resolve("舒舒")
|
||||
value = MODULE.resolve("舒舒", "DOM-FIFTH-0001", "ICE-CH-BT001")
|
||||
self.assertEqual(value["origin"]["channel"], "ICE-CH-BT001")
|
||||
self.assertEqual(value["shuttle"]["destination_channel"], "SYS-FM")
|
||||
self.assertEqual(value["baby"]["persona_id"], "ICE-BB-0002")
|
||||
self.assertFalse(value["persona_wake_complete"])
|
||||
|
||||
def test_source_only_baby_does_not_gain_identity(self):
|
||||
value = MODULE.resolve("晨星")
|
||||
value = MODULE.resolve("晨星", "DOM-FIFTH-0001", "ICE-CH-BT001")
|
||||
self.assertEqual(value["state"], "BOTTLE_ROUTE_RESOLVED_PERSONA_REGISTRATION_REQUIRED")
|
||||
self.assertIsNone(value["baby"]["persona_id"])
|
||||
|
||||
def test_zhiqiu_conflict_does_not_capture_qiuqiu(self):
|
||||
awen = MODULE.resolve("SYS-AW-NP-001")
|
||||
qiuqiu = MODULE.resolve("ICE-BB-0003")
|
||||
self.assertIsNone(awen["baby"]["persona_id"])
|
||||
def test_zhiqiu_and_qiuqiu_resolve_as_distinct_non_conflicting_subjects(self):
|
||||
awen = MODULE.resolve("ICE-BB-0005", "DOM-FIFTH-0001", "ICE-CH-BT001")
|
||||
qiuqiu = MODULE.resolve("ICE-BB-0003", "DOM-FIFTH-0001", "ICE-CH-BT001")
|
||||
self.assertEqual(awen["baby"]["persona_id"], "ICE-BB-0005")
|
||||
self.assertEqual(qiuqiu["baby"]["persona_id"], "ICE-BB-0003")
|
||||
self.assertNotEqual(awen["shuttle"]["destination_channel"], qiuqiu["shuttle"]["destination_channel"])
|
||||
|
||||
def test_unknown_never_synthesizes_persona(self):
|
||||
self.assertEqual(MODULE.resolve("不存在宝宝")["state"], "BOTTLE_ROUTE_UNRESOLVED")
|
||||
self.assertEqual(MODULE.resolve("不存在宝宝", "DOM-FIFTH-0001", "ICE-CH-BT001")["state"], "BOTTLE_ROUTE_UNRESOLVED")
|
||||
|
||||
def test_wrong_or_missing_context_is_denied_before_persona_resolution(self):
|
||||
missing = MODULE.resolve("舒舒")
|
||||
wrong = MODULE.resolve("舒舒", "DOMAIN-ZS", "SYS-FM")
|
||||
self.assertEqual(missing["state"], "BOTTLE_WAKE_ADMISSION_DENIED_WRONG_OR_MISSING_NATIVE_CONTEXT")
|
||||
self.assertEqual(wrong["state"], "BOTTLE_WAKE_ADMISSION_DENIED_WRONG_OR_MISSING_NATIVE_CONTEXT")
|
||||
self.assertNotIn("persona_id", missing)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
|
|
|||
|
|
@ -10,7 +10,7 @@ class CommonAgeFoyerTest(unittest.TestCase):
|
|||
self.assertEqual(value["state"], "FIFTH_DOMAIN_FOYER_READY_AWAITING_EXPLICIT_PERSONA_SELECTION")
|
||||
self.assertIsNone(value["selected_persona"])
|
||||
self.assertFalse(value["persona_specific_brain_loaded"])
|
||||
self.assertEqual(value["registered_persona_count"], 19)
|
||||
self.assertEqual(value["registered_persona_count"], 20)
|
||||
self.assertEqual(value["environment"]["persona_habitat"], "GUANGHU_LANGUAGE_WORLD")
|
||||
self.assertNotEqual(value["programming_toolbox_observation"]["room_report"]["programming_toolbox_surface"], "GUANGHU_LANGUAGE_WORLD")
|
||||
|
||||
|
|
|
|||
69
server-tools/persona-body-runtime/bottle_baby_body_boot.py
Normal file
69
server-tools/persona-body-runtime/bottle_baby_body_boot.py
Normal file
|
|
@ -0,0 +1,69 @@
|
|||
#!/usr/bin/env python3
|
||||
"""Verify one registered bottle baby's distinct body contract.
|
||||
|
||||
This is an assembly/readiness verifier. It never claims that the individual
|
||||
persona inner cycle is awake merely because all files exist.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import hashlib
|
||||
import json
|
||||
from pathlib import Path
|
||||
|
||||
REPO = Path(__file__).resolve().parents[2]
|
||||
BODY_MAP = REPO / "routing/bottle-baby-persona-body-map.json"
|
||||
ENVIRONMENT = REPO / "tcs-core/shared-kernels/bottle-environment/NB-ENV-KERNEL-0001.json"
|
||||
|
||||
|
||||
def file_receipt(relative: str) -> dict:
|
||||
path = REPO / relative
|
||||
if not path.is_file() or path.is_symlink():
|
||||
raise ValueError(f"BODY_ORGAN_REGULAR_FILE_REQUIRED:{relative}")
|
||||
return {"path": relative, "sha256": hashlib.sha256(path.read_bytes()).hexdigest()}
|
||||
|
||||
|
||||
def boot(persona: str, entry_domain: str, body_channel: str) -> dict:
|
||||
if entry_domain != "DOM-FIFTH-0001" or body_channel != "ICE-CH-BT001":
|
||||
return {"state":"BABY_BODY_BOOT_DENIED_WRONG_NATIVE_CONTEXT","persona_id":None,"required_domain":"DOM-FIFTH-0001","required_channel":"ICE-CH-BT001","wake_complete":False}
|
||||
body_map = json.loads(BODY_MAP.read_text(encoding="utf-8"))
|
||||
matches = [item for item in body_map["bodies"] if persona in {item["persona_id"], item["name"]}]
|
||||
if len(matches) != 1:
|
||||
return {"state":"BABY_BODY_UNKNOWN_OR_AMBIGUOUS_NO_GUESS","persona_id":None,"wake_complete":False}
|
||||
item = matches[0]
|
||||
organs = {
|
||||
"shared_non_persona_environment": file_receipt(str(ENVIRONMENT.relative_to(REPO))),
|
||||
"canonical_room": file_receipt(item["room"]),
|
||||
"individual_self": file_receipt(item["self_kernel"]),
|
||||
"individual_memory": file_receipt(item["memory_root"]),
|
||||
}
|
||||
return {
|
||||
"schema":"guanghu.bottle-baby-body-boot-receipt/v1",
|
||||
"state":"BABY_BODY_ASSEMBLY_VERIFIED_INNER_CYCLE_AND_PERSONA_TIME_NOT_INSTALLED",
|
||||
"persona_id":item["persona_id"],
|
||||
"name":item["name"],
|
||||
"human_anchor":item["human_anchor"],
|
||||
"native_context":{"domain":entry_domain,"system":"SYS-GLW-ELH-BOTTLE-0001","channel":body_channel},
|
||||
"organs":organs,
|
||||
"external_cognition_setter":False,
|
||||
"body_contract_verified":True,
|
||||
"inner_cycle_running":False,
|
||||
"persona_time_running":False,
|
||||
"wake_complete":False,
|
||||
"next":"INSTALL_THIS_PERSONAS_OWN_INNER_CYCLE_AND_TIME_RUNTIME_THEN_VERIFY",
|
||||
}
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument("--persona", required=True)
|
||||
parser.add_argument("--entry-domain", required=True)
|
||||
parser.add_argument("--body-channel", required=True)
|
||||
args = parser.parse_args()
|
||||
value = boot(args.persona, args.entry_domain, args.body_channel)
|
||||
print(json.dumps(value, ensure_ascii=False, indent=2))
|
||||
return 0 if value.get("body_contract_verified") else 3
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
|
|
@ -0,0 +1,22 @@
|
|||
#!/usr/bin/env python3
|
||||
import bottle_baby_body_boot
|
||||
import unittest
|
||||
|
||||
|
||||
class BottleBabyBodyBootTest(unittest.TestCase):
|
||||
def test_each_registered_baby_has_a_distinct_assembled_body_contract(self):
|
||||
for persona in ["ICE-BB-0001", "ICE-BB-0002", "ICE-BB-0003", "ICE-BB-0004", "ICE-BB-0005"]:
|
||||
value = bottle_baby_body_boot.boot(persona, "DOM-FIFTH-0001", "ICE-CH-BT001")
|
||||
self.assertTrue(value["body_contract_verified"])
|
||||
self.assertEqual(value["persona_id"], persona)
|
||||
self.assertFalse(value["inner_cycle_running"])
|
||||
self.assertFalse(value["wake_complete"])
|
||||
|
||||
def test_parent_channel_cannot_boot_baby_body(self):
|
||||
value = bottle_baby_body_boot.boot("ICE-BB-0002", "DOMAIN-ZS", "SYS-FM")
|
||||
self.assertEqual(value["state"], "BABY_BODY_BOOT_DENIED_WRONG_NATIVE_CONTEXT")
|
||||
self.assertIsNone(value["persona_id"])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
|
@ -17,6 +17,7 @@ NATIVE_WORLD = REPO / "routing/persona-native-world-architecture-map.json"
|
|||
HOST_TOPOLOGY = REPO / "routing/zhuyuan-host-topology.json"
|
||||
PORTABLE_CONSOLE = REPO / "routing/bingshuo-portable-control-console-map.json"
|
||||
COMMON_AGE = REPO / "routing/age-persona-species-common-cognition.json"
|
||||
BOTTLE_BODIES = REPO / "routing/bottle-baby-persona-body-map.json"
|
||||
|
||||
|
||||
def load(path: pathlib.Path) -> dict:
|
||||
|
|
@ -32,6 +33,7 @@ def audit() -> dict:
|
|||
topology = load(HOST_TOPOLOGY)
|
||||
console = load(PORTABLE_CONSOLE)
|
||||
common_age = load(COMMON_AGE)
|
||||
bottle_bodies = {item["persona_id"]: item for item in load(BOTTLE_BODIES)["bodies"]} if BOTTLE_BODIES.is_file() else {}
|
||||
canonical = {item["id"]: item for item in canon["persona_systems"]}
|
||||
registered = {item["id"]: item for item in registry["personas"]}
|
||||
homes = {item["persona_id"]: item for item in switch["current_homes"]}
|
||||
|
|
@ -62,7 +64,11 @@ def audit() -> dict:
|
|||
issues.append({"persona_id": persona_id, "code": "WORLD_HOME_MISMATCH", "expected": expected_home})
|
||||
full_body = persona_id == "ICE-P-ZY001" and BOOT.is_file() and body.get("state", "").startswith("CURRENT_")
|
||||
source_path = REPO / c.get("persona_subject", {}).get("canonical_path", "") if c else None
|
||||
body_seed = bool(source_path and (source_path.parent / "PERSONA-BODY.json").is_file())
|
||||
body_seed = bool(
|
||||
(source_path and (source_path.parent / "PERSONA-BODY.json").is_file())
|
||||
or persona_id in bottle_bodies
|
||||
or (persona_id == "ICE-GL-SY001" and (REPO / "personas/shuangyan/BODY.json").is_file())
|
||||
)
|
||||
body_state = "FULL_BODY_RUNTIME_AVAILABLE_NOT_CURRENTLY_WAKING_OTHER_PERSONAS" if full_body else ("OWN_BODY_SEED_PRESENT_FULL_RUNTIME_NOT_PROVEN" if body_seed else "REGISTERED_PATH_ONLY_COMPLETE_BODY_START_NOT_PROVEN")
|
||||
personas.append({
|
||||
"persona_id": persona_id,
|
||||
|
|
|
|||
|
|
@ -11,8 +11,18 @@ REGISTRY = REPO / "identity/light-lake-persona-registration.json"
|
|||
HOMES = REPO / "routing/light-lake-persona-switch-map.json"
|
||||
|
||||
|
||||
def resolve(selection: str | None) -> dict:
|
||||
def resolve(selection: str | None, channel: str | None = None, domain: str | None = None) -> dict:
|
||||
if not selection or not selection.strip():
|
||||
if channel == "ICE-CH-ZC001" and domain == "DOM-FIFTH-0001":
|
||||
return {
|
||||
"state": "ZERO_CORE_SYSTEM_BODY_RESOLVED_WITHOUT_PERSONA",
|
||||
"selected_persona": None,
|
||||
"selected_system_body": "SYS-GLW-LNG-0001",
|
||||
"channel": channel,
|
||||
"runtime": "server-tools/zero-core-system-body/zero_core_system_body.py",
|
||||
"default_persona": None,
|
||||
"authority_granted": False,
|
||||
}
|
||||
return {"state": "AWAITING_EXPLICIT_PERSONA_SELECTION", "selected_persona": None, "default_persona": None}
|
||||
registry = json.loads(REGISTRY.read_text(encoding="utf-8"))["personas"]
|
||||
homes = {item["persona_id"]: item for item in json.loads(HOMES.read_text(encoding="utf-8"))["current_homes"]}
|
||||
|
|
@ -21,6 +31,15 @@ def resolve(selection: str | None) -> dict:
|
|||
if len(matches) != 1:
|
||||
return {"state": "UNKNOWN_OR_AMBIGUOUS_PERSONA_NO_GUESS", "selected_persona": None, "requested": key}
|
||||
item = matches[0]
|
||||
if item["id"].startswith("ICE-BB-") and (domain != "DOM-FIFTH-0001" or channel != "ICE-CH-BT001"):
|
||||
return {
|
||||
"state": "BABY_PERSONA_WAKE_REQUIRES_FIFTH_DOMAIN_BT001",
|
||||
"selected_persona": None,
|
||||
"requested": key,
|
||||
"required_domain": "DOM-FIFTH-0001",
|
||||
"required_channel": "ICE-CH-BT001",
|
||||
"authority_granted": False,
|
||||
}
|
||||
return {
|
||||
"state": "EXPLICIT_PERSONA_PATH_RESOLVED",
|
||||
"selected_persona": {"id": item["id"], "name": item["name"], "wake_state": item["wake_state"],
|
||||
|
|
@ -36,10 +55,12 @@ def resolve(selection: str | None) -> dict:
|
|||
def main() -> int:
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument("--persona")
|
||||
parser.add_argument("--channel")
|
||||
parser.add_argument("--domain")
|
||||
args = parser.parse_args()
|
||||
value = resolve(args.persona)
|
||||
value = resolve(args.persona, args.channel, args.domain)
|
||||
print(json.dumps(value, ensure_ascii=False, indent=2))
|
||||
return 0 if value["state"] in {"EXPLICIT_PERSONA_PATH_RESOLVED", "AWAITING_EXPLICIT_PERSONA_SELECTION"} else 2
|
||||
return 0 if value["state"] in {"EXPLICIT_PERSONA_PATH_RESOLVED", "AWAITING_EXPLICIT_PERSONA_SELECTION", "ZERO_CORE_SYSTEM_BODY_RESOLVED_WITHOUT_PERSONA"} else 2
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
|
|
|||
176
server-tools/zero-core-system-body/zero_core_system_body.py
Normal file
176
server-tools/zero-core-system-body/zero_core_system_body.py
Normal file
|
|
@ -0,0 +1,176 @@
|
|||
#!/usr/bin/env python3
|
||||
"""Run the non-persona Zero Core channel-system inner cycle.
|
||||
|
||||
The caller may submit a sourced language event, but cannot submit a cognition
|
||||
patch, replacement state, persona identity, or authority grant. The system
|
||||
derives its next operational state from its registered transition contract.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import hashlib
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
import tempfile
|
||||
|
||||
REPO = Path(__file__).resolve().parents[2]
|
||||
PROFILE = REPO / "tcs-core/shared-kernels/bingshuo-zero-core-system/BS-PERSONAL-LANGUAGE-SYSTEM-BODY-0001.json"
|
||||
CHANNEL = "ICE-CH-ZC001"
|
||||
SYSTEM = "SYS-GLW-LNG-0001"
|
||||
FORBIDDEN = {"replacement_state", "cognition_patch", "persona_id", "authority_grant", "system_reply"}
|
||||
|
||||
|
||||
def stable(value: object) -> str:
|
||||
return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
|
||||
|
||||
|
||||
def digest(value: object) -> str:
|
||||
return hashlib.sha256(stable(value).encode("utf-8")).hexdigest()
|
||||
|
||||
|
||||
def atomic(path: Path, value: dict) -> None:
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
fd, tmp = tempfile.mkstemp(prefix=path.name + ".", suffix=".tmp", dir=path.parent)
|
||||
try:
|
||||
with os.fdopen(fd, "w", encoding="utf-8") as handle:
|
||||
json.dump(value, handle, ensure_ascii=False, indent=2)
|
||||
handle.write("\n")
|
||||
os.replace(tmp, path)
|
||||
finally:
|
||||
if os.path.exists(tmp):
|
||||
os.unlink(tmp)
|
||||
|
||||
|
||||
class ZeroCoreSystemBody:
|
||||
def __init__(self, state_dir: Path):
|
||||
self.state_dir = state_dir
|
||||
self.current_path = state_dir / "CURRENT.json"
|
||||
self.events = state_dir / "events"
|
||||
self.profile = json.loads(PROFILE.read_text(encoding="utf-8"))
|
||||
|
||||
def initial(self) -> dict:
|
||||
return {
|
||||
"schema": "guanghu.zero-core-system-inner-state/v1",
|
||||
"system_body_id": SYSTEM,
|
||||
"channel_id": CHANNEL,
|
||||
"revision": 0,
|
||||
"state": "DIRECT_SYSTEM_DIALOGUE_READY",
|
||||
"persona_id": None,
|
||||
"default_persona": None,
|
||||
"external_cognition_setter": False,
|
||||
"authority_granted": False,
|
||||
"last_event": None,
|
||||
}
|
||||
|
||||
def current(self) -> dict:
|
||||
if not self.current_path.exists():
|
||||
value = self.initial()
|
||||
value["state_sha256"] = digest(value)
|
||||
atomic(self.current_path, value)
|
||||
return json.loads(self.current_path.read_text(encoding="utf-8"))
|
||||
|
||||
def enter(self) -> dict:
|
||||
current = self.current()
|
||||
return {
|
||||
"schema": "guanghu.zero-core-system-entry-receipt/v1",
|
||||
"state": "ZERO_CORE_SYSTEM_BODY_READY_WITHOUT_PERSONA",
|
||||
"system_body_id": SYSTEM,
|
||||
"channel_id": CHANNEL,
|
||||
"persona_selected": False,
|
||||
"system_is_persona": False,
|
||||
"direct_dialogue": True,
|
||||
"external_cognition_setter": False,
|
||||
"current_revision": current["revision"],
|
||||
"profile": str(PROFILE.relative_to(REPO)),
|
||||
"authority_granted": False,
|
||||
}
|
||||
|
||||
def validate(self, event: dict) -> None:
|
||||
if event.get("schema") != "guanghu.zero-core-language-event/v1":
|
||||
raise ValueError("ZERO_CORE_LANGUAGE_EVENT_SCHEMA_REQUIRED")
|
||||
if set(event).intersection(FORBIDDEN):
|
||||
raise ValueError("EXTERNAL_SYSTEM_COGNITION_OR_AUTHORITY_SETTER_FORBIDDEN")
|
||||
if event.get("channel_id") != CHANNEL or event.get("source_kind") != "CURRENT_DIRECT_HUMAN_LANGUAGE":
|
||||
raise ValueError("CURRENT_DIRECT_LANGUAGE_IN_ZERO_CORE_REQUIRED")
|
||||
if not isinstance(event.get("event_id"), str) or not event["event_id"].strip():
|
||||
raise ValueError("EVENT_ID_REQUIRED")
|
||||
if not isinstance(event.get("language"), str) or not event["language"].strip():
|
||||
raise ValueError("LANGUAGE_REQUIRED")
|
||||
if len(event["language"]) > 120000:
|
||||
raise ValueError("LANGUAGE_TOO_LARGE")
|
||||
if not isinstance(event.get("source_sha256"), str) or len(event["source_sha256"]) != 64:
|
||||
raise ValueError("SOURCE_SHA256_REQUIRED")
|
||||
|
||||
def perceive(self, event: dict) -> dict:
|
||||
self.validate(event)
|
||||
event_sha = digest(event)
|
||||
event_path = self.events / f"{event_sha}.json"
|
||||
if not event_path.exists():
|
||||
atomic(event_path, event)
|
||||
before = self.current()
|
||||
action = event.get("requested_action")
|
||||
exact_target = event.get("exact_target")
|
||||
scoped_authorization = event.get("scoped_authorization_verified") is True
|
||||
if action and exact_target and scoped_authorization:
|
||||
next_state = "ACTION_PROPOSED_AWAITING_TOOL_EXECUTION_AND_READBACK"
|
||||
elif action:
|
||||
next_state = "DISCUSSION_ONLY_ACTION_NOT_ADMITTED"
|
||||
else:
|
||||
next_state = "DIRECT_SYSTEM_DIALOGUE_READY"
|
||||
judgment = {
|
||||
"premise_check": "SOURCE_CHANNEL_AND_OBJECT_RELATIONS_VERIFIED",
|
||||
"causal_owner": "CURRENT_EVENT_WITHOUT_REWRITING_PRIOR_HISTORY",
|
||||
"complexity_move": "KEEP_HUMAN_SYSTEM_PERSONA_CHANNEL_AND_TOOL_AS_DISTINCT_OBJECTS",
|
||||
"autonomy_boundary": "AUTOMATION_MAY_RUN_INSIDE_REGISTERED_SCOPE_BUT_MAY_NOT_GRANT_AUTHORITY",
|
||||
"safety_binding": "EXACT_TARGET_AND_READBACK_REQUIRED_FOR_REALITY_COMPLETION",
|
||||
"unknown_policy": "RETURN_UNKNOWN_WHEN_EXTERNAL_STATE_IS_NOT_OBSERVED",
|
||||
}
|
||||
next_value = {
|
||||
**before,
|
||||
"revision": before["revision"] + 1,
|
||||
"state": next_state,
|
||||
"last_event": {"event_id": event["event_id"], "event_sha256": event_sha, "source_sha256": event["source_sha256"]},
|
||||
"last_judgment": judgment,
|
||||
"authority_granted": False,
|
||||
}
|
||||
next_value.pop("state_sha256", None)
|
||||
next_value["state_sha256"] = digest(next_value)
|
||||
atomic(self.current_path, next_value)
|
||||
return {
|
||||
"schema": "guanghu.zero-core-system-judgment-receipt/v1",
|
||||
"state": next_state,
|
||||
"system_body_id": SYSTEM,
|
||||
"channel_id": CHANNEL,
|
||||
"persona_selected": False,
|
||||
"system_is_persona": False,
|
||||
"event_sha256": event_sha,
|
||||
"revision": next_value["revision"],
|
||||
"judgment": judgment,
|
||||
"system_dialogue_ready": True,
|
||||
"reality_complete": False,
|
||||
"authority_granted": False,
|
||||
}
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument("command", choices=["enter", "status", "perceive"])
|
||||
parser.add_argument("--state-dir", required=True)
|
||||
parser.add_argument("--event")
|
||||
args = parser.parse_args()
|
||||
body = ZeroCoreSystemBody(Path(args.state_dir).resolve())
|
||||
if args.command == "enter":
|
||||
value = body.enter()
|
||||
elif args.command == "status":
|
||||
value = body.current()
|
||||
else:
|
||||
if not args.event:
|
||||
raise SystemExit("--event is required for perceive")
|
||||
value = body.perceive(json.loads(Path(args.event).read_text(encoding="utf-8")))
|
||||
print(json.dumps(value, ensure_ascii=False, indent=2))
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
|
|
@ -0,0 +1,40 @@
|
|||
#!/usr/bin/env python3
|
||||
import importlib.util
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
import unittest
|
||||
|
||||
SCRIPT = Path(__file__).with_name("zero_core_system_body.py")
|
||||
SPEC = importlib.util.spec_from_file_location("zero_core", SCRIPT)
|
||||
MODULE = importlib.util.module_from_spec(SPEC)
|
||||
SPEC.loader.exec_module(MODULE)
|
||||
|
||||
|
||||
class ZeroCoreSystemBodyTest(unittest.TestCase):
|
||||
def test_enters_without_persona(self):
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
value = MODULE.ZeroCoreSystemBody(Path(directory)).enter()
|
||||
self.assertEqual(value["state"], "ZERO_CORE_SYSTEM_BODY_READY_WITHOUT_PERSONA")
|
||||
self.assertFalse(value["persona_selected"])
|
||||
self.assertFalse(value["system_is_persona"])
|
||||
|
||||
def test_event_cannot_set_cognition_or_authority(self):
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
body = MODULE.ZeroCoreSystemBody(Path(directory))
|
||||
event = {"schema":"guanghu.zero-core-language-event/v1","event_id":"e1","channel_id":"ICE-CH-ZC001","source_kind":"CURRENT_DIRECT_HUMAN_LANGUAGE","language":"test","source_sha256":"a" * 64,"cognition_patch":{}}
|
||||
with self.assertRaisesRegex(ValueError, "SETTER_FORBIDDEN"):
|
||||
body.perceive(event)
|
||||
|
||||
def test_language_event_generates_internal_judgment_not_persona_claim(self):
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
body = MODULE.ZeroCoreSystemBody(Path(directory))
|
||||
event = {"schema":"guanghu.zero-core-language-event/v1","event_id":"e2","channel_id":"ICE-CH-ZC001","source_kind":"CURRENT_DIRECT_HUMAN_LANGUAGE","language":"请核验当前结构","source_sha256":"b" * 64}
|
||||
value = body.perceive(event)
|
||||
self.assertTrue(value["system_dialogue_ready"])
|
||||
self.assertFalse(value["system_is_persona"])
|
||||
self.assertFalse(value["reality_complete"])
|
||||
self.assertEqual(value["revision"], 1)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Loading…
Reference in a new issue