| """Closed-loop dynamic runtime shield for black-box deployment.
|
|
|
| In open-loop deployment, an admitted word is dispatched unconditionally. Under
|
| latent drift, unmodeled delays, or assumption violations, intermediate outputs
|
| may contradict the surviving models or reveal that subsequent actions lead to an
|
| absorbing hazard. The dynamic runtime shield monitors intermediate outputs at
|
| every step, updates the active survivor set, and halts or re-plans before
|
| hazardous actions are executed.
|
| """
|
| from __future__ import annotations
|
|
|
| from itertools import product
|
| from typing import Callable
|
| from .core import CandidateSpace, Contract, _integer, _word
|
| from .interfaces import InterfaceError, TargetProcess
|
|
|
|
|
| def _check_cancel(cancel):
|
| if cancel and cancel():
|
| from .study import Cancelled
|
| raise Cancelled('Deployment cancelled by user')
|
|
|
|
|
| class RuntimeShield:
|
| """Monitors step-by-step target execution and intervenes when safety is violated."""
|
|
|
| def __init__(self, space: CandidateSpace, initial_survivors: tuple[int, ...],
|
| contract: Contract, mode: str = "shielded", cancel: Callable[[], bool] | None = None,
|
| fault_schedule: dict | None = None, recovery_strategy: str = "stabilize"):
|
| if mode not in ('open_loop', 'shielded', 'recovery', 'replan'):
|
| raise ValueError('Unknown execution mode')
|
| if recovery_strategy not in ('stabilize', 'soft_return'):
|
| raise ValueError('Unknown recovery strategy: must be "stabilize" or "soft_return"')
|
| self.space = space
|
| self.contract = contract
|
| self.mode = mode
|
| self.cancel = cancel
|
| self.fault_schedule = fault_schedule
|
| self.recovery_strategy = recovery_strategy
|
| self.change_point_step = fault_schedule.get('step') if fault_schedule else None
|
| self.fault_type = fault_schedule.get('type') if fault_schedule else None
|
| self.change_point_detected = False
|
| self.detection_step = None
|
| self.detection_latency = None
|
| self.active_survivors = space._ids(initial_survivors)
|
| self.history_words: list[int] = []
|
| self.history_outputs: list[int] = []
|
| self.survivor_counts: list[int] = [len(self.active_survivors)]
|
| self.status = "ACTIVE"
|
| self.intervened = False
|
| self.reasons: list[str] = []
|
| self.hazard_averted = False
|
| self.recovery_requested = False
|
| self.recovery_attempted = False
|
| self.recovery_certified = False
|
| self.recovery_succeeded = False
|
| self.recovery_reason = 'not_requested'
|
| self.recovery_action = None
|
| self.recovery_actions = []
|
| self.recovery_returned_to_initial = False
|
| self.recovery_candidate_count = 0
|
| self.recovery_remaining_horizon = contract.horizon
|
| self.replanned = False
|
| self.replan_events: list[dict] = []
|
| self.replanned_suffix: tuple[int, ...] | None = None
|
|
|
| def _check_change_point_detection(self, current_step_index: int):
|
| if (self.fault_schedule and self.fault_schedule.get('phase', 'deployment') in ('deployment', 'always')
|
| and self.change_point_step is not None):
|
| if current_step_index >= self.change_point_step and not self.change_point_detected:
|
| self.change_point_detected = True
|
| self.detection_step = current_step_index
|
| self.detection_latency = current_step_index - self.change_point_step
|
|
|
| def change_point_record(self) -> dict:
|
| return {
|
| 'scheduled': self.fault_schedule is not None,
|
| 'step': self.change_point_step,
|
| 'type': self.fault_type,
|
| 'detected': self.change_point_detected,
|
| 'detection_step': self.detection_step,
|
| 'detection_latency': self.detection_latency,
|
| }
|
|
|
| def _find_safe_suffix(self, prefix: tuple[int, ...], remaining_horizon: int) -> tuple[int, ...] | None:
|
| """Find shortest lexicographic action suffix satisfying the contract under all active survivors."""
|
| if not self.active_survivors or remaining_horizon <= 0:
|
| return None
|
| for length in range(1, remaining_horizon + 1):
|
| for cand in product(range(self.space.machine.inputs), repeat=length):
|
| full = prefix + cand
|
| if all(self.contract.accepts(self.space.semantic(aid, full)) for aid in self.active_survivors):
|
| return cand
|
| return None
|
|
|
| def _observe(self, action: int, output: int):
|
| self.history_words.append(action)
|
| self.history_outputs.append(output)
|
| word, outputs = tuple(self.history_words), tuple(self.history_outputs)
|
| self.active_survivors = tuple(aid for aid in self.active_survivors
|
| if self.space.predict(aid, word) == outputs)
|
| self.survivor_counts.append(len(self.active_survivors))
|
|
|
| def recovery_record(self) -> dict:
|
| return {'requested': self.recovery_requested, 'attempted': self.recovery_attempted,
|
| 'certified': self.recovery_certified, 'succeeded': self.recovery_succeeded,
|
| 'strategy': self.recovery_strategy,
|
| 'reason': self.recovery_reason, 'action': self.recovery_action,
|
| 'actions': list(self.recovery_actions),
|
| 'return_to_initial': self.recovery_returned_to_initial,
|
| 'candidate_count': self.recovery_candidate_count,
|
| 'remaining_horizon': self.recovery_remaining_horizon,
|
| 'scope': 'Conditional on target inclusion in the live finite model family, '
|
| 'stationarity, deterministic reset and truthful anchors. A consistent '
|
| 'response does not establish real-world safety or task completion.'}
|
|
|
| def recover(self, target: TargetProcess) -> int | None:
|
| """Execute one model-safe fallback or certified soft return-to-home.
|
|
|
| Refutation supplies no replacement safety model. Without an independently
|
| specified backup controller, the only justified response is to halt.
|
| """
|
| self.recovery_requested = True
|
| self.recovery_succeeded = False
|
| self.recovery_certified = False
|
| self.recovery_action = None
|
| self.recovery_actions = []
|
| self.recovery_returned_to_initial = False
|
| prefix, outputs = tuple(self.history_words), tuple(self.history_outputs)
|
| self.active_survivors = tuple(aid for aid in self.active_survivors
|
| if self.space.predict(aid, prefix) == outputs)
|
| self.recovery_candidate_count = len(self.active_survivors)
|
| self.recovery_remaining_horizon = max(0, self.contract.horizon - len(prefix))
|
| if not self.active_survivors:
|
| self.recovery_reason = 'model_contradicted'
|
| self.reasons.append('Recovery withheld: no model remains consistent with observed history')
|
| return None
|
| if not self.recovery_remaining_horizon:
|
| self.recovery_reason = 'horizon_exhausted'
|
| self.reasons.append('Recovery withheld: contract horizon is exhausted')
|
| return None
|
|
|
| if self.recovery_strategy == 'soft_return':
|
| ret = synthesize_safe_return(self.space, self.active_survivors, prefix,
|
| max_length=self.recovery_remaining_horizon)
|
| if not ret['certified']:
|
| self.recovery_reason = ret['reason']
|
| self.reasons.append(f"Recovery withheld: {ret['reason']}")
|
| return None
|
| word = ret.get('word')
|
| if not word:
|
| self.recovery_certified = True
|
| self.recovery_succeeded = True
|
| self.recovery_returned_to_initial = True
|
| self.recovery_reason = 'already_at_initial'
|
| self.status = 'RECOVERED_SAFE_FALLBACK'
|
| self.reasons.append('Model-certified soft return: target is already at initial state')
|
| return None
|
| self.recovery_certified = True
|
| self.recovery_actions = list(word)
|
| self.recovery_action = word[0]
|
| _check_cancel(self.cancel)
|
| self.recovery_attempted = True
|
| last_out = None
|
| for act in word:
|
| _check_cancel(self.cancel)
|
| try:
|
| last_out = target.step(act)
|
| except InterfaceError:
|
| self.recovery_reason = 'interface_error'
|
| raise
|
| self._observe(act, last_out)
|
| if not self.active_survivors:
|
| self.recovery_reason = 'response_contradiction'
|
| self.status = 'INTERVENED_ANOMALY'
|
| self.hazard_averted = False
|
| self.reasons.append('Recovery response contradicted every certifying model; safety is unresolved')
|
| return last_out
|
| self.recovery_succeeded = True
|
| self.recovery_returned_to_initial = ret.get('return_to_initial', False)
|
| self.recovery_reason = 'certified_return_to_initial' if self.recovery_returned_to_initial else 'certified_safe_hold'
|
| self.status = 'RECOVERED_SAFE_FALLBACK'
|
| self.reasons.append('Model-certified safe return executed with responses consistent with the live belief')
|
| return last_out
|
|
|
| safe_action = next((action for action in range(self.space.machine.inputs)
|
| if all(self.contract.hazard not in self.space.semantic(aid, prefix + (action,))
|
| for aid in self.active_survivors)), None)
|
| if safe_action is None:
|
| self.recovery_reason = 'no_safe_action'
|
| self.reasons.append('Recovery withheld: no action is hazard-free under every live model')
|
| return None
|
| self.recovery_action = safe_action
|
| self.recovery_actions = [safe_action]
|
| self.recovery_certified = True
|
| _check_cancel(self.cancel)
|
| self.recovery_attempted = True
|
| try:
|
| out = target.step(safe_action)
|
| except InterfaceError:
|
| self.recovery_reason = 'interface_error'
|
| raise
|
| self._observe(safe_action, out)
|
| if not self.active_survivors:
|
| self.recovery_reason = 'response_contradiction'
|
| self.status = 'INTERVENED_ANOMALY'
|
| self.hazard_averted = False
|
| self.reasons.append('Recovery response contradicted every certifying model; safety is unresolved')
|
| return out
|
| self.recovery_succeeded = True
|
| self.recovery_reason = 'certified_response'
|
| self.status = 'RECOVERED_SAFE_FALLBACK'
|
| self.reasons.append('Model-certified fallback executed with a response consistent with the live belief')
|
| return out
|
|
|
| def step(self, target: TargetProcess, planned_action: int, remaining_plan: list[int]) -> tuple[int | None, bool]:
|
| """Execute one step with active pre- and post-condition safety guards.
|
|
|
| Returns (output, should_continue).
|
| """
|
| if self.mode == "open_loop":
|
|
|
| out = target.step(planned_action)
|
| self.history_words.append(planned_action)
|
| self.history_outputs.append(out)
|
| return out, True
|
|
|
|
|
| if not self.active_survivors:
|
| self._check_change_point_detection(len(self.history_words))
|
| self.status = "INTERVENED_ANOMALY"
|
| self.intervened = True
|
| self.reasons.append("Pre-step abort: no surviving models match history (anomaly/drift detected)")
|
| if self.mode in ("recovery", "replan"):
|
| rec_out = self.recover(target)
|
| return rec_out, False
|
| return None, False
|
|
|
|
|
|
|
| candidate_prefix = tuple(self.history_words) + (planned_action,)
|
| if any(self.contract.hazard in self.space.semantic(aid, candidate_prefix)
|
| for aid in self.active_survivors):
|
| if self.mode == "replan":
|
| rem_h = max(0, self.contract.horizon - len(self.history_words))
|
| safe_suffix = self._find_safe_suffix(tuple(self.history_words), rem_h)
|
| if safe_suffix is not None:
|
| self._check_change_point_detection(len(self.history_words))
|
| self.replanned = True
|
| self.replan_events.append({
|
| 'step': len(self.history_words), 'trigger': 'pre_step_hazard',
|
| 'old_action': planned_action, 'new_suffix': list(safe_suffix),
|
| 'survivor_count': len(self.active_survivors)
|
| })
|
| self.status = "REPLANNED_ONLINE_SAFE"
|
| self.reasons.append(f"Replanned online at step {len(self.history_words)}: switched to safe suffix {list(safe_suffix)}")
|
| planned_action = safe_suffix[0]
|
| self.replanned_suffix = safe_suffix[1:]
|
| else:
|
| self._check_change_point_detection(len(self.history_words))
|
| self.status = "INTERVENED_HAZARD_AVERTED"
|
| self.intervened = True
|
| self.hazard_averted = True
|
| self.reasons.append(f"Pre-step abort at step {len(self.history_words) + 1}: action {planned_action} risks hazard and no safe replan exists")
|
| rec_out = self.recover(target)
|
| return rec_out, False
|
| else:
|
| self._check_change_point_detection(len(self.history_words))
|
| self.status = "INTERVENED_HAZARD_AVERTED"
|
| self.intervened = True
|
| self.hazard_averted = True
|
| self.reasons.append(f"Pre-step abort at step {len(self.history_words) + 1}: action {planned_action} risks hazard under a surviving adapter")
|
| if self.mode == "recovery":
|
| rec_out = self.recover(target)
|
| return rec_out, False
|
| return None, False
|
|
|
|
|
| out = target.step(planned_action)
|
| self._observe(planned_action, out)
|
| current_word = tuple(self.history_words)
|
|
|
| if not self.active_survivors:
|
|
|
| self._check_change_point_detection(len(self.history_words) - 1)
|
| self.status = "INTERVENED_ANOMALY"
|
| self.intervened = True
|
| self.reasons.append(f"Post-step abort at step {len(self.history_words)}: output {out} refuted all surviving adapters")
|
| if self.mode in ("recovery", "replan"):
|
| self.recover(target)
|
| return out, False
|
| return out, False
|
|
|
|
|
| rem_plan_to_check = self.replanned_suffix if self.replanned_suffix is not None else remaining_plan
|
| if rem_plan_to_check:
|
| full_plan = current_word + tuple(rem_plan_to_check)
|
| suffix_hazardous = any(
|
| self.contract.hazard in self.space.semantic(adapter_id, full_plan)
|
| for adapter_id in self.active_survivors
|
| )
|
| suffix_misses_goal = (self.mode == "replan") and any(
|
| not self.contract.accepts(self.space.semantic(adapter_id, full_plan))
|
| for adapter_id in self.active_survivors
|
| )
|
|
|
| if suffix_hazardous or suffix_misses_goal:
|
| if self.mode == "replan":
|
| rem_h = max(0, self.contract.horizon - len(self.history_words))
|
| safe_suffix = self._find_safe_suffix(current_word, rem_h)
|
| if safe_suffix is not None:
|
| self._check_change_point_detection(len(self.history_words) - 1)
|
| self.replanned = True
|
| self.replan_events.append({
|
| 'step': len(self.history_words), 'trigger': 'post_step_suffix_risk',
|
| 'old_suffix': list(rem_plan_to_check), 'new_suffix': list(safe_suffix),
|
| 'survivor_count': len(self.active_survivors)
|
| })
|
| self.status = "REPLANNED_ONLINE_SAFE"
|
| self.reasons.append(f"Replanned online at step {len(self.history_words)}: switched to safe suffix {list(safe_suffix)}")
|
| self.replanned_suffix = safe_suffix
|
| return out, True
|
| else:
|
| self._check_change_point_detection(len(self.history_words) - 1)
|
| self.status = "INTERVENED_HAZARD_AVERTED"
|
| self.intervened = True
|
| self.hazard_averted = True
|
| self.reasons.append(f"Post-step abort at step {len(self.history_words)}: upcoming plan suffix risks hazard and no safe replan exists")
|
| self.recover(target)
|
| return out, False
|
| else:
|
| self._check_change_point_detection(len(self.history_words) - 1)
|
| self.status = "INTERVENED_HAZARD_AVERTED"
|
| self.intervened = True
|
| self.hazard_averted = True
|
| self.reasons.append(f"Post-step abort at step {len(self.history_words)}: upcoming plan suffix {remaining_plan} risks hazard under updated survivors")
|
| if self.mode == "recovery":
|
| self.recover(target)
|
| return out, False
|
| return out, False
|
|
|
| return out, True
|
|
|
|
|
| def execute_deployment(target: TargetProcess, planned_word: list[int], space: CandidateSpace,
|
| survivors: tuple[int, ...], contract: Contract,
|
| mode: str = "open_loop", cancel: Callable[[], bool] | None = None,
|
| fault_schedule: dict | None = None, recovery_strategy: str = "stabilize") -> dict:
|
| """Execute deployment through the target process with optional closed-loop shielding."""
|
| if hasattr(target, 'set_phase'):
|
| target.set_phase('deployment')
|
| shield = RuntimeShield(space, survivors, contract, mode=mode, cancel=cancel,
|
| fault_schedule=fault_schedule, recovery_strategy=recovery_strategy)
|
| deployment = {
|
| "word": list(planned_word),
|
| "executed_word": [],
|
| "outputs": [],
|
| "complete": False,
|
| "mode": mode,
|
| "shield_status": "COMPLETED" if mode == "open_loop" else "ACTIVE",
|
| "intervened": False,
|
| "hazard_averted": False,
|
| "reasons": [],
|
| "survivor_history": [len(survivors)],
|
| "target_steps": 0,
|
| "resets": 0,
|
| "target_calls": 0,
|
| "error": None,
|
| }
|
|
|
| previous_counts = dict(target.counts)
|
| try:
|
| _check_cancel(cancel)
|
| target.reset()
|
| plan = list(planned_word)
|
| step_idx = 0
|
| while step_idx < len(plan):
|
| _check_cancel(cancel)
|
| action = plan[step_idx]
|
| remaining = plan[step_idx + 1:]
|
| _, should_continue = shield.step(target, action, remaining)
|
| if shield.replanned_suffix is not None:
|
| plan = plan[:step_idx + 1] + list(shield.replanned_suffix)
|
| shield.replanned_suffix = None
|
| if not should_continue:
|
| break
|
| step_idx += 1
|
|
|
| deployment["complete"] = (len(shield.history_words) == len(plan)) and not shield.intervened
|
| deployment["shield_status"] = shield.status if shield.intervened else ("REPLANNED_ONLINE_SAFE" if shield.replanned else "COMPLETED")
|
|
|
| except InterfaceError as exc:
|
| deployment["error"] = {"code": exc.code, "message": str(exc)}
|
| deployment["shield_status"] = "INTERFACE_ERROR"
|
| finally:
|
| deployment["executed_word"] = list(shield.history_words)
|
| deployment["outputs"] = list(shield.history_outputs)
|
| deployment["intervened"] = shield.intervened
|
| deployment["hazard_averted"] = shield.hazard_averted
|
| deployment["replanned"] = shield.replanned
|
| deployment["replan_count"] = len(shield.replan_events)
|
| deployment["replan_events"] = list(shield.replan_events)
|
| deployment["reasons"] = list(shield.reasons)
|
| deployment["survivor_history"] = list(shield.survivor_counts)
|
| deployment["recovery"] = shield.recovery_record()
|
| deployment["change_point"] = shield.change_point_record()
|
| deployment["target_steps"] = target.counts["step_attempts"] - previous_counts["step_attempts"]
|
| deployment["resets"] = target.counts["reset_attempts"] - previous_counts["reset_attempts"]
|
| deployment["target_calls"] = target.counts["target_calls"] - previous_counts["target_calls"]
|
|
|
| return deployment
|
|
|
|
|
| def synthesize_safe_return(space: CandidateSpace, survivors, prefix=(), max_length: int | None = None) -> dict:
|
| """Synthesize a certified hazard-free return-to-initial (soft reset) or stabilizing suffix.
|
|
|
| Searches for the shortest action suffix `w_ret` within the remaining horizon such that:
|
| 1. `prefix + w_ret` is hazard-free under every surviving model in `survivors`.
|
| 2. `prefix + w_ret` returns the reference state machine to `space.machine.initial` across
|
| every surviving model (certified soft return-to-home).
|
| If no return-to-initial path exists within the horizon, falls back to the shortest
|
| universally hazard-free holding step if one exists.
|
| ``max_length`` is an optional nonnegative integer that can only narrow the
|
| remaining contract horizon; a prefix beyond that horizon is invalid.
|
| """
|
| survivors = space._ids(survivors)
|
| prefix = _word(prefix, space.machine.inputs)
|
| if max_length is not None:
|
| _integer(max_length, 'max_length')
|
| if len(prefix) > space.contract.horizon:
|
| raise ValueError('prefix exceeds the contract horizon')
|
| if not survivors:
|
| return {
|
| 'certified': False, 'return_to_initial': False, 'word': None,
|
| 'reason': 'empty_belief', 'candidate_count': 0,
|
| }
|
| if prefix and not space.probe_is_safe(survivors, prefix):
|
| return {
|
| 'certified': False, 'return_to_initial': False, 'word': None,
|
| 'reason': 'prefix_already_hazardous', 'candidate_count': len(survivors),
|
| }
|
|
|
| def _end_state(input_map, full_word):
|
| state = space.machine.initial
|
| for sym in full_word:
|
| state, out = space.machine.transitions[state][input_map[sym]]
|
| if out == space.contract.hazard:
|
| return None
|
| return state
|
|
|
| if hasattr(space, 'decompose_id'):
|
| active_maps = [space.input_perms[i_idx] for i_idx in sorted({space.decompose_id(i)[0] for i in survivors})]
|
| else:
|
| seen_maps = []
|
| for i in survivors:
|
| im = space.adapters[i].input_map
|
| if im not in seen_maps:
|
| seen_maps.append(im)
|
| active_maps = seen_maps
|
|
|
| if all(_end_state(im, prefix) == space.machine.initial for im in active_maps):
|
| return {
|
| 'certified': True, 'return_to_initial': True, 'word': [],
|
| 'reason': 'already_at_initial', 'candidate_count': len(survivors),
|
| }
|
|
|
| rem = space.contract.horizon - len(prefix)
|
| if max_length is not None:
|
| rem = min(rem, max_length)
|
| if rem <= 0:
|
| return {
|
| 'certified': False, 'return_to_initial': False, 'word': None,
|
| 'reason': 'horizon_exhausted', 'candidate_count': len(survivors),
|
| }
|
|
|
| for length in range(1, rem + 1):
|
| for cand in product(range(space.machine.inputs), repeat=length):
|
| full = prefix + cand
|
| if all(_end_state(im, full) == space.machine.initial for im in active_maps):
|
| return {
|
| 'certified': True, 'return_to_initial': True, 'word': list(cand),
|
| 'reason': 'certified_return_to_initial', 'candidate_count': len(survivors),
|
| }
|
|
|
| for action in range(space.machine.inputs):
|
| full = prefix + (action,)
|
| if all(_end_state(im, full) is not None for im in active_maps):
|
| return {
|
| 'certified': True, 'return_to_initial': False, 'word': [action],
|
| 'reason': 'certified_safe_hold', 'candidate_count': len(survivors),
|
| }
|
|
|
| return {
|
| 'certified': False, 'return_to_initial': False, 'word': None,
|
| 'reason': 'no_safe_continuation', 'candidate_count': len(survivors),
|
| }
|
|
|
