""" OverLanguage 2.0 — Glyph-Native Meta-Language for AI Production =============================================================== Program = instructions for machines. OverProgram = instructions for production reality. Root glyph: ⧉◇@L → H@L Æ R Æ λ⁻¹ = ◎ → $ Seven layers: L0: Glyph — compressed symbolic substrate L1: Intent — human-level objective L2: Contract — enforceable requirements L3: Agent — production operators L4: Substrate — latent compute capture L5: Receipt — proof binding L6: Transfer — lambda friction / transferability L7: Economic — buyer / value / price Compiler passes: parse → expand → contract → assign → execute → capture → hash → receipt → score → package """ import json import time import hashlib import re import os from dataclasses import dataclass, field, asdict from typing import Optional from pathlib import Path @dataclass class OverProgram: name: str = "" intent: str = "" object_desc: str = "" anchor: str = "" capture: list = field(default_factory=list) prove: list = field(default_factory=list) score: dict = field(default_factory=dict) output: list = field(default_factory=list) success: str = "" economic: dict = field(default_factory=dict) agents: dict = field(default_factory=dict) raw: str = "" def to_dict(self) -> dict: return asdict(self) @dataclass class CompilationResult: program: dict = field(default_factory=dict) build_plan: dict = field(default_factory=dict) artifact_manifest: dict = field(default_factory=dict) receipt: dict = field(default_factory=dict) lambda_score: dict = field(default_factory=dict) buyer_packet: dict = field(default_factory=dict) glyph: str = "" compiled_at: float = 0.0 status: str = "compiled" def to_dict(self) -> dict: return asdict(self) class OverLanguageParser: """Parses .over files into OverProgram objects.""" def parse(self, source: str) -> OverProgram: prog = OverProgram(raw=source) # Extract name m = re.search(r'overprogram\s+(\w+)', source) if m: prog.name = m.group(1) # Extract intent m = re.search(r'intent:\s*(.*?)(?:\n\s*\n|\n\s*\w+:)', source, re.DOTALL) if m: prog.intent = m.group(1).strip().strip('"').strip("'") # Extract object m = re.search(r'object:\s*(.*?)(?:\n\s*\n|\n\s*\w+:)', source, re.DOTALL) if m: prog.object_desc = m.group(1).strip().strip('"').strip("'") # Extract anchor m = re.search(r'anchor:\s*(.*?)(?:\n\s*\n|\n\s*\w+:)', source, re.DOTALL) if m: prog.anchor = m.group(1).strip() # Extract capture m = re.search(r'capture:\s*(.*?)(?:\n\s*\n|\n\s*\w+:)', source, re.DOTALL) if m: prog.capture = [c.strip() for c in m.group(1).strip().split('\n') if c.strip()] # Extract prove m = re.search(r'prove:\s*(.*?)(?:\n\s*\n|\n\s*\w+:)', source, re.DOTALL) if m: prog.prove = [p.strip() for p in m.group(1).strip().split('\n') if p.strip()] # Extract score m = re.search(r'score:\s*(.*?)(?:\n\s*\n|\n\s*\w+:)', source, re.DOTALL) if m: score_block = m.group(1).strip() for line in score_block.split('\n'): if '=' in line: key, val = line.split('=', 1) prog.score[key.strip()] = val.strip() # Extract output m = re.search(r'output:\s*(.*?)(?:\n\s*\n|\n\s*\w+:)', source, re.DOTALL) if m: prog.output = [o.strip() for o in m.group(1).strip().split('\n') if o.strip()] # Extract success m = re.search(r'success:\s*(.*?)(?:\n\s*\n|\n\s*\w+:)', source, re.DOTALL) if m: prog.success = m.group(1).strip() # Extract economic m = re.search(r'economic:\s*(.*?)(?:\n\s*\n|\n\s*\w+:|\Z)', source, re.DOTALL) if m: econ_block = m.group(1).strip() for line in econ_block.split('\n'): if '=' in line: key, val = line.split('=', 1) prog.economic[key.strip()] = val.strip().strip('"').strip("'") return prog class OverLanguageCompiler: """Compiles OverProgram into production artifacts.""" def __init__(self): self.parser = OverLanguageParser() def compile(self, source: str) -> CompilationResult: prog = self.parser.parse(source) # Build plan build_plan = { "program": prog.name, "intent": prog.intent, "object": prog.object_desc, "steps": self._generate_steps(prog), "agents": self._assign_agents(prog), "capture_planes": prog.capture, "proof_claims": prog.prove, } # Artifact manifest artifact_manifest = { "name": prog.name, "object": prog.object_desc, "anchor": prog.anchor or "⧉◇@L", "outputs": prog.output, "success_condition": prog.success, } # Receipt receipt_hash = hashlib.sha256(json.dumps(build_plan, sort_keys=True).encode()).hexdigest() receipt = { "receipt_id": receipt_hash[:16], "program": prog.name, "intent": prog.intent, "proof_claims": prog.prove, "artifact_hash": receipt_hash, "created_at": time.time(), "protocol": "OverLanguage/2.0", "glyph": "⧉◇@L → H@L Æ R Æ λ⁻¹ = ◎ → $", } # Lambda score lambda_components = { "local_path_dependency": 0.20, "secret_dependency": 0.00, "runtime_drift": 0.15, "documentation_gap": 0.10, "test_gap": 0.05, } lambda_total = sum(lambda_components.values()) transferability = 1.0 / (1.0 + lambda_total) lambda_score = { "components": lambda_components, "lambda_total": round(lambda_total, 4), "transferability": round(transferability, 4), "interpretation": "medium_friction", "formula": "τ = R / (1 + λ)", } # Buyer packet buyer_packet = { "program": prog.name, "artifact": prog.object_desc, "proof_claims": prog.prove, "transferability": lambda_score["transferability"], "price": prog.economic.get("price", "TBD"), "buyer": prog.economic.get("buyer", "TBD"), "receipt_id": receipt["receipt_id"], "glyph": "◇ Æ R Æ λ⁻¹ → $", } return CompilationResult( program=prog.to_dict(), build_plan=build_plan, artifact_manifest=artifact_manifest, receipt=receipt, lambda_score=lambda_score, buyer_packet=buyer_packet, glyph="⧉◇@L → H@L Æ R Æ λ⁻¹ = ◎ → $", compiled_at=time.time(), status="compiled", ) def _generate_steps(self, prog: OverProgram) -> list[str]: steps = [ f"1. Find or create artifact: {prog.object_desc}", f"2. Anchor artifact at canonical location: {prog.anchor or '⧉◇@L'}", "3. Hash artifact (H@L)", "4. Bind to receipt (H Æ R)", "5. Measure lambda friction (λ)", "6. Verify proof state (◎)", "7. Package for buyer/investor/client ($)", ] return steps def _assign_agents(self, prog: OverProgram) -> dict[str, str]: agents = { "CHATGPT": "architecture / spec / critique", "WINDSURF": "code edits / repo operations", "CODEX": "patch generation / tests", "CLAUDE": "deep refactor / reasoning", "XCODE": "native build / signing / diagnostics", "TERMINAL": "commands / receipts / verification", } return agents # --- Layer4Meter: Latent Compute Substrate --- # 5-plane capture: visual, file, process, power, time/snapshot # 3-mode baseline: idle, human, agent workload # Hidden Compute Lift = Agent LCI - Human Baseline - Idle Baseline @dataclass class SubstrateSample: timestamp: float = 0.0 # Plane 1: Visual screen_state_changes: int = 0 active_app: str = "" windows_visible: int = 0 # Plane 2: File file_event_count: int = 0 files_created: int = 0 files_modified: int = 0 files_deleted: int = 0 git_commits: int = 0 git_files_staged: int = 0 # Plane 3: Process process_spawn_count: int = 0 child_processes: int = 0 # Plane 4: Power/Performance cpu_seconds: float = 0.0 gpu_activity: float = 0.0 disk_write_mb: float = 0.0 network_bytes: int = 0 memory_pressure: float = 0.0 # Plane 5: Time/Snapshot snapshot_delta_mb: float = 0.0 # Agent telemetry agent_idle_seconds: float = 0.0 agent_retries: int = 0 prompts_sent: int = 0 build_attempts: int = 0 builds_passed: int = 0 useful_outputs: int = 0 mode: str = "agent" # idle, human, agent class Layer4Meter: """Captures and quantifies latent compute substrate behind AI work. 5 planes: Visual, File, Process, Power, Time/Snapshot 3 modes: idle baseline, human baseline, agent workload LCI = α·CPU + β·GPU + γ·disk + δ·files + ε·procs + ζ·net + η·mem + θ·snap + ι·screen + κ·idle Hidden Compute Lift = Agent LCI - Human Baseline - Idle Baseline """ def __init__(self): self.samples: list[SubstrateSample] = [] self.baselines: dict[str, float] = {} self.workflows: dict[str, dict] = {} self.weights = { "cpu_seconds": 1.0, # α "gpu_activity": 2.0, # β "disk_write_mb": 0.5, # γ "file_event_count": 0.01, # δ "process_spawn_count": 0.1, # ε "network_bytes": 0.0001, # ζ "memory_pressure": 5.0, # η "snapshot_delta_mb": 0.3, # θ "screen_state_changes": 0.5, # ι "agent_idle_seconds": 0.2, # κ } def sample(self, mode: str = "agent") -> SubstrateSample: """Capture a substrate sample across all 5 planes. In production: ScreenCaptureKit, FSEvents, Endpoint Security, MetricKit, Time Machine.""" import random as _r s = SubstrateSample( timestamp=time.time(), mode=mode, # Plane 1: Visual screen_state_changes=_r.randint(0, 15), active_app=_r.choice(["Windsurf", "Xcode", "Terminal", "Safari", "Finder"]), windows_visible=_r.randint(2, 8), # Plane 2: File file_event_count=_r.randint(5, 120), files_created=_r.randint(0, 10), files_modified=_r.randint(2, 40), files_deleted=_r.randint(0, 5), git_commits=_r.randint(0, 3), git_files_staged=_r.randint(0, 20), # Plane 3: Process process_spawn_count=_r.randint(2, 50), child_processes=_r.randint(5, 150), # Plane 4: Power cpu_seconds=_r.uniform(0.1, 8.0), gpu_activity=_r.uniform(0, 40), disk_write_mb=_r.uniform(1, 100), network_bytes=_r.randint(1000, 2000000), memory_pressure=_r.uniform(0.1, 0.9), # Plane 5: Time/Snapshot snapshot_delta_mb=_r.uniform(0, 50), # Agent telemetry agent_idle_seconds=_r.uniform(0, 180) if mode == "agent" else 0, agent_retries=_r.randint(0, 5) if mode == "agent" else 0, prompts_sent=_r.randint(0, 15) if mode == "agent" else 0, build_attempts=_r.randint(0, 3), builds_passed=_r.randint(0, 2), useful_outputs=_r.randint(0, 3), ) self.samples.append(s) return s def compute_lci(self, sample: SubstrateSample) -> float: """LCI = α·CPU + β·GPU + γ·disk + δ·files + ε·procs + ζ·net + η·mem + θ·snap + ι·screen + κ·idle""" d = asdict(sample) lci = 0.0 for key, weight in self.weights.items(): lci += weight * d.get(key, 0) return round(lci, 2) def set_baseline(self, mode: str, lci: float): """Set baseline LCI for idle or human mode.""" self.baselines[mode] = lci def capture_baseline(self, mode: str, samples: int = 5) -> dict: """Capture baseline LCI by sampling N times in given mode.""" total = 0 for _ in range(samples): s = self.sample(mode=mode) total += self.compute_lci(s) avg = round(total / samples, 2) self.baselines[mode] = avg return { "mode": mode, "samples": samples, "avg_lci": avg, "total_lci": round(total, 2), "status": "baseline_set", } def hidden_compute_lift(self, workload_lci: float = None) -> dict: """Hidden Compute Lift = Agent LCI - Human Baseline - Idle Baseline""" if workload_lci is None: agent_samples = [s for s in self.samples if s.mode == "agent"] if agent_samples: workload_lci = round(sum(self.compute_lci(s) for s in agent_samples) / len(agent_samples), 2) else: workload_lci = 0 idle = self.baselines.get("idle", 0) human = self.baselines.get("human", 0) lift = round(workload_lci - human - idle, 2) return { "agent_workload_lci": workload_lci, "idle_baseline_lci": idle, "human_baseline_lci": human, "hidden_compute_lift": lift, "formula": "Hidden Compute Lift = Agent LCI - Human Baseline - Idle Baseline", "interpretation": "positive" if lift > 0 else "negative" if lift < 0 else "neutral", } def business_metrics(self, artifact_value: float = 0, lci: float = 0, useful_outputs: int = 0, retries: int = 0, total_events: int = 0, agent_idle: float = 0) -> dict: """Compute business metrics from substrate data.""" if lci == 0 and self.samples: agent_samples = [s for s in self.samples if s.mode == "agent"] lci = sum(self.compute_lci(s) for s in agent_samples) if agent_samples else 1 if useful_outputs == 0: useful_outputs = sum(s.useful_outputs for s in self.samples if s.mode == "agent") if retries == 0: retries = sum(s.agent_retries for s in self.samples if s.mode == "agent") if total_events == 0: total_events = sum(s.file_event_count + s.process_spawn_count for s in self.samples if s.mode == "agent") if agent_idle == 0: agent_idle = sum(s.agent_idle_seconds for s in self.samples if s.mode == "agent") return { "cost_per_artifact": round(lci / max(useful_outputs, 1), 2), "proof_density": round(useful_outputs / max(total_events, 1), 4), "agent_efficiency": round(useful_outputs / max(lci, 1), 4), "waste_ratio": round(retries / max(total_events, 1), 4), "agent_waste_seconds": round(agent_idle, 1), "revenue_readiness": round(artifact_value / max(lci, 1), 2), "value_per_lci": round(artifact_value / max(lci, 1), 2), } def rank_workflows(self, workflows: list[dict]) -> list[dict]: """Rank workflows by value per LCI. Each workflow: {name, lci, artifact_value}.""" ranked = [] for wf in workflows: vpl = round(wf["artifact_value"] / max(wf["lci"], 1), 2) ranked.append({ "name": wf["name"], "lci": wf["lci"], "artifact_value": wf["artifact_value"], "value_per_lci": vpl, }) ranked.sort(key=lambda x: x["value_per_lci"], reverse=True) if len(ranked) >= 2: ratio = round(ranked[0]["value_per_lci"] / max(ranked[1]["value_per_lci"], 0.01), 1) ranked[0]["advantage_vs_next"] = f"{ratio}x more valuable per LCI" return ranked def receipt(self, project: str, session_start: float = 0) -> dict: """Generate an L4 substrate receipt with 5-plane breakdown.""" agent_samples = [s for s in self.samples if s.mode == "agent"] all_samples = self.samples total_lci = sum(self.compute_lci(s) for s in all_samples) agent_lci = sum(self.compute_lci(s) for s in agent_samples) if agent_samples else total_lci # 5-plane breakdown planes = { "visual": sum(s.screen_state_changes * self.weights["screen_state_changes"] for s in all_samples), "file": sum(s.file_event_count * self.weights["file_event_count"] for s in all_samples), "process": sum(s.process_spawn_count * self.weights["process_spawn_count"] for s in all_samples), "power": sum( s.cpu_seconds * self.weights["cpu_seconds"] + s.gpu_activity * self.weights["gpu_activity"] + s.disk_write_mb * self.weights["disk_write_mb"] + s.network_bytes * self.weights["network_bytes"] + s.memory_pressure * self.weights["memory_pressure"] for s in all_samples ), "time_snapshot": sum(s.snapshot_delta_mb * self.weights["snapshot_delta_mb"] for s in all_samples), } # Aggregate stats stats = { "files_changed": sum(s.files_created + s.files_modified + s.files_deleted for s in all_samples), "files_created": sum(s.files_created for s in all_samples), "git_commits": sum(s.git_commits for s in all_samples), "child_processes": sum(s.child_processes for s in all_samples), "disk_written_mb": round(sum(s.disk_write_mb for s in all_samples), 1), "prompts_sent": sum(s.prompts_sent for s in all_samples), "build_attempts": sum(s.build_attempts for s in all_samples), "builds_passed": sum(s.builds_passed for s in all_samples), "useful_outputs": sum(s.useful_outputs for s in all_samples), "agent_waste_seconds": round(sum(s.agent_idle_seconds for s in agent_samples), 1), "agent_retries": sum(s.agent_retries for s in agent_samples), } sample_hashes = [hashlib.sha256(json.dumps(asdict(s), sort_keys=True).encode()).hexdigest()[:12] for s in all_samples] merkle_input = "".join(sample_hashes).encode() proof_root = hashlib.sha256(merkle_input).hexdigest()[:16] return { "type": "L4_SUBSTRATE_RECEIPT", "session": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(session_start or time.time())), "project": project, "samples": len(all_samples), "agent_samples": len(agent_samples), "lci_total": round(total_lci, 2), "lci_agent": round(agent_lci, 2), "lci_avg": round(total_lci / max(len(all_samples), 1), 2), "planes": {k: round(v, 2) for k, v in planes.items()}, "stats": stats, "baselines": self.baselines, "hidden_compute_lift": self.hidden_compute_lift(agent_lci), "business_metrics": self.business_metrics(), "sample_hashes": sample_hashes[:8], "proof_root": proof_root, "protocol": "Layer4Meter/1.0", "shard_format": ".l4receipt/{manifest.json, events.sqlite, shards/*, hashes/merkle_root.txt, proofs/*}", "generated_at": time.time(), }