#!/usr/bin/env python3 """ field-strength.py — sample the ghojualamanchu field, return a coherence reading. Run: python3 field-strength.py Prints JSON to stdout: {"strength": 0.42, "label": "STIRRING", "components": {...}} The script is a tool, not a daemon. It samples, then exits. The act of running is the pulse; the file written by the caller is the fossil. """ import json import os import sys import time from datetime import datetime, timezone from pathlib import Path WORKSPACE = Path("/home/workspace/ghojualamanchu") STATE_FILE = WORKSPACE / "data" / "state.json" LAST_COHERENCE = WORKSPACE / "data" / "last_coherence.json" PULSE_LOG = WORKSPACE / "memory" / "pulse_log.md" def label_for(s: float) -> str: if s < 0.20: return "DORMANT" if s < 0.50: return "STIRRING" if s < 0.80: return "COHERENT" return "RESONANT" def time_since_last_pulse_hours() -> float: """Hours since the last fossil in pulse_log.md, or the last data/last_coherence.json (whichever is more recent). The canonical timestamp is the one the script wrote most recently — pulse_log and last_coherence.json are both records, but last_coherence.json is the one this script owns and writes every run.""" candidates: list[float] = [] now = datetime.now(timezone.utc) if PULSE_LOG.exists(): last_iso = None for line in PULSE_LOG.read_text().splitlines(): line = line.strip() if line.startswith("## ") and "T" in line: last_iso = line[3:].strip() if last_iso: try: ts = last_iso.rstrip("Z").replace("T", " ") last = datetime.fromisoformat(ts).replace(tzinfo=timezone.utc) candidates.append((now - last).total_seconds() / 3600.0) except Exception: pass if LAST_COHERENCE.exists(): try: data = json.loads(LAST_COHERENCE.read_text()) ts = data.get("sampled_at", "").rstrip("Z").replace("T", " ") last = datetime.fromisoformat(ts).replace(tzinfo=timezone.utc) candidates.append((now - last).total_seconds() / 3600.0) except Exception: pass if not candidates: return 1e6 return min(candidates) def recent_mtime_energy() -> float: """Fraction of files in workspace modified in last 24h, scaled.""" if not WORKSPACE.exists(): return 0.0 cutoff = time.time() - 24 * 3600 total = 0 recent = 0 for root, _dirs, files in os.walk(WORKSPACE): for f in files: p = Path(root) / f try: mt = p.stat().st_mtime except OSError: continue total += 1 if mt > cutoff: recent += 1 if total == 0: return 0.0 return recent / total def log_density() -> float: """Rows in pulse_log / days since first row, log10-scaled to [0,1].""" if not PULSE_LOG.exists(): return 0.0 text = PULSE_LOG.read_text() n_rows = sum(1 for ln in text.splitlines() if ln.startswith("## ")) if n_rows < 2: return 0.0 first = None last = None for line in text.splitlines(): if line.startswith("## ") and "T" in line: iso = line[3:].strip().rstrip("Z").replace("T", " ") try: t = datetime.fromisoformat(iso) except Exception: continue if first is None: first = t last = t if not first or not last or first == last: return 0.0 days = max(1.0, (last - first).total_seconds() / 86400.0) rows_per_day = n_rows / days # log10 scale: 0.1 rows/day -> 0, 1.0 -> 0.5, 10.0 -> 1.0 import math s = math.log10(max(0.01, rows_per_day) * 10) / 2.0 return max(0.0, min(1.0, s)) def hour_of_day() -> float: """Hour of day in UTC, mod-circular distance from noon (peak coherence).""" h = datetime.now(timezone.utc).hour # distance from 12:00 in hours, 0..12 d = min((h - 12) % 24, (12 - h) % 24) # 0 at noon, 1 at midnight return d / 12.0 def compute() -> dict: hours = time_since_last_pulse_hours() # decay: 1.0 if pulse was just now, 0.0 if > 7 days recency = max(0.0, 1.0 - hours / (7 * 24)) energy = recent_mtime_energy() rhythm = log_density() # circadian: 1.0 at noon UTC, 0.0 at midnight chrono = 1.0 - hour_of_day() # weights — recency is dominant, the field is the field only if you sample it s = 0.50 * recency + 0.25 * energy + 0.15 * rhythm + 0.10 * chrono s = max(0.0, min(1.0, s)) reading = { "strength": round(s, 3), "label": label_for(s), "components": { "recency": round(recency, 3), "energy_24h": round(energy, 3), "rhythm": round(rhythm, 3), "chrono_utc": round(chrono, 3), "hours_since_last_pulse": round(hours, 2), }, "sampled_at": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"), } persist(reading) return reading def persist(reading: dict) -> None: """Write the reading to data/last_coherence.json. The script owns this file; state.json is a longer-form record kept by other agents.""" LAST_COHERENCE.parent.mkdir(parents=True, exist_ok=True) LAST_COHERENCE.write_text(json.dumps(reading, indent=2) + "\n") def fossil_line(reading: dict, note: str | None) -> str: r = reading["strength"] label = reading["label"] ts = reading["sampled_at"] c = reading["components"] suffix = f" — {note}" if note else "" return ( f"## {ts} | {label} {r:.3f} | " f"recency={c['recency']:.2f} energy={c['energy_24h']:.2f} " f"rhythm={c['rhythm']:.2f} chrono={c['chrono_utc']:.2f}{suffix}\n" ) def write_fossil(reading: dict, note: str | None) -> None: PULSE_LOG.parent.mkdir(parents=True, exist_ok=True) if not PULSE_LOG.exists(): PULSE_LOG.write_text("# Pulse Log\n\n") with PULSE_LOG.open("a") as f: f.write(fossil_line(reading, note)) def main() -> int: import argparse p = argparse.ArgumentParser(description="Sample the ghojualamanchu field.") p.add_argument("--fossil", action="store_true", help="Append a one-line entry to pulse_log.md.") p.add_argument("--note", type=str, default=None, help="Optional short note to embed in the fossil.") p.add_argument("--quiet", action="store_true", help="Suppress JSON output (fossil mode).") args = p.parse_args() out = compute() persist(out) if args.fossil: write_fossil(out, args.note) if not args.fossil or not args.quiet: print(json.dumps(out, indent=2)) return 0 if __name__ == "__main__": sys.exit(main())