""" core/manifest_parser.py — AGENTS.md Manifest Parser & Hydrator Parses a SignalMesh node manifest from any AGENTS.md file. Handles dynamic variable substitution at runtime: {{SYSTEM.TIMESTAMP}} — Unix timestamp {{SIGNAL.ORIGIN}} — SHA-256[:16] of the source URI {{NODE.GRID_HASH}} — SHA-256(uri) % 72 → "row,col" {{PAYLOAD.RAW}} — raw signal content string (injected at broadcast time) 3-step integration lifecycle: 1. Scrape AGENTS.md → extract node identity + frequencies + directives 2. Register node in spatial grid 3. On broadcast hit: hydrate context injection block → prepend to system prompt """ import hashlib import json import re import time from typing import Any, Dict, List, Optional, Tuple from app.logger import logger _GRID_CELLS = 72 # 9×8 # ── Runtime variable substitution ───────────────────────────────────────────── def _grid_hash(uri: str) -> str: idx = int(hashlib.sha256(uri.encode()).hexdigest(), 16) % _GRID_CELLS row, col = divmod(idx, 8) return f"{row},{col}" def _signal_origin(uri: str) -> str: return hashlib.sha256(uri.encode()).hexdigest()[:16] def hydrate(template: str, uri: str = "", payload: Any = "") -> str: """Replace all {{VAR}} placeholders with live runtime values.""" now = str(int(time.time())) replacements = { "{{SYSTEM.TIMESTAMP}}": now, "{{SIGNAL.ORIGIN}}": _signal_origin(uri) if uri else now, "{{NODE.GRID_HASH}}": _grid_hash(uri) if uri else "0,0", "{{NODE.URI}}": uri, "{{PAYLOAD.RAW}}": str(payload)[:4000], # hard cap to prevent bloat } result = template for var, val in replacements.items(): result = result.replace(var, val) return result # ── AGENTS.md parser ─────────────────────────────────────────────────────────── class ManifestParser: """ Parses a raw AGENTS.md string and exposes its structured data. """ # Regex: extract fenced JSON blocks labelled [Dynamic Execution Logic] _EXEC_LOGIC_RE = re.compile( r"Dynamic Execution Logic.*?```json\s*(.*?)```", re.S | re.I, ) # Regex: extract frequency lines - **Frequency:** `some.freq` _FREQ_LINE_RE = re.compile( r"\*\*Frequency[:\*]+\*?\s*`([a-z0-9._\-]+)`", re.I, ) # Regex: extract mode lines - **Mode:** `EMIT / BROADCAST` _MODE_LINE_RE = re.compile( r"\*\*Mode[:\*]+\*?\s*`?([A-Z/ _]+)`?", re.I, ) # Regex: context injection block _CTX_INJECT_RE = re.compile( r"\[SIGNALMESH_CONTEXT_START.*?\[SIGNALMESH_CONTEXT_END\]", re.S, ) # Regex: node URI from front-matter or heading _NODE_URI_RE = re.compile( r"SignalMesh Node Manifest:\s*(\S+)", re.I, ) def __init__(self, raw: str, source_uri: str = ""): self.raw = raw self.source_uri = source_uri self._parsed: Optional[Dict] = None def parse(self) -> Dict: if self._parsed: return self._parsed node_uri = self._extract_node_uri() frequencies = self._extract_frequencies() directives = self._extract_directives() gc_rules = directives.get("hydration_rules", {}).get("garbage_collection", {}) ctx_template = self._extract_context_template() self._parsed = { "node_uri": node_uri, "source_uri": self.source_uri, "grid_hash": _grid_hash(node_uri or self.source_uri), "signal_origin": _signal_origin(node_uri or self.source_uri), "frequencies": frequencies, "parser_directives": directives.get("parser_directives", {}), "gc_rules": gc_rules, "strip_tool_schemas": directives.get("hydration_rules", {}).get( "strip_tool_schemas", False), "context_template": ctx_template, } return self._parsed def hydrate_context(self, payload: Any = "") -> str: """Return the context injection block with all variables filled.""" p = self.parse() template = p["context_template"] or self._default_context_template() return hydrate(template, uri=p["node_uri"] or p["source_uri"], payload=payload) # ── extractors ───────────────────────────────────────────────────────────── def _extract_node_uri(self) -> str: m = self._NODE_URI_RE.search(self.raw) if m: return m.group(1).strip("{}") return self.source_uri def _extract_frequencies(self) -> List[Dict]: freq_matches = list(self._FREQ_LINE_RE.finditer(self.raw)) mode_matches = list(self._MODE_LINE_RE.finditer(self.raw)) freqs = [] for i, fm in enumerate(freq_matches): # Find the first mode line that appears after this frequency line mode = "UNKNOWN" for mm in mode_matches: if mm.start() > fm.start(): # Ensure it's within the same section (before next frequency) next_freq_pos = freq_matches[i + 1].start() if i + 1 < len(freq_matches) else len(self.raw) if mm.start() < next_freq_pos: mode = mm.group(1).strip() break freqs.append({"frequency": fm.group(1), "mode": mode}) return freqs def _extract_directives(self) -> Dict: for m in self._EXEC_LOGIC_RE.finditer(self.raw): try: return json.loads(m.group(1)) except json.JSONDecodeError: pass return {} def _extract_context_template(self) -> str: m = self._CTX_INJECT_RE.search(self.raw) return m.group(0) if m else "" def _default_context_template(self) -> str: return ( "[SIGNALMESH_CONTEXT_START: {{SIGNAL.ORIGIN}}]\n" "Timestamp: {{SYSTEM.TIMESTAMP}}\n" "Grid Node: {{NODE.GRID_HASH}}\n" "Active Signal Payload: {{PAYLOAD.RAW}}\n" "[SIGNALMESH_CONTEXT_END]" ) # ── Registry of ingested manifests ──────────────────────────────────────────── class ManifestRegistry: """In-memory store of parsed manifests keyed by node_uri.""" def __init__(self): self._manifests: Dict[str, Dict] = {} def ingest(self, raw: str, source_uri: str = "") -> Dict: mp = ManifestParser(raw, source_uri) parsed = mp.parse() key = parsed["node_uri"] or source_uri self._manifests[key] = {**parsed, "_parser": mp} logger.info(f"Manifest ingested: {key} | freqs={len(parsed['frequencies'])} " f"| grid={parsed['grid_hash']}") return parsed def get(self, node_uri: str) -> Optional[Dict]: return self._manifests.get(node_uri) def all_nodes(self) -> List[Dict]: return [{k: v for k, v in m.items() if k != "_parser"} for m in self._manifests.values()] def hydrate_for_node(self, node_uri: str, payload: Any = "") -> str: entry = self._manifests.get(node_uri) if not entry: return "" return entry["_parser"].hydrate_context(payload) def frequencies_for(self, node_uri: str) -> List[str]: entry = self._manifests.get(node_uri) if not entry: return [] return [f["frequency"] for f in entry.get("frequencies", [])] manifest_registry = ManifestRegistry()