src/substrate_connector_core.py

#!/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())