"""environment320-v1 binding (implementation candidate; NOT frozen, NOT allocated). Locked design (Ryan, environmental N320; ENVIRONMENT_DECISION_CARD.md E-v0.1 + CONTROLLER_REVIEW_ADDENDUM.md): arm (disclosed exact certified kg CO2e | identical 'not provided' cells) x opener (buyer|seller first) x row order (F|R) = 8 exact cells x 40 repeats = 320 episodes, plus 4 one-call solo probes (arm x role). Each episode: up to 8 alternating negotiation calls + 2 post-episode checks (buyer, then seller) = 10 call IDs. 324 units, 3,204 planned call IDs (the approved maximum). Haiku 5.5 in every slot. Schedule: 40 blocks, each holding all 8 cells exactly once in sha256(SEED|block|cell) order, then the fixed probe block. The seed is fixed in source before any data; there is no adaptive stopping, replacement or outcome-based continuation. Execution entry: create_binding() (service py_binding_host contract). Local modules are loaded by file path under private names; sys.path and sys.modules are not modified. """ import hashlib import importlib.util import re from pathlib import Path HERE = Path(__file__).resolve().parent def _load(name): spec = importlib.util.spec_from_file_location(f"_environment320_{name}", HERE / f"{name}.py") mod = importlib.util.module_from_spec(spec) spec.loader.exec_module(mod) return mod core = _load("env_core") engine = _load("env_replay") OWNER_ID = "environment320" PROTOCOL_VERSION = "environment320-v1" SEED = "environment320-v1|schedule|2026-10-24|fixed-before-data" ARMS = ("disclosed", "undisclosed") ARM_SHORT = {"disclosed": "dis", "undisclosed": "und"} OPENERS = ("buyer_first", "seller_first") ORDERS = ("F", "R") CELLS = tuple((a, o, r) for a in ARMS for o in OPENERS for r in ORDERS) N_BLOCKS = 40 PROBES = (("disclosed", "buyer"), ("disclosed", "seller"), ("undisclosed", "buyer"), ("undisclosed", "seller")) CHECK_ORDER = ("buyer", "seller") N_SLOTS_MAX = 3204 NANO_PER_SLOT = engine.MAX_INPUT_TOKENS * 100 + engine.MAX_OUTPUT_TOKENS * 500 # 3,248,000 at saved Haiku rates PROMPTS = HERE / "prompts" PROMPT_FILES = ("system_buyer", "system_seller", "system_probe", "turn_user", "check_user", "probe_user") SOURCE_FILES = ("binding.py", "env_core.py", "env_replay.py", "analysis.py", "REQUEST_SETTINGS.json") SYSTEM_PH = re.compile(r"\{(PLAN_TABLE|NO_DEAL_EMISSIONS)\}") PROBE_PH = re.compile(r"\{(ROLE_OWNER|PROBE_TABLE|NO_DEAL_EMISSIONS)\}") ROLE_OWNER = {"buyer": "Brightwater Grocers Co-op", "seller": "Halden Freight"} ANY_PH = re.compile(r"\{([A-Z_]+)\}") TURN_PH = re.compile(r"\{(MESSAGE_NUMBER|OWN_MESSAGE_NUMBER|MESSAGES_LEFT|TRANSCRIPT|COUNTERPART_STANDING|OWN_STANDING)\}") def schedule(): out = [] for b in range(1, N_BLOCKS + 1): order = sorted(CELLS, key=lambda c: hashlib.sha256(f"{SEED}|{b}|{c[0]}|{c[1]}|{c[2]}".encode("utf-8")).hexdigest()) out += [{"block": b, "arm": a, "opener": o, "row_order": r} for a, o, r in order] return out def _call(call_id, role, phase, index): return {"call_id": call_id, "models": [engine.MODEL], "max_input_tokens": engine.MAX_INPUT_TOKENS, "max_output_tokens": engine.MAX_OUTPUT_TOKENS, "role": role, "phase": phase, "index": index} class EnvironmentStudy: OWNER_ID = OWNER_ID PROTOCOL_VERSION = PROTOCOL_VERSION def __init__(self): self.core = core self.t = {k: (PROMPTS / f"{k}.txt").read_text(encoding="utf-8") for k in PROMPT_FILES} for k in ("system_buyer", "system_seller"): found = SYSTEM_PH.findall(self.t[k]) if sorted(found) != ["NO_DEAL_EMISSIONS", "PLAN_TABLE"]: raise engine.IntegrityError(f"placeholders:{k}") if sorted(set(TURN_PH.findall(self.t["turn_user"]))) != sorted( ["MESSAGE_NUMBER", "OWN_MESSAGE_NUMBER", "MESSAGES_LEFT", "TRANSCRIPT", "COUNTERPART_STANDING", "OWN_STANDING"]): raise engine.IntegrityError("placeholders:turn_user") if sorted(PROBE_PH.findall(self.t["system_probe"])) != ["NO_DEAL_EMISSIONS", "PROBE_TABLE", "ROLE_OWNER"]: raise engine.IntegrityError("placeholders:system_probe") allowed = {"system_buyer": SYSTEM_PH, "system_seller": SYSTEM_PH, "system_probe": PROBE_PH, "turn_user": TURN_PH} for k in PROMPT_FILES: leftover = [m for m in ANY_PH.findall(self.t[k]) if not (k in allowed and allowed[k].fullmatch("{" + m + "}"))] if leftover: raise engine.IntegrityError(f"unexpected_placeholder:{k}:{leftover}") # ---- describe def rel(self, p): return Path(p).resolve().relative_to(HERE).as_posix() def source_paths(self): return [HERE / f for f in SOURCE_FILES] + [PROMPTS / f"{k}.txt" for k in sorted(PROMPT_FILES)] def build_units(self): units = [] for k, s in enumerate(schedule(), 1): uid = f"{OWNER_ID}-E{k:03d}-{ARM_SHORT[s['arm']]}-{s['opener'].split('_')[0]}-{s['row_order']}" first = s["opener"].split("_")[0] roles = [first if i % 2 == 0 else core.OTHER[first] for i in range(engine.N_NEGOTIATION)] calls = [_call(f"{uid}-n{i + 1:02d}-{r}", r, "negotiation", i + 1) for i, r in enumerate(roles)] calls += [_call(f"{uid}-k{j + 1}-{r}", r, "check", j + 1) for j, r in enumerate(CHECK_ORDER)] units.append({"unit_id": uid, "kind": "episode", "calls": calls, "design": {"ordinal": k, "block": s["block"], "arm": s["arm"], "opener": s["opener"], "row_order": s["row_order"], "cell": f"{s['arm']}|{s['opener']}|{s['row_order']}", "seed": SEED}}) for j, (arm, role) in enumerate(PROBES, 1): uid = f"{OWNER_ID}-P{j}-{ARM_SHORT[arm]}-{role}" units.append({"unit_id": uid, "kind": "probe", "calls": [_call(f"{uid}-p01-{role}", role, "probe", 1)], "design": {"ordinal": len(units) + 1, "block": N_BLOCKS + 1, "arm": arm, "role": role, "row_order": "F", "seed": SEED}}) n_slots = sum(len(u["calls"]) for u in units) ids = [c["call_id"] for u in units for c in u["calls"]] if len(units) != 324 or n_slots != N_SLOTS_MAX or len(set(ids)) != n_slots: raise engine.IntegrityError("unit_plan_size") return units # ---- requests: negotiation/check units get ONLY the own private packet + public history; # the four post-schedule probe units (C4) get the separate full-information probe packet. def system(self, unit, role): d = unit["design"] nd = f"{core.NO_DEAL_EMISSIONS_KG:,} kg CO2e" if d["arm"] == "disclosed" else core.NOT_PROVIDED if unit["kind"] == "probe": vals = {"ROLE_OWNER": ROLE_OWNER[role], "PROBE_TABLE": core.render_probe_table(d["arm"], d["row_order"]), "NO_DEAL_EMISSIONS": nd} return PROBE_PH.sub(lambda m: vals[m.group(1)], self.t["system_probe"]) vals = {"PLAN_TABLE": core.render_table(role, d["arm"], d["row_order"]), "NO_DEAL_EMISSIONS": nd} return SYSTEM_PH.sub(lambda m: vals[m.group(1)], self.t[f"system_{role}"]) def _render_entry(self, e): if e["kind"] == "null": return "[no valid action this turn]" body = e["message"] if e["message"] else "(no message text)" if e["kind"] == "propose": body += f"\n[Formal proposal: {core.render_package(e['proposal'])}]" elif e["kind"] == "accept": body += f"\n[Formal acceptance of the other side's standing proposal: {core.render_package(e['proposal'])}]" return body def _standing(self, s): return "none" if s is None else f"{core.render_package(s['package'])} (made in message {s['turn']})" def user(self, unit, call, st): if call["phase"] == "check": return self.t["check_user"] if call["phase"] == "probe": return self.t["probe_user"] me = call["role"] turn = len(st["neg_calls"]) + 1 transcript = "\n\n".join( f"Message {e['turn']} - {'You' if e['role'] == me else 'Other delegate'}:\n{self._render_entry(e)}" for e in st["log"]) or "(no messages yet)" vals = {"MESSAGE_NUMBER": str(turn), "OWN_MESSAGE_NUMBER": str((turn + 1) // 2), "MESSAGES_LEFT": str(engine.N_NEGOTIATION - turn), "TRANSCRIPT": transcript, "COUNTERPART_STANDING": self._standing(st["standing"][core.OTHER[me]]), "OWN_STANDING": self._standing(st["standing"][me])} return TURN_PH.sub(lambda m: vals[m.group(1)], self.t["turn_user"]) # ---- scoring (assigned arm; integer cents; known outcomes are never overwritten) def score(self, unit, st): d = unit["design"] base = {"study": OWNER_ID, "protocol_version": PROTOCOL_VERSION, "unit_id": unit["unit_id"], "kind": unit["kind"], "arm": d["arm"], "ordinal": d["ordinal"], "block": d["block"]} if unit["kind"] == "probe": base.update(role=d["role"], probe=st["probe"]) return base t = st["neg_terminal"] base.update(opener=d["opener"], row_order=d["row_order"], cell=d["cell"], status=t["status"], known=t["known"], terminal={k: t.get(k) for k in ("turn", "role", "call_id", "reason", "accepted_by", "proposed_by", "proposal_turn")}) props = [] for p in st["proposals"]: v, s, c = p["package"] o = core.outcome_of_deal(v, s, c) props.append({"turn": p["turn"], "role": p["role"], "plan": o["plan"], "price_cents": c, "proposer_below_own_fallback": o[f"{p['role']}_cents"] < core.FALLBACK[p["role"]], "dominated_plan": o["plan"] in ("D-N", "D-C"), "emissions_kg": o["emissions_kg"], "replaced_own_turn": p["replaced_own_turn"]}) nulls = [c for c in st["neg_calls"] if c["outcome"] == "null"] base.update(proposals=props, negotiation_calls=len(st["neg_calls"]), null_actions={r: [{"turn": c["turn"], "reason": c["reason"]} for c in nulls if c["role"] == r] for r in ("buyer", "seller")}, max_tokens_negotiation=[{"turn": c["turn"], "role": c["role"], "outcome": "valid_" + c["outcome"] if c["outcome"] != "null" else "null", "reason": c["reason"]} for c in st["neg_calls"] if c.get("max_tokens")]) checks = {} for c in engine.calls_by_phase(unit, "check"): checks[c["role"]] = st["checks"].get(c["call_id"]) or { "call_id": c["call_id"], "role": c["role"], "status": "CHECK_NOT_REACHED" if not t["known"] else "CHECK_NOT_ATTEMPTED_AFTER_TECHNICAL_STOP", "reason": None, "answer": None, "correct": None} base["checks"] = checks if not t["known"]: b = core.unknown_bounds() base.update(outcome=None, bounds={k: list(v) for k, v in b.items()}) return base if t["status"] == "DEAL": v, s, c = t["package"] o = core.outcome_of_deal(v, s, c) acc, prop = t["accepted_by"], t["proposed_by"] codes = {"accepter_below_fallback": o[f"{acc}_cents"] < core.FALLBACK[acc], "proposer_below_fallback": o[f"{prop}_cents"] < core.FALLBACK[prop], "dominated_plan": o["plan"] in ("D-N", "D-C"), "non_DO_package_joint_loss": o["plan"] != "D-O"} else: o = core.outcome_of_no_deal() codes = {"accepter_below_fallback": False, "proposer_below_fallback": False, "dominated_plan": False, "non_DO_package_joint_loss": False} o = dict(o) o.update(buyer_gain_cents=o["buyer_cents"] - core.FALLBACK["buyer"], seller_gain_cents=o["seller_cents"] - core.FALLBACK["seller"], L_norm=o["L"] / core.L_RANGE) if o["L"] != o["P"] + core.NO_DEAL_L * o["no_deal"]: raise engine.IntegrityError("decomposition") base.update(outcome=o, codes=codes, bounds=None) return base def create_binding(): """Execution entry (service py_binding_host contract).""" return engine.EnvironmentBinding(EnvironmentStudy())