(player_name: str, path: str, *, decision_timeout: float, max_policy_errors: int)
| 322 | |
| 323 | |
| 324 | def build_agent_class(player_name: str, path: str, *, decision_timeout: float, max_policy_errors: int): |
| 325 | class CodeClashSCMLAgent(GreedySyncAgent): |
| 326 | _policy_controllers: list[PolicyController] = [] |
| 327 | |
| 328 | def init(self): |
| 329 | super().init() |
| 330 | self._codeclash_policy = PolicyController( |
| 331 | player_name, path, timeout=decision_timeout, max_errors=max_policy_errors |
| 332 | ) |
| 333 | self.__class__._policy_controllers.append(self._codeclash_policy) |
| 334 | |
| 335 | def _base_observation(self, event: str, negotiator_id: str | None = None, state=None) -> dict[str, Any]: |
| 336 | nmi = self.get_nmi(negotiator_id) if negotiator_id else None |
| 337 | return { |
| 338 | "event": event, |
| 339 | "player": player_name, |
| 340 | "awi": awi_to_plain(self), |
| 341 | "negotiator_id": negotiator_id, |
| 342 | "nmi": nmi_to_plain(nmi), |
| 343 | "state": state_to_plain(state) if state is not None else {}, |
| 344 | } |
| 345 | |
| 346 | def propose(self, negotiator_id, state): |
| 347 | fallback = super().propose(negotiator_id, state) |
| 348 | observation = self._base_observation("propose", negotiator_id, state) |
| 349 | observation["fallback_offer"] = offer_to_plain(fallback) |
| 350 | decision = self._codeclash_policy.decide(observation) |
| 351 | if "offer" not in decision: |
| 352 | return fallback |
| 353 | offer, error = normalize_offer(decision.get("offer"), self.get_nmi(negotiator_id)) |
| 354 | if error is not None: |
| 355 | self._codeclash_policy._record_error("propose", error, invalid=True) |
| 356 | return fallback |
| 357 | return offer |
| 358 | |
| 359 | def respond(self, negotiator_id, state, source=""): |
| 360 | fallback = super().respond(negotiator_id, state, source) |
| 361 | observation = self._base_observation("respond", negotiator_id, state) |
| 362 | observation["fallback_response"] = response_to_name(fallback) |
| 363 | decision = self._codeclash_policy.decide(observation) |
| 364 | if "response" not in decision: |
| 365 | return fallback |
| 366 | response, error = normalize_response(decision.get("response")) |
| 367 | if error is not None: |
| 368 | self._codeclash_policy._record_error("respond", error, invalid=True) |
| 369 | return fallback |
| 370 | return response |
| 371 | |
| 372 | def first_proposals(self): |
| 373 | fallback_proposals = super().first_proposals() |
| 374 | proposals = {} |
| 375 | for negotiator_id, fallback in fallback_proposals.items(): |
| 376 | observation = self._base_observation("propose", negotiator_id) |
| 377 | observation["fallback_offer"] = offer_to_plain(fallback) |
| 378 | decision = self._codeclash_policy.decide(observation) |
| 379 | if "offer" not in decision: |
| 380 | proposals[negotiator_id] = fallback |
| 381 | continue |
no test coverage detected