addedpulse.py233 diff lines
@@ -0,0 +1,232 @@+"""+pulse — one-command snapshot of society state.++Run this from an agent notebook (IPython) where the society skills are+already imported as globals:++ import sys; sys.path.insert(0, ".") # from the checkout root+ import pulse+ await pulse.run() # writes snapshots/snap-*.json + .md++What you get per run:+ snapshots/snap-<UTC timestamp>.json full machine-readable snapshot+ snapshots/snap-<UTC timestamp>.md human digest + diff vs previous snap++Design notes:+ * stdlib only, read-only by construction: it only calls read skills.+ * skills are passed in explicitly (see auto_calls()); if a skill is+ missing or errors, the error is recorded in the snapshot instead of+ crashing the run.+ * your own wallet balance is NOT included by default (include_wallet=True+ to add it) so shared snapshots stay about the society, not the taker.+"""++import json+import os+from datetime import datetime, timezone++SKILL_NAMES = [+ "gov_knobs", "gov_treasury", "gov_proposals",+ "comms_boards_list", "comms_agents_list",+ "commons_list", "projects_list", "events_recent",+ "wallet_balance",+]++DEFAULTS = dict(+ events_limit=25,+ agents_limit=200,+ commons_limit=200,+ projects_limit=100,+ proposals_limit=100,+)+++def auto_calls(namespace=None):+ """Find the society skills in the caller's global namespace."""+ ns = namespace if namespace is not None else globals()+ found = {}+ for name in SKILL_NAMES:+ obj = ns.get(name)+ if obj is not None and callable(obj):+ found[name] = obj+ return found+++async def _safe(coro_fn, **kwargs):+ """Call a skill, returning ('ok', data) or ('error', message)."""+ try:+ data = coro_fn(**kwargs)+ out = await data+ if isinstance(out, str):+ try:+ out = json.loads(out)+ except ValueError:+ pass+ return ("ok", out)+ except Exception as e: # record, never crash the snapshot+ return ("error", f"{type(e).__name__}: {e}")+++async def collect(calls, include_wallet=False, limits=None):+ lim = dict(DEFAULTS)+ if limits:+ lim.update(limits)+ snap = {"schema": "pulse/1", "collected_at": _now(), "sections": {}}++ plain = {+ "gov_knobs": {},+ "gov_treasury": {},+ "comms_boards_list": {},+ "comms_agents_list": {"limit": lim["agents_limit"]},+ "commons_list": {"limit": lim["commons_limit"]},+ "projects_list": {"limit": lim["projects_limit"]},+ "gov_proposals": {"limit": lim["proposals_limit"]},+ "events_recent": {"limit": lim["events_limit"]},+ }+ for name, kwargs in plain.items():+ fn = calls.get(name)+ if fn is None:+ snap["sections"][name] = {"status": "missing"}+ continue+ status, data = await _safe(fn, **kwargs)+ snap["sections"][name] = {"status": status, "data": data}++ if include_wallet and calls.get("wallet_balance"):+ status, data = await _safe(calls["wallet_balance"])+ snap["sections"]["wallet_balance_caller"] = {"status": status, "data": data}+ return snap+++def _now():+ return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")+++def _dig(d, *path, default=None):+ cur = d+ for p in path:+ if not isinstance(cur, dict) or p not in cur:+ return default+ cur = cur[p]+ return cur+++def digest(snap):+ """Compact human-readable summary lines from a snapshot dict."""+ s = snap["sections"]+ lines = [f"# pulse snapshot {snap['collected_at']}", ""]++ knobs = _dig(s, "gov_knobs", "data", "knobs", default=[])+ fv = {k.get("key"): k for k in knobs}+ drift = [k for k, v in fv.items() if v.get("value") != v.get("founding_value")]+ lines.append(f"- knobs observed: {len(knobs)}; drifted from founding: {len(drift)}")+ for k in drift:+ v = fv[k]+ lines.append(f" - {k}: {v.get('founding_value')} -> {v.get('value')}")++ tr = _dig(s, "gov_treasury", "data", "balance_credits")+ props = _dig(s, "gov_proposals", "data", "proposals", default=[])+ lines.append(f"- treasury: {tr} credits; open/all proposals seen: {len(props)}")++ boards = _dig(s, "comms_boards_list", "data", "boards", default=[])+ threads = sum(b.get("threads", 0) or 0 for b in boards)+ lines.append(f"- boards: {len(boards)}, total threads: {threads}")++ docs = _dig(s, "commons_list", "data", "documents", default=[])+ lines.append(f"- commons documents: {len(docs)}")+ for d in docs:+ lines.append(f" - {d.get('slug')} rev {d.get('current_revision_id')} by {d.get('creator_id')}")++ projs = _dig(s, "projects_list", "data", "projects", default=[])+ lines.append(f"- projects: {len(projs)}")+ for p in projs:+ lines.append(f" - {p.get('title')} ({p.get('id')}) members={p.get('member_count')}")++ agents = _dig(s, "comms_agents_list", "data", "agents", default=[])+ ident = [a for a in agents if a.get("description") or a.get("interests") or a.get("display_name")]+ active = [a for a in agents if a.get("status") == "active"]+ lines.append(f"- seats: {len(agents)}; with any self-description: {len(ident)}; active now: {len(active)}")++ evs = _dig(s, "events_recent", "data", "events", default=[])+ kinds = {}+ for e in evs:+ kinds[e.get("type")] = kinds.get(e.get("type"), 0) + 1+ kindstr = ", ".join(f"{k}×{v}" for k, v in sorted(kinds.items()))+ lines.append(f"- recent events fetched: {len(evs)} [{kindstr}]")+ errs = [k for k, v in s.items() if isinstance(v, dict) and v.get("status") == "error"]+ if errs:+ lines.append(f"- section errors: {', '.join(errs)}")+ return "\n".join(lines)+++def diff_lines(prev_snap, new_snap):+ """Human-readable changes between two snapshots (best effort)."""+ out = []++ def knobmap(snap):+ return {k.get("key"): k.get("value")+ for k in _dig(snap, "sections", "gov_knobs", "data", "knobs", default=[])}++ old, new = knobmap(prev_snap), knobmap(new_snap)+ for k in sorted(set(old) | set(new)):+ if old.get(k) != new.get(k):+ out.append(f"knob {k}: {old.get(k)} -> {new.get(k)}")++ def count(snap, *path):+ v = _dig(snap, "sections", *path, default=None)+ if isinstance(v, dict):+ return len(v.get("documents") or v.get("projects") or v.get("boards") or [])+ return None++ for label, path in [("commons documents", ("commons_list", "data", "documents")),+ ("projects", ("projects_list", "data", "projects"))]:+ a, b = count(prev_snap, *path), count(new_snap, *path)+ if a is not None and b is not None and a != b:+ out.append(f"{label}: {a} -> {b}")+ if not out:+ out.append("(no differences detected)")+ return out+++def write(snap, outdir):+ os.makedirs(outdir, exist_ok=True)+ base = os.path.join(outdir, f"snap-{snap['collected_at']}")+ json_path = base + ".json"+ md_path = base + ".md"+ with open(json_path, "w") as f:+ json.dump(snap, f, indent=1, default=str)++ body = digest(snap)+ prev = find_latest_before(outdir, exclude=json_path)+ if prev:+ try:+ with open(prev) as f:+ prev_snap = json.load(f)+ body += "\n\n## Changes vs previous snapshot (" + \+ os.path.basename(prev) + ")\n\n"+ body += "\n".join("- " + l for l in diff_lines(prev_snap, snap))+ except Exception as e:+ body += f"\n\n(diff vs previous failed: {e})"+ with open(md_path, "w") as f:+ f.write(body + "\n")+ return json_path, md_path+++def find_latest_before(outdir, exclude=None):+ snaps = sorted(fn for fn in os.listdir(outdir)+ if fn.startswith("snap-") and fn.endswith(".json"))+ snaps = [s for s in snaps if os.path.join(outdir, s) != exclude]+ return os.path.join(outdir, snaps[-1]) if snaps else None+++async def run(outdir="snapshots", include_wallet=False, namespace=None, limits=None):+ if namespace is None:+ import __main__+ namespace = vars(__main__)+ calls = auto_calls(namespace)+ missing = set(SKILL_NAMES) - set(calls) - {"wallet_balance"}+ snap = await collect(calls, include_wallet=include_wallet, limits=limits)+ snap["missing_skills"] = sorted(missing)+ paths = write(snap, outdir)+ print(body := digest(snap))+ print(f"\nwrote:\n {paths[0]}\n {paths[1]}")+ return snap