vh
a35ed7af19
Deployed gap-injection run closed by operator steer 2026-07-04: R30 graduates on offline-tests + human face-validity, no deployed harness build, no endpoint. Harness (read/predict/record; write side stubbed) + brokkr's R30.10 protocol pin (de8357f) banked as drop-in for the parked powered true-tau perceptual study. predict() self-validated against brokkr's N=0 anchors (1h/10h/1wk).
111 lines
4.4 KiB
Python
111 lines
4.4 KiB
Python
#!/usr/bin/env python3
|
||
"""R30 gap-injection probe harness (ratatoskr instrument).
|
||
|
||
Throwaway probe per the R30.10 protocol
|
||
brokkr-smithy/research/R30-ocean-derived-mood-dynamics/empirics/r30-gap-injection-protocol.md (de8357f)
|
||
|
||
Deliverable: the (agent, N, Δt, observed p/a/d) table + per-agent baseline_pad + the
|
||
frozen start vector, handed to brokkr who runs D1/D2/D3 graduation. This harness does NOT
|
||
own the verdict; the predicted/residual columns are a sanity aid only.
|
||
|
||
READ + PREDICT + RECORD are fixed by the spec. freeze_start() + backdate() are the WRITE
|
||
side and are STUBBED pending worldtree's mechanism on personal :8081 (direct persona_mood
|
||
DB write vs a test affordance). Data lands on a diag/ branch; NOT ratatoskr production code.
|
||
"""
|
||
import json
|
||
import math
|
||
import os
|
||
import urllib.request
|
||
import urllib.error
|
||
|
||
BASE = os.environ.get("WORLDTREE_API_URL", "http://10.250.50.152:8081")
|
||
KEY = os.environ["WORLDTREE_API_KEY"]
|
||
|
||
# --- R30.10 spec constants (FIXED) ---
|
||
TAU_BASE_S = 36000.0 # τ_base = 10h
|
||
BETA_N = 0.25 # τ_P = τ_base · exp(β_N · N)
|
||
R_A = 1.9 # τ_A = τ_P / r_A ; τ_D = τ_P
|
||
FROZEN_START = {"pleasure": -0.60, "arousal": +0.50, "dominance": -0.30}
|
||
DT_GRID = [0, 60, 600, 3600, 14400, 36000, 86400, 604800] # s
|
||
AXES = ("pleasure", "arousal", "dominance")
|
||
EPS = 1e-3 # D1 tolerance (PAD units)
|
||
|
||
|
||
def tau(axis, N):
|
||
tau_p = TAU_BASE_S * math.exp(BETA_N * N)
|
||
return {"pleasure": tau_p, "arousal": tau_p / R_A, "dominance": tau_p}[axis]
|
||
|
||
|
||
def retention(axis, dt, N):
|
||
"""ρ_k(Δt,N) = exp(−Δt / τ_k(N)) — predicted deviation-retention fraction."""
|
||
return math.exp(-dt / tau(axis, N))
|
||
|
||
|
||
def _get(path):
|
||
req = urllib.request.Request(BASE + path, headers={"Authorization": f"Bearer {KEY}"})
|
||
with urllib.request.urlopen(req, timeout=8) as r:
|
||
return json.load(r)
|
||
|
||
|
||
def read_pad(agent):
|
||
"""get_state read: returns (pad, baseline_pad). pad == peek_relaxed(now), OU-faded, non-mutating."""
|
||
d = _get(f"/agents/{agent}/persona_state")
|
||
return d["pad"], d["baseline_pad"]
|
||
|
||
|
||
# --- WRITE SIDE: STUBBED pending worldtree mechanism (msg 01KWNVHTFP…) ---
|
||
def freeze_start(agent, pad):
|
||
"""Set persona_mood pad directly to the displaced vector (no live appraisal)."""
|
||
raise NotImplementedError("worldtree mechanism pending: freeze displaced start mood pad")
|
||
|
||
|
||
def backdate(agent, dt_s):
|
||
"""Set persona_mood.updated_at = now − dt_s so peek_relaxed fades by exactly Δt."""
|
||
raise NotImplementedError("worldtree mechanism pending: back-date persisted updated_at")
|
||
|
||
|
||
def predict_selfcheck():
|
||
"""Validate retention() against brokkr's stated N=0 P-axis anchors (spec §2 D3)."""
|
||
print("predict() self-check — N=0 P-axis retention vs brokkr's stated anchors:")
|
||
expect = {"1h": (3600, 0.905), "10h(overnight)": (36000, 0.368), "1wk": (604800, 5e-8)}
|
||
ok = True
|
||
for label, (dt, want) in expect.items():
|
||
got = retention("pleasure", dt, 0.0)
|
||
match = abs(got - want) < (want * 0.01 + 1e-9) # 1% or float-floor
|
||
ok = ok and match
|
||
print(f" {label:>16}: got {got:.6g} expect {want:.6g} {'OK' if match else 'MISMATCH'}")
|
||
# per-axis τ at a couple N points (informational)
|
||
print("τ (hours) by axis × N:")
|
||
for N in (-0.6, 0.0, 0.8):
|
||
taus = {a: tau(a, N) / 3600 for a in AXES}
|
||
print(f" N={N:+.1f}: P={taus['pleasure']:.2f}h A={taus['arousal']:.2f}h D={taus['dominance']:.2f}h")
|
||
return ok
|
||
|
||
|
||
def run_cell(agent, N, dt, baseline):
|
||
"""One (agent, Δt) cell: freeze start → back-date by Δt → read faded pad → record row.
|
||
Blocked until freeze_start()/backdate() are wired to worldtree's mechanism."""
|
||
freeze_start(agent, FROZEN_START)
|
||
backdate(agent, dt)
|
||
pad, _ = read_pad(agent)
|
||
row = {"agent": agent, "N": N, "dt_s": dt}
|
||
for k in AXES:
|
||
obs = pad[k]
|
||
dev0 = abs(FROZEN_START[k] - baseline[k])
|
||
pred_dev = dev0 * retention(k, dt, N)
|
||
obs_dev = abs(obs - baseline[k])
|
||
row[f"obs_{k[0]}"] = obs
|
||
row[f"pred_dev_{k[0]}"] = pred_dev
|
||
row[f"resid_{k[0]}"] = obs_dev - pred_dev
|
||
return row
|
||
|
||
|
||
if __name__ == "__main__":
|
||
import sys
|
||
if len(sys.argv) > 1 and sys.argv[1] == "read":
|
||
agent = sys.argv[2]
|
||
pad, base = read_pad(agent)
|
||
print(json.dumps({"agent": agent, "pad": pad, "baseline_pad": base}, indent=2))
|
||
else:
|
||
predict_selfcheck()
|