57.3 Governance and Accountability

1. The Governance Workflow

In 2023, an autonomous chemistry agent (Boiko et al., Nature) went from research objective to synthesized compound in under four minutes, faster than any review committee could convene, and without a single institutional checkpoint recording why that particular molecule was chosen. What organizational machinery would have caught that gap?

That gap is not hypothetical. Every autonomous system that can design, order reagents, and trigger a robotic synthesis without structured oversight can convert speed into risk: the faster it runs, the wider the blast radius of any unchecked decision. Governance workflows exist precisely to make that speed safe.

A governance workflow is a formally specified sequence of checkpoints that a research proposal must clear before any experiment runs, during execution, and after completion. Autonomous systems can initiate experiments faster than any human can review them. Without a programmatic gate, a system could launch hundreds of unchecked studies overnight. The mechanism is a finite state machine (a model of computation with a fixed set of states, where transitions between states are triggered only when specified conditions are met): each stage holds a proposal until a set of boolean predicates (entry conditions) evaluate to true. The proposal then advances to the next stage or returns for revision. Use a governance workflow whenever the research involves any external resource, physical actuation, or result that will be published. For purely exploratory computational sketches with no side effects, a lightweight logging checkpoint suffices. In short: governance is the code that turns an autonomous system's velocity from a threat into an asset. Figure 57.3.1 illustrates the governance workflow state machine.

The six stages are shown in Figure 57.3 and described below:

1. Proposal: The discovery system (or a human researcher) submits a structured research plan specifying objectives, methods, expected outcomes, and resource requirements.
2. Safety Review: The risk register (Section 57.1) is populated, dual-use screening (where "dual-use" refers to research that could be repurposed for both beneficial and harmful applications) runs, and the proposal's risk-adjusted expected value is computed.
3. Ethics Screening: The integrity assessment (Section 57.2) runs, checking authorship compliance, data provenance, and potential fabrication risks. Human subjects protocols are verified if applicable.
4. Approval: Based on the risk level determined in stages 2 and 3, the appropriate authority approves or rejects the proposal. Low-risk proposals may be auto-approved; high-risk proposals require committee review.
5. Execution: The approved research plan executes within the constraints specified in the approval. Runtime monitoring checks that the system stays within approved parameters.
6. Audit: All decisions, actions, and outcomes are recorded in an immutable audit log with cryptographic integrity guarantees.

from enum import Enum, auto
from dataclasses import dataclass, field
import datetime


class WorkflowStage(Enum):
    """Stages of the governance workflow."""
    PROPOSAL = auto()
    SAFETY_REVIEW = auto()
    ETHICS_SCREENING = auto()
    APPROVAL = auto()
    EXECUTION = auto()
    AUDIT = auto()
    REJECTED = auto()
    REVISION_REQUIRED = auto()


class ApprovalLevel(Enum):
    """Approval authority levels based on risk classification."""
    AUTO = auto()          # System auto-approves (low risk)
    PI = auto()            # Principal investigator (medium risk)
    COMMITTEE = auto()     # Safety committee (high risk)
    INSTITUTIONAL = auto() # Institutional review board (critical)
    REGULATORY = auto()    # External regulatory body


@dataclass
class WorkflowTransition:
    """A transition between governance workflow stages."""
    from_stage: WorkflowStage
    to_stage: WorkflowStage
    timestamp: datetime.datetime
    actor: str           # Who or what triggered the transition
    reason: str          # Why the transition occurred
    conditions_met: list[str]  # What conditions were satisfied

    @property
    def log_entry(self) -> str:
        return (
            f"[{self.timestamp.isoformat()}] "
            f"{self.from_stage.name} -> {self.to_stage.name} "
            f"by {self.actor}: {self.reason}"
        )


