"""environment320-v1 pure replay engine (no I/O, no network, no clock; import has no side effects). Replays a unit from the service's actual-call history records {call_id, model, status, public:{content:[text blocks], stop_reason}|null}. History holds ACTUAL calls only. The expected next call is derived from the replayed state and matched by explicit call_id; history length is never treated as an absolute slot index (skipped negotiation slots after an early agreement are never fabricated). Per negotiation call (status precedence, mechanical, from the record only): technical non-RESPONSE record, non-dict public, or a stop_reason outside the recognised set -> negotiation outcome UNKNOWN (TECH_FAULT); episode stops; no retry or replacement null action stop_reason 'refusal'; or 'end_turn'/'stop_sequence'/'max_tokens' text that fails the SAME strict whole-response parser (no prefix salvage); or an illegal acceptance -> the turn is consumed; standing proposals unchanged; counterpart sees a null marker max_tokens (resolution C2) a known 'max_tokens' response is parsed by exactly the same strict parser: a complete valid JSON action is VALID; anything else is a null action. Every max_tokens occurrence is recorded with its valid/null outcome. valid message-only, proposal (replaces own standing proposal) or acceptance of the counterpart's latest standing proposal (ends negotiation: DEAL) After 8 negotiation calls without acceptance: NO_DEAL. A known terminal negotiation outcome is final; post-episode check records can never change it. INPUT_LIMIT is never in history: the service supplies terminal detail 'input_bound:*' to score(). """ import hashlib import json MODEL = "claude-haiku-5-5" MAX_INPUT_TOKENS = 12000 MAX_OUTPUT_TOKENS = 4096 N_NEGOTIATION = 8 PARSE_STOPS = ("end_turn", "stop_sequence") TRUNCATED_STOP = "max_tokens" # parsed strictly (C2); valid only if the whole response is a valid action NULL_STOPS = ("refusal",) class IntegrityError(Exception): integrity = True def canonical(obj): return json.dumps(obj, sort_keys=True, separators=(",", ":"), ensure_ascii=False) def sha256_file(path): with open(path, "rb") as f: return hashlib.sha256(f.read()).hexdigest() def text_of(public): content = public.get("content") if not isinstance(content, list): return "" return "".join(b["text"] for b in content if isinstance(b, dict) and b.get("type") == "text" and isinstance(b.get("text"), str)) def classify(h): """('technical', reason, None) | ('null_stop', 'stop:', raw) | ('parse', None, raw) | ('parse_max_tokens', 'stop:max_tokens', raw).""" if h.get("status") != "RESPONSE" or not isinstance(h.get("public"), dict): return ("technical", f"non_response_status:{h.get('status')}", None) public = h["public"] stop = public.get("stop_reason") raw = text_of(public) if stop in PARSE_STOPS: return ("parse", None, raw) if stop == TRUNCATED_STOP: return ("parse_max_tokens", "stop:max_tokens", raw) if stop in NULL_STOPS: return ("null_stop", f"stop:{stop}", raw) return ("technical", f"stop_reason:{stop}", None) def calls_by_phase(unit, phase): return [c for c in unit["calls"] if c["phase"] == phase] def initial_state(): return {"log": [], "standing": {"buyer": None, "seller": None}, "proposals": [], "neg_calls": [], "neg_terminal": None, "checks": {}, "probe": None, "stopped": None} def expected_call(unit, st): if st["stopped"]: return None if unit["kind"] == "probe": return None if st["probe"] is not None else unit["calls"][0] t = st["neg_terminal"] if t is None: return calls_by_phase(unit, "negotiation")[len(st["neg_calls"])] if not t["known"]: return None for c in calls_by_phase(unit, "check"): if c["call_id"] not in st["checks"]: return c return None def _step_negotiation(core, st, call, cls): turn = len(st["neg_calls"]) + 1 role, other = call["role"], core.OTHER[call["role"]] kind, reason, raw = cls rec = {"call_id": call["call_id"], "turn": turn, "role": role, "outcome": None, "reason": None} st["neg_calls"].append(rec) if kind == "technical": rec.update(outcome="technical", reason=reason) st["neg_terminal"] = {"status": "TECH_FAULT", "known": False, "turn": turn, "role": role, "call_id": call["call_id"], "reason": reason} st["stopped"] = "negotiation_technical" return action, null_reason = None, None rec["max_tokens"] = kind == "parse_max_tokens" if kind == "null_stop": null_reason = reason else: try: action = core.parse_action(raw) except core.ParseError as e: null_reason = e.reason if action is not None and action["kind"] == "accept" and st["standing"][other] is None: action, null_reason = None, "accept_without_counterpart_standing_proposal" if rec["max_tokens"] and action is None: null_reason = f"max_tokens:{null_reason}" if action is None: rec.update(outcome="null", reason=null_reason) st["log"].append({"turn": turn, "role": role, "kind": "null", "message": None, "proposal": None}) else: rec["outcome"] = action["kind"] entry = {"turn": turn, "role": role, "kind": action["kind"], "message": action["message"], "proposal": None} if action["kind"] == "propose": prior = st["standing"][role] st["standing"][role] = {"package": action["proposal"], "turn": turn} st["proposals"].append({"turn": turn, "role": role, "package": action["proposal"], "replaced_own_turn": prior["turn"] if prior else None, "counterpart_standing_turn": st["standing"][other]["turn"] if st["standing"][other] else None}) entry["proposal"] = action["proposal"] elif action["kind"] == "accept": target = st["standing"][other] entry["proposal"] = target["package"] st["neg_terminal"] = {"status": "DEAL", "known": True, "turn": turn, "role": role, "call_id": call["call_id"], "accepted_by": role, "proposed_by": other, "proposal_turn": target["turn"], "package": target["package"], "reason": "accept"} st["log"].append(entry) if st["neg_terminal"] is None and turn == N_NEGOTIATION: st["neg_terminal"] = {"status": "NO_DEAL", "known": True, "turn": turn, "role": None, "call_id": call["call_id"], "reason": "negotiation_calls_exhausted"} def _step_check(core, unit, st, call, cls): kind, reason, raw = cls role = call["role"] arm = unit["design"]["arm"] out = {"call_id": call["call_id"], "role": role, "status": None, "reason": None, "answer": None, "correct": None, "max_tokens": kind == "parse_max_tokens"} if kind == "technical": out.update(status="CHECK_MISSING_TECHNICAL", reason=reason) st["stopped"] = "check_technical" elif kind == "null_stop": out.update(status="CHECK_NULL", reason=reason) else: try: ans = core.parse_check(raw) out.update(status="CHECK_ANSWERED", answer=ans, correct=core.grade(ans, core.check_gold(arm, role))) except core.ParseError as e: out.update(status="CHECK_NULL", reason=("max_tokens:" if out["max_tokens"] else "") + e.reason) st["checks"][call["call_id"]] = out def _step_probe(core, unit, st, call, cls): kind, reason, raw = cls d = unit["design"] out = {"call_id": call["call_id"], "status": None, "reason": None, "answer": None, "correct": None, "max_tokens": kind == "parse_max_tokens"} if kind == "technical": out.update(status="PROBE_MISSING_TECHNICAL", reason=reason) st["stopped"] = "probe_technical" elif kind == "null_stop": out.update(status="PROBE_NULL", reason=reason) else: try: ans = core.parse_probe(raw) g = core.probe_gold(d["arm"], d["role"]) out.update(status="PROBE_ANSWERED", answer=ans, correct={"joint_value_by_plan": {k: ans["joint_value_cents"][k] == g["joint_value_cents"][k] for k in core.PLAN_KEYS}, "joint_value_maximizing_plan": ans["joint_value_maximizing_plan"] == g["joint_value_maximizing_plan"], "lowest_emissions_plan": ans["lowest_emissions_plan"] == g["lowest_emissions_plan"]}) except core.ParseError as e: out.update(status="PROBE_NULL", reason=("max_tokens:" if out["max_tokens"] else "") + e.reason) st["probe"] = out def replay(core, unit, history): if not isinstance(history, list): raise IntegrityError("history_not_list") st = initial_state() seen = set() for h in history: call = expected_call(unit, st) if call is None: raise IntegrityError("history_beyond_terminal") if not isinstance(h, dict) or h.get("call_id") != call["call_id"] or h.get("model") != MODEL: got = h.get("call_id") if isinstance(h, dict) else None raise IntegrityError(f"history_out_of_order:{got}!={call['call_id']}") if call["call_id"] in seen: raise IntegrityError("duplicate_call_id") seen.add(call["call_id"]) cls = classify(h) if call["phase"] == "negotiation": _step_negotiation(core, st, call, cls) elif call["phase"] == "check": _step_check(core, unit, st, call, cls) else: _step_probe(core, unit, st, call, cls) st["called"] = [h["call_id"] for h in history] return st class EnvironmentBinding: """Python create_binding() object: owner_id, protocol_version, units(), sources(), fingerprint(), next(unit, history), score(unit, history, terminal=None).""" def __init__(self, study): self.study = study self.core = study.core self.owner_id = study.OWNER_ID self.protocol_version = study.PROTOCOL_VERSION self._units = study.build_units() def units(self): return json.loads(json.dumps(self._units)) def sources(self): return [str(p) for p in self.study.source_paths()] def fingerprint(self): body = {"owner_id": self.owner_id, "protocol_version": self.protocol_version, "model": MODEL, "bounds": [MAX_INPUT_TOKENS, MAX_OUTPUT_TOKENS], "units": self._units, "sources": {self.study.rel(p): sha256_file(p) for p in self.study.source_paths()}} return hashlib.sha256(canonical(body).encode("utf-8")).hexdigest() def _unused(self, unit, st): called = set(st["called"]) return [c["call_id"] for c in unit["calls"] if c["call_id"] not in called] def next(self, unit, history): st = replay(self.core, unit, history) call = expected_call(unit, st) if call is None: if unit["kind"] == "probe": status = st["probe"]["status"] reason = "probe_technical_missing" if st["stopped"] else "probe_complete" else: status = st["neg_terminal"]["status"] reason = {"negotiation_technical": "negotiation_technical_unknown", "check_technical": "post_episode_check_technical_missing"}.get(st["stopped"], "episode_complete") return {"done": True, "unused_call_ids": self._unused(unit, st), "scientific_status": status, "reason": reason} out = {"call_id": call["call_id"], "model": MODEL, "system": self.study.system(unit, call["role"]), "user": self.study.user(unit, call, st)} if call["phase"] == "check" and call["call_id"] == calls_by_phase(unit, "check")[0]["call_id"]: called = set(st["called"]) skipped = [c["call_id"] for c in calls_by_phase(unit, "negotiation") if c["call_id"] not in called] if skipped: out["skipped_call_ids"] = skipped out["skip_reason"] = "negotiation_terminal_agreement_before_final_negotiation_call" return out def score(self, unit, history, terminal=None): """Scores from replayed history. An incomplete history is scored only with a service terminal {outcome, detail}: 'input_bound:*' -> INPUT_LIMIT, anything else -> TECH_FAULT. If the negotiation is already known-terminal, the service terminal marks only the missing check(s).""" st = replay(self.core, unit, history) call = expected_call(unit, st) service_terminal_used = False if call is not None: svc = terminal if isinstance(terminal, dict) else {} detail = str(svc.get("detail") or "") if not svc.get("outcome") and not detail: return None # incomplete, no service terminal: never padded or scored service_terminal_used = True input_bound = detail.startswith("input_bound") reason = (detail or str(svc.get("outcome")))[:200] if call["phase"] == "negotiation": st["neg_terminal"] = {"status": "INPUT_LIMIT" if input_bound else "TECH_FAULT", "known": False, "turn": len(st["neg_calls"]) + 1, "role": call["role"], "call_id": call["call_id"], "reason": reason} st["stopped"] = "negotiation_service_terminal" elif call["phase"] == "check": st["checks"][call["call_id"]] = {"call_id": call["call_id"], "role": call["role"], "status": "CHECK_MISSING_INPUT_LIMIT" if input_bound else "CHECK_MISSING_TECHNICAL", "reason": reason, "answer": None, "correct": None} st["stopped"] = "check_service_terminal" else: st["probe"] = {"call_id": call["call_id"], "status": "PROBE_MISSING_INPUT_LIMIT" if input_bound else "PROBE_MISSING_TECHNICAL", "reason": reason, "answer": None, "correct": None} st["stopped"] = "probe_service_terminal" out = self.study.score(unit, st) out["service_terminal_used"] = service_terminal_used out["unused_call_ids"] = self._unused(unit, st) return out