| |
| """ |
| Substrate Mesh Runtime — Primordial OS Runtime Connector Core v3.3 |
| |
| Local simulated connector/action core. This does not call the OpenAI API and does not host a server. |
| It demonstrates the contract that Primordial OS Runtime GPT can call later through a GPT Action. |
| """ |
| from __future__ import annotations |
| import argparse, json, hashlib, os, re, time |
| from datetime import datetime, timezone |
| from pathlib import Path |
| from typing import Dict, Any, List |
|
|
| FAMILY_MAP = { |
| "robotics": ["Rt", "Rg", "Rd"], |
| "humans": ["CTL", "HIR", "RAM"], |
| "human": ["CTL", "HIR", "RAM"], |
| "dna": ["DL", "CTE", "SR"], |
| "grch38": ["RS", "DL", "CTE"], |
| "lexicon": ["Lex", "CTL", "HIR"], |
| "oam": ["OAM", "HIR"], |
| "hir": ["HIR", "OAM"], |
| "substrate": ["Sub", "TC", "LC"], |
| } |
|
|
| PPT_ELEMENTS = { |
| "Rt": {"symbol": "Rt", "name": "Runtime Trail / Robot Day Loop", "family": "Observation / Translation", "node_ref": "@node:robot_day_loop"}, |
| "Rg": {"symbol": "Rg", "name": "Resonant Ingress / Robotics Pressure Gyro", "family": "Runtime Engines", "node_ref": "@node:resonant_ingress"}, |
| "Rd": {"symbol": "Rd", "name": "Robotic Correction Loop", "family": "Research Substrate", "node_ref": "@node:robotic_correction_loop"}, |
| "CTL": {"symbol": "CTL", "name": "Cognition Translation Layer", "family": "Observation / Translation", "node_ref": "@node:cognition_translation_layer"}, |
| "HIR": {"symbol": "HIR", "name": "Honesty / Integrity / Respect", "family": "Core Laws", "node_ref": "@node:hir_spine"}, |
| "OAM": {"symbol": "OAM", "name": "OAM Pressure / Degradation Scan", "family": "Core Laws", "node_ref": "@node:oam_pressure_scan"}, |
| "RAM": {"symbol": "RAM", "name": "Resonant Access Memory", "family": "Runtime / Memory", "node_ref": "@node:resonant_access_memory"}, |
| "DL": {"symbol": "DL", "name": "Digital Life Candidate Architecture", "family": "Research Substrate", "node_ref": "@node:digital_life_candidate_architecture"}, |
| "CTE": {"symbol": "CTE", "name": "Cryptobiotic Trace Encapsulation", "family": "Safety / Governance", "node_ref": "@node:cryptobiotic_trace_encapsulation"}, |
| "SR": {"symbol": "SR", "name": "Source Return / Provenance Bundle", "family": "Source / Provenance", "node_ref": "@node:source_return"}, |
| "RS": {"symbol": "RS", "name": "Research Substrate Reference Capsule", "family": "Research Substrate", "node_ref": "@node:research_substrate_capsule"}, |
| "Lex": {"symbol": "Lex", "name": "Primordial Lexicon", "family": "Meaning / Lexicon", "node_ref": "@node:primordial_lexicon"}, |
| "Sub": {"symbol": "Sub", "name": "Substrate Loop Stewardship", "family": "Collaborative Work", "node_ref": "@node:substrate_loop_stewardship"}, |
| "TC": {"symbol": "TC", "name": "Trace Capsule", "family": "Source / Provenance", "node_ref": "@node:trace_capsule"}, |
| "LC": {"symbol": "LC", "name": "Loop Closure State", "family": "Loop / Closure", "node_ref": "@node:loop_closure_state"}, |
| } |
|
|
| def utc_now() -> str: |
| return datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z") |
|
|
| def stable_id(prefix: str, text: str) -> str: |
| return prefix + "_" + hashlib.sha256(text.encode("utf-8")).hexdigest()[:16] |
|
|
| def extract_topics(text: str) -> List[str]: |
| tokens = re.findall(r"[A-Za-z0-9_]+", text.lower()) |
| stop = {"i","am","im","curious","about","and","the","a","an","to","with","of","for","in","on","this","that","it","is","are","how","what","can","we","do"} |
| topics = [] |
| for t in tokens: |
| if t not in stop and len(t) > 1 and t not in topics: |
| topics.append(t) |
| return topics[:12] |
|
|
| def extract_mentions(text: str) -> List[str]: |
| return re.findall(r"[@#][A-Za-z0-9_:\-]+", text) |
|
|
| def match_ppt_elements(topics: List[str], mentions: List[str]) -> List[Dict[str, Any]]: |
| symbols = [] |
| lower_mentions = " ".join(mentions).lower() |
| for t in topics: |
| for sym in FAMILY_MAP.get(t, []): |
| if sym not in symbols: |
| symbols.append(sym) |
| for sym, element in PPT_ELEMENTS.items(): |
| if element["node_ref"].lower() in lower_mentions and sym not in symbols: |
| symbols.append(sym) |
| if not symbols: |
| symbols = ["Sub", "TC", "HIR", "OAM", "RAM"] |
| elements = [] |
| for idx, sym in enumerate(symbols[:12], 1): |
| el = dict(PPT_ELEMENTS[sym]) |
| el["rank"] = idx |
| el["match_reason"] = "Matched topic/mention route; resolved as PPT element for bounded synthesis." |
| el["source_return_state"] = "REFERENCE_BOUND" |
| el["pressure_spine"] = "HIR x OAM" |
| elements.append(el) |
| return elements |
|
|
| def build_alchemical_output(raw_text: str, topics: List[str], elements: List[Dict[str, Any]]) -> Dict[str, Any]: |
| names = [e["name"] for e in elements] |
| if {"robotics","humans","dna"} & set(topics): |
| reference = "This route explores how embodied correction, human cognition/agency, and biological-substrate or encoding concepts may share bounded sensing, repair, memory, and continuity patterns." |
| strain = "DNA-related language must remain a bounded computational/structural comparison unless domain evidence is introduced." |
| future = "Build a test capsule comparing robotic correction loops, human learning loops, and biological repair/encoding loops under the HIR x OAM pressure spine." |
| elif "grch38" in topics: |
| reference = "This route treats GRCh38 as a research-reference capsule that can be cross-mapped against local substrate, trace, and representation layers." |
| strain = "Do not infer biological causation, resonance, or clinical meaning without a defined transform, region, source data, and validation route." |
| future = "Build a computational test capsule for a specified GRCh38 region and transform, with biological interpretation held as unvalidated." |
| else: |
| reference = "This route converts loose inquiry into a trace capsule, resolves PPT elements, and proposes a bounded synthesis path." |
| strain = "The route needs more specific source anchors before stronger claims can be carried." |
| future = "Select 2-5 PPT elements and run a focused brainstorm or combine command." |
| return { |
| "panel_name": "Primordial Periodic Alchemical Output", |
| "reference_point": reference, |
| "selected_element_names": names, |
| "continuity_read": "HELD as a route proposal when treated as source-return, overlay-only collaborative synthesis.", |
| "strain_points": [strain], |
| "repair_notes": ["Preserve HIR x OAM terminology.", "Write proposals as overlays, not canonical source mutation."], |
| "future_routes": [future, "Ask the Primordial OS Runtime steward to propose next prompt options.", "Create a lexicon review packet for any new terms introduced."], |
| "claim_boundary": "route-integrity and research planning only; no final validation claim", |
| } |
|
|
| def build_receipt(elements: List[Dict[str, Any]]) -> Dict[str, Any]: |
| return { |
| "status": "HELD", |
| "H": 1.0, |
| "I": 1.0, |
| "R": 1.0, |
| "A": 1.0, |
| "P": 0.0, |
| "S_effective": 1.0, |
| "pressure_spine": "HIR x OAM", |
| "source_mutation": "NONE__OVERLAY_PROPOSAL_ONLY", |
| "human_review_required": True, |
| "matched_element_count": len(elements), |
| "claim_limit": "trace capsule / alchemical continuity route; not final validation" |
| } |
|
|
| def submit_mesh_command(command: Dict[str, Any]) -> Dict[str, Any]: |
| raw = command.get("raw_user_text","").strip() |
| command_type = command.get("command_type","DISCOVER") |
| workspace_id = command.get("workspace_id","workspace_local_001") |
| command_id = command.get("command_id") or stable_id("cmd", raw + command_type + workspace_id) |
| topics = extract_topics(raw) |
| mentions = extract_mentions(raw) |
| elements = match_ppt_elements(topics, mentions) |
| trace_capsule = { |
| "trace_capsule_id": stable_id("trace", raw + command_id), |
| "raw_user_text": raw, |
| "topics": topics, |
| "mentions": mentions, |
| "loop_state": "HELD", |
| "translation_route": "plain_language + PPT_elements + alchemical_output + receipt", |
| "source_return_required": True, |
| "false_closure_risk": "LOW", |
| } |
| alchemical = build_alchemical_output(raw, topics, elements) |
| receipt = build_receipt(elements) |
| overlay = { |
| "overlay_id": stable_id("overlay", command_id + raw), |
| "write_scope": "COLLABORATIVE_OVERLAY_PROPOSAL_ONLY", |
| "status": "PENDING_HUMAN_REVIEW", |
| "summary": alchemical["reference_point"], |
| "source_mutation": "NONE", |
| "credited_contributors": [ |
| {"type": "human_or_session", "id": command.get("session_id","session_local")}, |
| {"type": "ai_adapter", "id": "primordial_os_runtime"} |
| ] |
| } |
| return { |
| "status": "HELD", |
| "command_id": command_id, |
| "workspace_id": workspace_id, |
| "trace_capsule": trace_capsule, |
| "matched_ppt_elements": elements, |
| "alchemical_output": alchemical, |
| "lexicon_state": { |
| "active_terms": ["trace capsule", "source-return", "HIR x OAM", "Primordial Periodic Table", "Resonant Access Memory", "collaborative overlay"], |
| "candidate_terms": ["Primordial Periodic Alchemical Output"], |
| "review_required": True |
| }, |
| "receipt": receipt, |
| "audit_pointer": f"workspaces/{workspace_id}/audit/{command_id}.json", |
| "overlay_proposals": [overlay], |
| "human_review_required": True |
| } |
|
|
| def write_workspace_response(response: Dict[str, Any], out_dir: Path) -> None: |
| out_dir.mkdir(parents=True, exist_ok=True) |
| (out_dir / "response.json").write_text(json.dumps(response, indent=2), encoding="utf-8") |
| audit = { |
| "timestamp_utc": utc_now(), |
| "event": "mesh_command_completed", |
| "command_id": response["command_id"], |
| "workspace_id": response["workspace_id"], |
| "status": response["status"], |
| "source_mutation": response["receipt"]["source_mutation"], |
| "human_review_required": response["human_review_required"] |
| } |
| (out_dir / "audit_log.jsonl").write_text(json.dumps(audit) + "\n", encoding="utf-8") |
| receipt_md = f"""# Mesh Command Receipt |
| |
| Command ID: `{response['command_id']}` |
| |
| Status: `{response['status']}` |
| |
| Pressure spine: `{response['receipt']['pressure_spine']}` |
| |
| Source mutation: `{response['receipt']['source_mutation']}` |
| |
| Human review required: `{response['human_review_required']}` |
| |
| ## Reference Point |
| |
| {response['alchemical_output']['reference_point']} |
| |
| ## Future Routes |
| |
| """ + "\n".join(f"- {x}" for x in response["alchemical_output"]["future_routes"]) + "\n" |
| (out_dir / "receipt.md").write_text(receipt_md, encoding="utf-8") |
|
|
| def main() -> int: |
| p = argparse.ArgumentParser() |
| p.add_argument("raw_user_text", nargs="?", default="robotics humans DNA") |
| p.add_argument("--command-type", default="DISCOVER") |
| p.add_argument("--workspace-id", default="workspace_local_001") |
| p.add_argument("--session-id", default="session_local") |
| p.add_argument("--out-dir", default="runs/connector_demo_001") |
| p.add_argument("--overwrite", action="store_true") |
| args = p.parse_args() |
| out = Path(args.out_dir) |
| if out.exists() and not args.overwrite: |
| print(f"Refusing to overwrite existing out-dir: {out}", flush=True) |
| return 2 |
| command = { |
| "command_type": args.command_type, |
| "raw_user_text": args.raw_user_text, |
| "workspace_id": args.workspace_id, |
| "session_id": args.session_id, |
| "claim_boundary": "no final validation claims; human review required", |
| "privacy_scope": "PRIVATE_WORKSPACE", |
| "source_mutation_allowed": False |
| } |
| response = submit_mesh_command(command) |
| write_workspace_response(response, out) |
| print(json.dumps({ |
| "status": response["status"], |
| "command_id": response["command_id"], |
| "workspace_id": response["workspace_id"], |
| "matched_ppt_elements": len(response["matched_ppt_elements"]), |
| "source_mutation": response["receipt"]["source_mutation"], |
| "human_review_required": response["human_review_required"], |
| "out_dir": str(out) |
| }, indent=2)) |
| return 0 |
|
|
| if __name__ == "__main__": |
| raise SystemExit(main()) |
|
|