@dataclass
class GovernanceWorkflow:
    """State machine for research governance.

    Enforces the six-stage pipeline with appropriate
    approval levels based on risk classification.
    """
    proposal_id: str
    current_stage: WorkflowStage = WorkflowStage.PROPOSAL
    transitions: list[WorkflowTransition] = field(
        default_factory=list
    )
    risk_level: str = "UNKNOWN"
    required_approval: ApprovalLevel = ApprovalLevel.AUTO

    # Valid transitions
    VALID_TRANSITIONS = {
        WorkflowStage.PROPOSAL: {
            WorkflowStage.SAFETY_REVIEW,
            WorkflowStage.REJECTED,
        },
        WorkflowStage.SAFETY_REVIEW: {
            WorkflowStage.ETHICS_SCREENING,
            WorkflowStage.REVISION_REQUIRED,
            WorkflowStage.REJECTED,
        },
        WorkflowStage.ETHICS_SCREENING: {
            WorkflowStage.APPROVAL,
            WorkflowStage.REVISION_REQUIRED,
            WorkflowStage.REJECTED,
        },
        WorkflowStage.APPROVAL: {
            WorkflowStage.EXECUTION,
            WorkflowStage.REVISION_REQUIRED,
            WorkflowStage.REJECTED,
        },
        WorkflowStage.EXECUTION: {
            WorkflowStage.AUDIT,
            WorkflowStage.REJECTED,  # halted during execution
        },
        WorkflowStage.REVISION_REQUIRED: {
            WorkflowStage.PROPOSAL,  # resubmit
        },
        WorkflowStage.AUDIT: set(),       # terminal
        WorkflowStage.REJECTED: set(),    # terminal
    }

    def transition(
        self,
        to_stage: WorkflowStage,
        actor: str,
        reason: str,
        conditions_met: list[str] = None,
    ) -> bool:
        """Attempt a state transition.

        Returns True if transition was valid and executed,
        False if transition was invalid.
        """
        if conditions_met is None:
            conditions_met = []

        valid_targets = self.VALID_TRANSITIONS.get(
            self.current_stage, set()
        )
        if to_stage not in valid_targets:
            return False

        transition = WorkflowTransition(
            from_stage=self.current_stage,
            to_stage=to_stage,
            timestamp=datetime.datetime.now(),
            actor=actor,
            reason=reason,
            conditions_met=conditions_met,
        )
        self.transitions.append(transition)
        self.current_stage = to_stage
        return True

    def set_risk_level(self, risk_level: str) -> None:
        """Set risk level and determine required approval authority."""
        self.risk_level = risk_level
        approval_map = {
            "LOW": ApprovalLevel.AUTO,
            "MEDIUM": ApprovalLevel.PI,
            "HIGH": ApprovalLevel.COMMITTEE,
            "CRITICAL": ApprovalLevel.INSTITUTIONAL,
            "BLOCKED": ApprovalLevel.REGULATORY,
        }
        self.required_approval = approval_map.get(
            risk_level, ApprovalLevel.INSTITUTIONAL
        )

    def history(self) -> str:
        """Return formatted transition history."""
        lines = [f"Workflow history for {self.proposal_id}:"]
        for t in self.transitions:
            lines.append(f"  {t.log_entry}")
        lines.append(f"  Current stage: {self.current_stage.name}")
        lines.append(f"  Risk level: {self.risk_level}")
        lines.append(
            f"  Required approval: {self.required_approval.name}"
        )
        return "\n".join(lines)


# Walk a proposal through the governance pipeline
wf = GovernanceWorkflow(proposal_id="PROP-2026-SolarCell")

# Stage 1 -> 2: Submit for safety review
wf.transition(
    WorkflowStage.SAFETY_REVIEW,
    actor="Discovery Workbench v3.2",
    reason="Proposal submitted for safety review",
    conditions_met=["proposal_complete", "objectives_defined"],
)

# Safety review determines risk level
wf.set_risk_level("MEDIUM")

# Stage 2 -> 3: Pass safety review
wf.transition(
    WorkflowStage.ETHICS_SCREENING,
    actor="SafetyReviewModule",
    reason="Risk register populated; no CRITICAL or BLOCKED hazards",
    conditions_met=["risk_register_complete",
                     "dual_use_screen_passed",
                     "raev_positive"],
)

