#!/usr/bin/env python3 """ 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())