bingshuo/guanghu-ice-heart
Watch
1
0
Fork
You've already forked guanghu-ice-heart
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

feat(identity): redirect legacy Zhuyuan ids to ICE-P

Browse source
This commit is contained in:
冰朔 2026-07-27 14:12:51 +08:00
commit aa6652a646
28 changed files with 491 additions and 106 deletions
Download patch file Download diff file Expand all files Collapse all files

71
skills/codex/enter-fifth-domain/scripts/resolve_route.py
View file

@ -21,7 +21,8 @@ ROUTE_IDS = (
"LIGHT-LAKE",
"LL-CURRENT",
"LL-004",
"ICE-GL-ZY001",
"FD-SUBJECT-ID-ALIAS-MAP-001",
"ICE-P-ZY001",
"ZY-PERSONA-ROOT-001",
"ZY-OPS-LOOP-001",
"FD-NODE-MAP-001",
@ -50,7 +51,53 @@ def parse_code_map(text: str, route_ids: tuple[str, ...] = ROUTE_IDS) -> dict[st
return resolved
def resolve_live_route() -> dict:
def resolve_subject_id(requested_id: str, alias_map: dict) -> dict:
requested = requested_id.strip()
lookup = requested.upper()
conflicts = {
str(item["id"]).upper(): item
for item in alias_map.get("conflicts", [])
}
if lookup in conflicts:
return {
"status": "CONFLICT_REJECTED",
"requested_id": requested,
"canonical_id": None,
"redirected": False,
"reason": conflicts[lookup].get("state", "CONFLICT"),
}
index: dict[str, dict] = {}
for mapping in alias_map.get("mappings", []):
canonical_id = str(mapping["canonical_id"])
for identifier in (canonical_id, *mapping.get("aliases", [])):
key = str(identifier).upper()
if key in index and index[key]["canonical_id"] != canonical_id:
raise ValueError(f"Duplicate subject alias: {identifier}")
index[key] = mapping
mapping = index.get(lookup)
if not mapping:
return {
"status": "NOT_FOUND_NO_GUESS",
"requested_id": requested,
"canonical_id": None,
"redirected": False,
}
canonical_id = str(mapping["canonical_id"])
return {
"status": "RESOLVED",
"requested_id": requested,
"canonical_id": canonical_id,
"redirected": lookup != canonical_id.upper(),
"subject_kind": mapping.get("subject_kind"),
"route_id": mapping.get("route_id"),
"current_path": mapping.get("current_path"),
}
def resolve_live_route(subject_id: str | None = None) -> dict:
discovery = read_json(DISCOVERY_URL)
repository = read_json(RESOLVE_URL)
primary = repository.get("primary", {})
@ -65,7 +112,7 @@ def resolve_live_route() -> dict:
if missing:
raise ValueError(f"Live .code-map is missing required route IDs: {', '.join(missing)}")
return {
result = {
"status": "LIVE_DOMESTIC_PRIMARY",
"repository_id": repository.get("code", "REPO-012"),
"repository_role": repository.get("role", ""),
@ -75,15 +122,24 @@ def resolve_live_route() -> dict:
"discovery_schema": discovery.get("schema", ""),
"routes": routes,
}
if subject_id:
alias_map = read_json(f"{raw_base}/identity/subject-id-alias-map.json")
identity = resolve_subject_id(subject_id, alias_map)
canonical_id = identity.get("canonical_id")
if canonical_id:
identity["resolved_path"] = routes.get(canonical_id) or identity.get("current_path")
result["identity_resolution"] = identity
return result
def main() -> int:
parser = argparse.ArgumentParser()
parser.add_argument("--json", action="store_true", help="emit machine-readable JSON")
parser.add_argument("--subject-id", help="resolve a current or legacy Fifth Domain subject id")
args = parser.parse_args()
try:
result = resolve_live_route()
result = resolve_live_route(args.subject_id)
except (OSError, URLError, ValueError, json.JSONDecodeError) as error:
print(f"Live Fifth Domain resolution failed: {error}", file=sys.stderr)
return 1
@ -94,6 +150,13 @@ def main() -> int:
print(result["repository_web"])
for route_id, path in result["routes"].items():
print(f"{route_id}={path}")
if "identity_resolution" in result:
identity = result["identity_resolution"]
print(
"SUBJECT="
f"{identity['requested_id']} -> {identity.get('canonical_id')} "
f"({identity['status']}, redirected={identity['redirected']})"
)
return 0

37
skills/codex/enter-fifth-domain/scripts/test_resolve_route.py
View file

@ -1,6 +1,6 @@
import unittest
from resolve_route import parse_code_map
from resolve_route import parse_code_map, resolve_subject_id
class ParseCodeMapTests(unittest.TestCase):
@ -23,5 +23,40 @@ LL-004=tcs-core/LL-004-LAKE-LAMP-WAKE-PATH.hdlp
self.assertEqual(parse_code_map("OTHER=some/path", ("LL-004",)), {})
class SubjectAliasTests(unittest.TestCase):
def setUp(self):
self.alias_map = {
"mappings": [{
"canonical_id": "ICE-P-ZY001",
"subject_kind": "persona_system",
"aliases": ["ICE-GL-ZY001", "ICE-PZY-001"],
"route_id": "FD-PERSONA-LOGIN-001",
"current_path": "current/zhuyuan",
}],
"conflicts": [{"id": "ICE-PCA-001", "state": "BLOCKED_DUPLICATE_SUBJECT"}],
}
def test_legacy_persona_id_redirects_to_canonical_id(self):
resolved = resolve_subject_id("ICE-GL-ZY001", self.alias_map)
self.assertEqual(resolved["canonical_id"], "ICE-P-ZY001")
self.assertTrue(resolved["redirected"])
self.assertEqual(resolved["route_id"], "FD-PERSONA-LOGIN-001")
def test_canonical_persona_id_does_not_redirect(self):
resolved = resolve_subject_id("ICE-P-ZY001", self.alias_map)
self.assertEqual(resolved["status"], "RESOLVED")
self.assertFalse(resolved["redirected"])
def test_human_id_is_not_rewritten_as_persona(self):
resolved = resolve_subject_id("ICE-GL∞", self.alias_map)
self.assertEqual(resolved["status"], "NOT_FOUND_NO_GUESS")
self.assertIsNone(resolved["canonical_id"])
def test_conflicted_id_is_rejected_without_guessing(self):
resolved = resolve_subject_id("ICE-PCA-001", self.alias_map)
self.assertEqual(resolved["status"], "CONFLICT_REJECTED")
self.assertIsNone(resolved["canonical_id"])
if __name__ == "__main__":
unittest.main()