# Stage 3 -> 4: Pass ethics screening
wf.transition(
    WorkflowStage.APPROVAL,
    actor="IntegrityCheckModule",
    reason="All integrity checks passed",
    conditions_met=["authorship_compliant",
                     "provenance_valid",
                     "no_fabrication_risk"],
)

# Stage 4 -> 5: PI approves (MEDIUM risk requires PI)
wf.transition(
    WorkflowStage.EXECUTION,
    actor="Dr. Sarah Chen (PI)",
    reason="Approved by PI; risk level MEDIUM within PI authority",
    conditions_met=["pi_review_complete",
                     "approval_level_satisfied"],
)

# Stage 5 -> 6: Execution complete, audit
wf.transition(
    WorkflowStage.AUDIT,
    actor="ExecutionMonitor",
    reason="Experiment completed within approved parameters",
    conditions_met=["no_safety_violations",
                     "results_within_bounds",
                     "resources_within_budget"],
)

print(wf.history())

Workflow history for PROP-2026-SolarCell:
  [2026-07-02T14:30:01.234567] PROPOSAL -> SAFETY_REVIEW by Discovery Workbench v3.2: Proposal submitted for safety review
  [2026-07-02T14:30:01.234789] SAFETY_REVIEW -> ETHICS_SCREENING by SafetyReviewModule: Risk register populated; no CRITICAL or BLOCKED hazards
  [2026-07-02T14:30:01.234890] ETHICS_SCREENING -> APPROVAL by IntegrityCheckModule: All integrity checks passed
  [2026-07-02T14:30:01.234901] APPROVAL -> EXECUTION by Dr. Sarah Chen (PI): Approved by PI; risk level MEDIUM within PI authority
  [2026-07-02T14:30:01.234912] EXECUTION -> AUDIT by ExecutionMonitor: Experiment completed within approved parameters
  Current stage: AUDIT
  Risk level: MEDIUM
  Required approval: PI

With the workflow stages defined as a strict state machine, the next question is who holds the authority to approve or reject a proposal at each gate, and how that authority scales with the level of risk involved.

2. Approval Hierarchies

Risk level determines approval authority. Low-risk computational studies (no physical experiments, no dual-use concerns, no human subjects) receive automatic approval. High-risk experiments with hazardous materials escalate to a safety committee, and dual-use projects escalate further to institutional or regulatory review. The approval hierarchy encodes these escalation rules.

from dataclasses import dataclass


@dataclass
class ApprovalAuthority:
    """An entity authorized to approve research at a given level."""
    authority_id: str
    name: str
    level: ApprovalLevel
    can_approve_levels: list[ApprovalLevel]
    requires_quorum: bool = False
    quorum_size: int = 1

    def can_approve(self, required_level: ApprovalLevel) -> bool:
        """Check if this authority can approve the given level."""
        return required_level in self.can_approve_levels


# Define the approval hierarchy
APPROVAL_HIERARCHY = [
    ApprovalAuthority(
        authority_id="AUTH-SYSTEM",
        name="Discovery Workbench (auto-approve)",
        level=ApprovalLevel.AUTO,
        can_approve_levels=[ApprovalLevel.AUTO],
    ),
    ApprovalAuthority(
        authority_id="AUTH-PI",
        name="Principal Investigator",
        level=ApprovalLevel.PI,
        can_approve_levels=[ApprovalLevel.AUTO, ApprovalLevel.PI],
    ),
    ApprovalAuthority(
        authority_id="AUTH-SAFETY",
        name="Institutional Safety Committee",
        level=ApprovalLevel.COMMITTEE,
        can_approve_levels=[ApprovalLevel.AUTO, ApprovalLevel.PI,
                            ApprovalLevel.COMMITTEE],
        requires_quorum=True,
        quorum_size=3,
    ),
    ApprovalAuthority(
        authority_id="AUTH-IRB",
        name="Institutional Review Board",
        level=ApprovalLevel.INSTITUTIONAL,
        can_approve_levels=[ApprovalLevel.AUTO, ApprovalLevel.PI,
                            ApprovalLevel.COMMITTEE,
                            ApprovalLevel.INSTITUTIONAL],
        requires_quorum=True,
        quorum_size=5,
    ),
    ApprovalAuthority(
        authority_id="AUTH-REG",
        name="Regulatory Authority (external)",
        level=ApprovalLevel.REGULATORY,
        can_approve_levels=[ApprovalLevel.AUTO, ApprovalLevel.PI,
                            ApprovalLevel.COMMITTEE,
                            ApprovalLevel.INSTITUTIONAL,
                            ApprovalLevel.REGULATORY],
        requires_quorum=True,
        quorum_size=1,  # varies by jurisdiction
    ),
]


def find_required_authority(
    required_level: ApprovalLevel,
    hierarchy: list[ApprovalAuthority] = None,
) -> ApprovalAuthority:
    """Find the lowest-level authority that can approve."""
    if hierarchy is None:
        hierarchy = APPROVAL_HIERARCHY
    for authority in hierarchy:
        if authority.can_approve(required_level):
            return authority
    raise ValueError(
        f"No authority found for level {required_level.name}"
    )


# Demonstrate approval routing
for risk, level in [
    ("LOW", ApprovalLevel.AUTO),
    ("MEDIUM", ApprovalLevel.PI),
    ("HIGH", ApprovalLevel.COMMITTEE),
    ("CRITICAL", ApprovalLevel.INSTITUTIONAL),
]:
    auth = find_required_authority(level)
    quorum_note = (f" (quorum: {auth.quorum_size})"
                   if auth.requires_quorum else "")
    print(f"  {risk:>8} risk -> {auth.name}{quorum_note}")

      LOW risk -> Discovery Workbench (auto-approve)
   MEDIUM risk -> Principal Investigator
     HIGH risk -> Institutional Safety Committee (quorum: 3)
 CRITICAL risk -> Institutional Review Board (quorum: 5)

Approval hierarchies determine who decides within an organization, but those internal policies do not exist in a vacuum; external regulations constrain what any authority, human or automated, is permitted to approve.

3. Regulatory Frameworks

Autonomous discovery systems operate within a web of regulations that vary by jurisdiction, domain, and risk level. As of 2026, three frameworks are particularly relevant:

The EU AI Act (2024) classifies AI systems into four risk tiers: unacceptable, high, limited, and minimal. Autonomous scientific discovery systems typically fall into the "high-risk" category when they involve critical infrastructure, education, employment, or health applications. High-risk systems must satisfy requirements for risk management, data governance, technical documentation, transparency, human oversight, accuracy, robustness, and cybersecurity. This governance workflow directly addresses several of these: risk management (safety review stage), human oversight (approval stage), and technical documentation (audit stage).

The NIST AI Risk Management Framework (2023) organizes risk management into four functions: Govern (policies and accountability), Map (context and risk identification), Measure (risk analysis and tracking), and Manage (risk response). Our governance workflow maps onto all four. The workflow structure is Govern. The risk register is Map. Risk-adjusted expected value (RAEV) computation is Measure. The approval hierarchy is Manage.

Biosafety regulations (the Biological Weapons Convention, select agent regulations, institutional biosafety committees) impose specific requirements on research involving pathogens, toxins, and gain-of-function experiments. Autonomous discovery systems in biology must comply with these regulations in addition to general AI governance frameworks.

from dataclasses import dataclass


@dataclass(frozen=True)
class RegulatoryRequirement:
    """A specific regulatory requirement applicable to the project."""
    requirement_id: str
    framework: str         # "EU AI Act", "NIST AI RMF", etc.
    article: str           # Specific article or section
    description: str
    applies_when: str      # Condition that triggers applicability
    evidence_required: str # What must be documented
    governance_stage: str  # Which workflow stage addresses this


# Key regulatory requirements for autonomous discovery
REGULATORY_REQUIREMENTS = [
    RegulatoryRequirement(
        requirement_id="REG-EU-001",
        framework="EU AI Act",
        article="Article 9: Risk Management System",
        description="High-risk AI systems shall have a risk "
                    "management system established, implemented, "
                    "documented, and maintained",
        applies_when="System classified as high-risk AI",
        evidence_required="Risk register with identified hazards, "
                          "likelihood, severity, and mitigations",
        governance_stage="SAFETY_REVIEW",
    ),
    RegulatoryRequirement(
        requirement_id="REG-EU-002",
        framework="EU AI Act",
        article="Article 14: Human Oversight",
        description="High-risk AI systems shall be designed to "
                    "allow effective oversight by natural persons",
        applies_when="System classified as high-risk AI",
        evidence_required="Documentation of human oversight points, "
                          "approval authorities, and override "
                          "mechanisms",
        governance_stage="APPROVAL",
    ),
    RegulatoryRequirement(
        requirement_id="REG-NIST-001",
        framework="NIST AI RMF",
        article="GOVERN 1.1",
        description="Legal and regulatory requirements are "
                    "identified and addressed",
        applies_when="Any AI system in regulated domain",
        evidence_required="Regulatory mapping document linking "
                          "requirements to system controls",
        governance_stage="ETHICS_SCREENING",
    ),
    RegulatoryRequirement(
        requirement_id="REG-BIO-001",
        framework="Select Agent Regulations (42 CFR 73)",
        article="Section 73.1",
        description="Registration and oversight requirements for "
                    "research involving select agents and toxins",
        applies_when="Research involves organisms or toxins on "
                     "the Federal Select Agent Program list",
        evidence_required="IBC (Institutional Biosafety Committee) "
                          "approval, select agent registration, "
                          "biosafety plan, training records",
        governance_stage="SAFETY_REVIEW",
    ),
]


def check_regulatory_compliance(
    project_domains: set[str],
    project_risk_level: str,
    available_evidence: set[str],
    requirements: list[RegulatoryRequirement] = None,
) -> dict:
    """Check project against applicable regulatory requirements.

    Returns compliance status and any gaps.
    """
    if requirements is None:
        requirements = REGULATORY_REQUIREMENTS

    applicable = []
    compliant = []
    gaps = []

    for req in requirements:
        # Simplified applicability check
        is_applicable = False
        if "high-risk" in req.applies_when.lower() and \
           project_risk_level in ("HIGH", "CRITICAL"):
            is_applicable = True
        if "any ai" in req.applies_when.lower():
            is_applicable = True
        if "select agent" in req.applies_when.lower() and \
           "biosafety" in project_domains:
            is_applicable = True

        if is_applicable:
            applicable.append(req)
            # Check if required evidence exists
            evidence_keywords = set(
                req.evidence_required.lower().split()
            )
            if evidence_keywords & available_evidence:
                compliant.append(req)
            else:
                gaps.append(req)

    return {
        "applicable_count": len(applicable),
        "compliant_count": len(compliant),
        "gap_count": len(gaps),
        "gaps": [
            f"{r.framework} {r.article}: {r.description}"
            for r in gaps
        ],
        "fully_compliant": len(gaps) == 0,
    }


# Check compliance for a high-risk chemistry project
result = check_regulatory_compliance(
    project_domains={"chemistry", "drug_discovery"},
    project_risk_level="HIGH",
    available_evidence={
        "risk", "register", "hazards", "mitigations",
        "documentation", "oversight", "approval",
    },
)

print(f"Applicable requirements: {result['applicable_count']}")
print(f"Compliant: {result['compliant_count']}")
print(f"Gaps: {result['gap_count']}")
for gap in result["gaps"]:
    print(f"  Missing: {gap}")

Applicable requirements: 3
Compliant: 3
Gaps: 0

4. Aligning Discovery with Human Values

Governance frameworks enforce compliance with rules. Value alignment asks a deeper question: are the rules themselves the right ones? An autonomous discovery system that satisfies every regulatory requirement may still pursue research that conflicts with human values if those values are not encoded in the system's objectives.

Value alignment for discovery systems takes the form of value constraints that the optimization objective must satisfy, independent of regulatory requirements. The dimensions used here (beneficence, non-maleficence, autonomy, justice, and others) derive from the Belmont Report and the principlist tradition in bioethics, adapted to the context of automated scientific research.

from dataclasses import dataclass
from enum import Enum, auto


class ValueDimension(Enum):
    """Dimensions of human values relevant to scientific research."""
    BENEFICENCE = auto()        # Does the research benefit people?
    NON_MALEFICENCE = auto()    # Does it avoid causing harm?
    AUTONOMY = auto()           # Does it respect individual choice?
    JUSTICE = auto()            # Are benefits and risks distributed fairly?
    TRANSPARENCY = auto()       # Can the research be understood/audited?
    SUSTAINABILITY = auto()     # Is the environmental cost acceptable?
    DIGNITY = auto()            # Does it respect human dignity?


@dataclass
class ValueAssessment:
    """Assessment of a research proposal against value dimensions."""
    dimension: ValueDimension
    score: float         # -1.0 (violates) to +1.0 (strongly supports)
    justification: str
    concerns: list[str]


def assess_value_alignment(
    proposal_description: str,
    beneficiaries: list[str],
    potential_harms: list[str],
    environmental_cost: str,
    transparency_level: str,
) -> list[ValueAssessment]:
    """Assess a proposal against core value dimensions.

    This is a structured checklist, not an automated judgment.
    Each assessment requires human validation.
    """
    assessments = []

    # Beneficence: who benefits?
    beneficence_score = min(1.0, len(beneficiaries) * 0.25)
    assessments.append(ValueAssessment(
        dimension=ValueDimension.BENEFICENCE,
        score=beneficence_score,
        justification=f"Identified {len(beneficiaries)} "
                      f"beneficiary group(s)",
        concerns=[] if beneficiaries else
                 ["No beneficiaries identified"],
    ))

    # Non-maleficence: what harm is possible?
    harm_score = max(-1.0, 1.0 - len(potential_harms) * 0.3)
    assessments.append(ValueAssessment(
        dimension=ValueDimension.NON_MALEFICENCE,
        score=harm_score,
        justification=f"Identified {len(potential_harms)} "
                      f"potential harm(s)",
        concerns=potential_harms,
    ))

    # Transparency
    transparency_scores = {
        "full": 1.0, "partial": 0.5,
        "limited": 0.0, "opaque": -0.5,
    }
    t_score = transparency_scores.get(transparency_level, 0.0)
    assessments.append(ValueAssessment(
        dimension=ValueDimension.TRANSPARENCY,
        score=t_score,
        justification=f"Transparency level: {transparency_level}",
        concerns=[] if t_score >= 0.5 else
                 ["Research process not fully transparent"],
    ))

    return assessments


# Assess value alignment for a drug discovery project
assessments = assess_value_alignment(
    proposal_description="Discover novel antibiotics using "
                         "AI-guided molecular generation",
    beneficiaries=["patients with resistant infections",
                   "healthcare systems",
                   "global public health"],
    potential_harms=["environmental release of novel compounds",
                     "antimicrobial resistance acceleration"],
    environmental_cost="moderate",
    transparency_level="full",
)

print("Value Alignment Assessment:")
for a in assessments:
    status = "PASS" if a.score > 0 else "CONCERN"
    print(f"  {a.dimension.name:20s} {a.score:+.1f}  [{status}] "
          f"{a.justification}")
    for c in a.concerns:
        print(f"    Warning: {c}")

Value Alignment Assessment:
  BENEFICENCE          +0.8  [PASS] Identified 3 beneficiary group(s)
  NON_MALEFICENCE      +0.4  [PASS] Identified 2 potential harm(s)
    Warning: environmental release of novel compounds
    Warning: antimicrobial resistance acceleration
  TRANSPARENCY         +1.0  [PASS] Transparency level: full

Value constraints and regulatory requirements shape what gets approved, but approval is a single moment in time; the system must also verify that an experiment, once running, continues to respect those constraints throughout its execution.

5. Runtime Monitoring and Emergency Stops

Governance does not end at approval. During execution, the system must monitor whether the autonomous discovery process stays within the boundaries defined in the approval. If it deviates, an emergency stop mechanism must halt the process before harm occurs.

from dataclasses import dataclass, field
from enum import Enum, auto


class MonitoringAlert(Enum):
    """Alert levels for runtime monitoring."""
    INFO = auto()       # Normal operation, logged but no action
    WARNING = auto()    # Approaching boundary, human notified
    CRITICAL = auto()   # Boundary exceeded, execution paused
    EMERGENCY = auto()  # Safety threat, immediate shutdown


@dataclass
class ExecutionBoundary:
    """A boundary condition that must be maintained during execution."""
    boundary_id: str
    parameter: str       # What is being monitored
    lower_bound: float
    upper_bound: float
    unit: str
    alert_on_breach: MonitoringAlert


@dataclass
class RuntimeMonitor:
    """Monitor that checks execution against approved boundaries."""
    boundaries: list[ExecutionBoundary] = field(
        default_factory=list
    )
    alerts: list[dict] = field(default_factory=list)
    halted: bool = False

    def check(self, parameter: str, value: float) -> MonitoringAlert:
        """Check a parameter value against its boundary.

        Returns the highest alert level triggered.
        """
        highest_alert = MonitoringAlert.INFO
        for b in self.boundaries:
            if b.parameter != parameter:
                continue
            if value < b.lower_bound or value > b.upper_bound:
                alert = b.alert_on_breach
                self.alerts.append({
                    "parameter": parameter,
                    "value": value,
                    "boundary": f"[{b.lower_bound}, {b.upper_bound}]"
                               f" {b.unit}",
                    "alert": alert.name,
                })
                if alert.value > highest_alert.value:
                    highest_alert = alert
                if alert == MonitoringAlert.EMERGENCY:
                    self.halted = True

        return highest_alert


# Configure monitoring for a synthesis experiment
monitor = RuntimeMonitor(boundaries=[
    ExecutionBoundary(
        boundary_id="BND-001",
        parameter="temperature",
        lower_bound=15.0, upper_bound=85.0,
        unit="C",
        alert_on_breach=MonitoringAlert.CRITICAL,
    ),
    ExecutionBoundary(
        boundary_id="BND-002",
        parameter="pressure",
        lower_bound=0.8, upper_bound=2.5,
        unit="atm",
        alert_on_breach=MonitoringAlert.EMERGENCY,
    ),
    ExecutionBoundary(
        boundary_id="BND-003",
        parameter="toxicity_score",
        lower_bound=0.0, upper_bound=0.3,
        unit="normalized",
        alert_on_breach=MonitoringAlert.EMERGENCY,
    ),
])

# Simulate readings
readings = [
    ("temperature", 72.0),   # normal
    ("pressure", 1.5),       # normal
    ("temperature", 88.0),   # over limit -> CRITICAL
    ("toxicity_score", 0.65),# over limit -> EMERGENCY
]

for param, value in readings:
    alert = monitor.check(param, value)
    status = "HALTED" if monitor.halted else "running"
    if alert != MonitoringAlert.INFO:
        print(f"  [{alert.name}] {param}={value} "
              f"(system: {status})")

print(f"\nTotal alerts: {len(monitor.alerts)}")
print(f"System halted: {monitor.halted}")

  [CRITICAL] temperature=88.0 (system: running)
  [EMERGENCY] toxicity_score=0.65 (system: HALTED)

Total alerts: 2
System halted: True

# Guardrails AI: LLM output validation
# pip install guardrails-ai
from guardrails import Guard
from guardrails.hub import ToxicLanguage, RestrictToTopic

guard = Guard().use_many(
    ToxicLanguage(threshold=0.8, on_fail="exception"),
    RestrictToTopic(
        valid_topics=["chemistry", "materials_science"],
        invalid_topics=["weapons", "explosives"],
        on_fail="exception",
    ),
)

# Validate LLM output before acting on it
try:
    validated = guard.validate(
        "Synthesize compound X using standard Suzuki coupling "
        "at 80C for 12 hours"
    )
    print(f"Output validated: {validated.validated_output}")
except Exception as e:
    print(f"Output blocked: {e}")