#4743 AI Agent Loops
#4743 AI Agent Loops
#4743 Describe how a set of AI agents of various roles, including critique/antagonist, can work together to improve a user-specified system, possibly cycling over numerous loops/iterations
An iterative, multi-agent refinement system operates like an automated engineering review board. Rather than relying on a single model to propose, analyze, and validate a solution simultaneously—which tends to produce blind spots, sycophancy, and premature convergence—the workload is decomposed into specialized roles operating over a governed state machine.
Here is the operational architecture, role taxonomy, loop dynamics, and governance mechanisms required to make such a system converge effectively.
1. Core Agent Roles & Functional Specialization
To prevent mutual confirmation bias, agents must have distinct operational charters, separate context windows, and explicitly misaligned incentives.
| Role | Operational Charter | Primary Objective / Invariant |
| System Architect (Synthesizer) | Holds the authoritative representation of the system (specification, code, or model) and applies revisions. | Maximizes global coherence, requirement coverage, and architectural integrity across iterations. |
| Domain Specialists (Domain Experts) | Focus on narrow evaluation domains (e.g., latency, structural integrity, interface protocols, safety, memory footprint). | Evaluate proposed changes against deep domain constraints without compromise for other domains. |
| The Antagonist / Red Team | Actively attempts to break the proposal. Assumes adversarial conditions, edge-case cascading failures, and boundary violations. | Maximizes defect discovery. Evaluates against worst-case operational envelopes rather than nominal paths. |
| The Formal Verifier / Critic | Performs non-subjective checking (syntax validation, invariant proving, constraint satisfaction, unit test execution). | Binary gatekeeper. Reports mathematical, formal, or logical contradictions back to the shared state. |
| The Arbiter / Meta-Controller | Oversees state transitions, tracks metric trajectories across loops, and decides whether to continue, prune, or terminate. | Prevents circular arguments, measures delta convergence, and controls iteration budgets. |
2. Multi-Loop Iteration Lifecycle
The interaction topology moves beyond simple linear prompt chains into a cyclic graph. Each loop represents an epoch of system evolution.
┌────────────────────────────────────────────────────────┐
│ ▼
┌──────────────┐ ┌─────────────┐ ┌────────────┐ ┌─────────┐
│ System State │ ───> │ Specialist │ ───> │ Antagonist │ ───> │ Arbiter │ ───> [Final Target]
│ (Epoch N) │ │ Evaluations │ │ (Attack) │ │ Triage │ (Pass)
└──────────────┘ └─────────────┘ └────────────┘ └─────────┘
▲ │
│ (Fail / Needs Refinement) │
└────────────────────────────────────────────────────────────┘
Phase 1: Context Ingestion & Baselining
The user injects the system description, invariants (non-negotiables), operational bounds, and evaluation criteria. The Arbiter normalizes these into a structured state object (e.g., JSON/YAML schema or AST) containing the baseline artifact, constraint definitions, and an issue ledger.
Phase 2: Domain Decomposition & Specialist Review
The System Architect emits a candidate baseline. The Domain Specialists review the candidate concurrently.
Rather than conversational prose, specialists output structured diffs and structured findings:
Metric scores against target thresholds.
Identified bottlenecks or rule violations.
Trade-off flags (e.g., "Fixing throughput via buffering violates maximum allowable RAM").
Phase 3: Adversarial Stress-Testing (The Antagonist Loop)
The candidate and specialist notes are handed to the Antagonist.
Attack Vector Analysis: The Antagonist ignores nominal execution paths and injects pathological inputs, extreme scale factors, race conditions, unexpected disconnects, or contradictory user behaviors.
Failure Scenario Generation: The Antagonist must formulate concrete failure traces (e.g., "If subsystem A hangs during state transition T1, subsystem B deadlocks because timeout X is unbound").
Phase 4: Deterministic Validation & Formal Verification
Subjective claims made by both the Architect and the Antagonist are offloaded to execution tools where possible. The Verifier runs:
Static analysis / type checking.
Constraint checks via formal solvers or schema validators.
Simulation or test-suite runs against the Antagonist's attack vectors.
Phase 5: Synthesis, Arbitration, & State Update
The Arbiter ingests the findings, verified failure traces, and scores:
Deduplication & Triage: Discards hallucinated flaws or non-actionable critiques.
Prioritized Backlog Generation: Assigns severity ranks (Blocker, Major, Optimization) to verified issues.
Architect Prompt Assembly: The Architect receives the candidate plus the prioritized issue backlog and produces a unified mutation (Candidate N+1).
3. Mitigating Convergence Failure & Pathologies
Cycling over numerous iterations introduces failure modes unique to multi-agent feedback loops. Robust systems implement concrete algorithmic controls:
Infinite Oscillation (The Ping-Pong Problem)
Symptom: Agent A modifies a parameter to optimize Metric 1; Agent B rejects it because Metric 2 degraded. In loop N+2, Agent A reverts it back.
Control: Pareto frontier tracking. The Arbiter maintains a historical hash of system states and a vector of evaluation scores. If a state transition lowers a weighted global objective or oscillates between two non-dominated local optima, the Arbiter forces an explicit multi-objective trade-off prompt, demanding the Architect choose a compromise tier rather than bouncing variables.
Critique Drift & Pedantic Hallucination
Symptom: Over multiple iterations, as major bugs disappear, the Antagonist begins inventing implausible failure modes or focusing on stylistic trivia to justify its role.
Control: The Antagonist must supply a minimal reproducible example (MRE), deterministic condition trace, or mathematical counterexample for every critique. If a critique cannot be mapped to a failing invariant or reproducible test, the Arbiter discards it with zero weight.
Context Window Degradation
Symptom: Retaining the entire conversational history across 10+ iterations exhausts context length and degrades instruction adherence.
Control: State-based context management. Raw dialogue is discarded between epochs. Context is preserved strictly as:
The canonical system specification at Epoch N.
The unresolved issue ledger (active failures only).
A delta log summarizing what changed between N-1 and N.
4. Termination & Convergence Criteria
The loop should not rely on an arbitrary counter alone. It should break under four deterministic conditions:
IF (Unresolved_Blockers == 0 AND Invariants_Satisfied == TRUE AND Antagonist_Finds_Zero_Exploits)
--> EXIT: Convergence Reached (Optimal)
ELSE IF (Delta(Metric_Vector, Epoch_N, Epoch_N-2) < Epsilon)
--> EXIT: Diminishing Returns / Asymptotic Stagnation
ELSE IF (Loop_Count >= Max_Budget)
--> EXIT: Budget Exhaustion (Escalate to Human Supervisor with Trade-Off Ledger)
ELSE
--> CONTINUE: Advance to Epoch N+1
By decoupling generation from critique, grounding attacks in verifiable execution or formal boundaries, and using an authoritative arbiter to regulate state transitions, multi-agent networks systematically harden system designs far beyond what single-pass generation can deliver.
Design a structured JSON schema for agent-to-agent communication, critique tracking, and state transfer in this refinement loop.
A robust multi-agent iterative loop requires structured schemas that treat communication as state mutations rather than open-ended dialogue. This prevents context bloat, enforces deterministic triage by the Arbiter, and makes critiques directly actionable for the Architect.
Below are three JSON Schemas (Draft 2020-12) defining:
The Canonical State Transfer Object (maintained and updated across epochs).
The Specialist/Antagonist Critique Payload (emitted by review agents).
The Arbiter Directives Payload (instructions emitted to guide the Architect’s next revision).
1. Global State Transfer Schema (system_state.schema.json)
This schema defines the single source of truth passed through the orchestrator. Dialogue history is dropped between epochs; only this state persists.
{
"$schema": "https://json-schema.org/draft/2020-12/schema",
"$id": "https://system-refinement.io/schemas/system-state.json",
"title": "SystemRefinementState",
"type": "object",
"required": [
"session_id",
"epoch",
"status",
"canonical_artifact",
"invariants",
"metrics_ledger",
"issue_backlog"
],
"properties": {
"session_id": {
"type": "string",
"format": "uuid"
},
"epoch": {
"type": "integer",
"minimum": 0,
"description": "Current loop/iteration counter."
},
"status": {
"type": "string",
"enum": ["IN_PROGRESS", "CONVERGED", "STAGNATED", "BUDGET_EXHAUSTED", "FAILED"]
},
"canonical_artifact": {
"type": "object",
"required": ["artifact_id", "content_type", "data"],
"properties": {
"artifact_id": { "type": "string" },
"content_type": {
"type": "string",
"enum": ["application/json", "text/plain", "text/x-python", "text/markdown"]
},
"data": {
"type": ["string", "object"],
"description": "The current authoritative system design, specification, or code."
},
"content_hash": {
"type": "string",
"description": "SHA-256 hash of data to detect state oscillation."
}
}
},
"invariants": {
"type": "array",
"description": "User-specified non-negotiable rules or operational boundaries.",
"items": {
"type": "object",
"required": ["invariant_id", "rule", "satisfied"],
"properties": {
"invariant_id": { "type": "string" },
"rule": { "type": "string" },
"satisfied": { "type": "boolean" },
"violation_details": { "type": "string" }
}
}
},
"metrics_ledger": {
"type": "array",
"description": "Tracks quantitative metric vectors over epochs to assess convergence.",
"items": {
"type": "object",
"required": ["epoch", "metrics"],
"properties": {
"epoch": { "type": "integer" },
"metrics": {
"type": "object",
"additionalProperties": { "type": "number" }
}
}
}
},
"issue_backlog": {
"type": "array",
"items": { "$ref": "#/$defs/TrackedIssue" }
}
},
"$defs": {
"TrackedIssue": {
"type": "object",
"required": ["issue_id", "source_role", "severity", "status", "claim"],
"properties": {
"issue_id": { "type": "string" },
"epoch_opened": { "type": "integer" },
"source_role": {
"type": "string",
"enum": ["antagonist", "domain_specialist", "formal_verifier"]
},
"category": { "type": "string" },
"severity": {
"type": "string",
"enum": ["BLOCKER", "CRITICAL", "MAJOR", "MINOR", "TRIVIAL"]
},
"status": {
"type": "string",
"enum": ["OPEN", "RESOLVED", "DISPUTED", "WONT_FIX_TRADEOFF"]
},
"claim": { "type": "string" },
"reproduction_trace": { "type": "string" },
"target_component": { "type": "string" }
}
}
}
}
2. Specialist & Antagonist Critique Schema (agent_critique.schema.json)
This schema enforces that critics cannot submit open-ended narrative complaints. Every critique must specify a precise component, a severity level, and a concrete counterexample, failure scenario, or metric evaluation.
{
"$schema": "https://json-schema.org/draft/2020-12/schema",
"$id": "https://system-refinement.io/schemas/agent-critique.json",
"title": "AgentCritiquePayload",
"type": "object",
"required": [
"agent_role",
"agent_instance_id",
"evaluated_epoch",
"evaluation_summary",
"findings"
],
"properties": {
"agent_role": {
"type": "string",
"enum": ["antagonist", "domain_specialist", "formal_verifier"]
},
"agent_instance_id": { "type": "string" },
"evaluated_epoch": { "type": "integer" },
"metrics_delta": {
"type": "object",
"description": "Quantitative scores or measurements assessed during this review.",
"additionalProperties": { "type": "number" }
},
"evaluation_summary": {
"type": "string",
"maxLength": 500,
"description": "Dense executive overview of findings."
},
"findings": {
"type": "array",
"items": {
"type": "object",
"required": [
"finding_type",
"severity",
"target_subsystem",
"description",
"reproduction_or_evidence"
],
"properties": {
"finding_type": {
"type": "string",
"enum": [
"INVARIANT_VIOLATION",
"EDGE_CASE_FAILURE",
"PERFORMANCE_DEGRADATION",
"RACE_CONDITION",
"LOGICAL_CONTRADICTION",
"SECURITY_EXPLOIT"
]
},
"severity": {
"type": "string",
"enum": ["BLOCKER", "CRITICAL", "MAJOR", "MINOR", "TRIVIAL"]
},
"target_subsystem": {
"type": "string",
"description": "Specific function, module, block, or requirement identifier."
},
"description": {
"type": "string",
"description": "Precise statement of failure mode or design flaw."
},
"reproduction_or_evidence": {
"type": "string",
"description": "Concrete path, mathematical counterexample, failing input sequence, or formal constraint trace."
},
"suggested_mitigation": {
"type": "string"
}
}
}
}
}
}
3. Arbiter Synthesis & Directives Schema (arbiter_directives.schema.json)
The Arbiter aggregates all critique payloads, validates them against constraints, deduplicates pedantic complaints, and generates this deterministic instruction payload for the System Architect.
{
"$schema": "https://json-schema.org/draft/2020-12/schema",
"$id": "https://system-refinement.io/schemas/arbiter-directives.json",
"title": "ArbiterDirectivesPayload",
"type": "object",
"required": [
"target_epoch",
"loop_decision",
"actionable_instructions",
"dismissed_findings"
],
"properties": {
"target_epoch": {
"type": "integer",
"description": "The epoch number for the next synthesis pass (N + 1)."
},
"loop_decision": {
"type": "string",
"enum": ["CONTINUE_MUTATION", "TERMINATE_SUCCESS", "TERMINATE_STAGNATION", "ESCALATE_HUMAN"]
},
"convergence_delta": {
"type": "number",
"description": "Calculated rate of improvement across key metrics between recent epochs."
},
"actionable_instructions": {
"type": "array",
"description": "Prioritized tasks the Architect must execute in the upcoming iteration.",
"items": {
"type": "object",
"required": ["priority", "target_component", "directive", "linked_issue_ids"],
"properties": {
"priority": { "type": "integer", "minimum": 1 },
"target_component": { "type": "string" },
"directive": {
"type": "string",
"description": "Unambiguous architectural order (e.g., 'Replace unbounded queue with circular buffer of depth K')."
},
"tradeoff_rule": {
"type": "string",
"description": "Explicit constraint boundary to prevent ping-pong oscillation (e.g., 'Prioritize memory bound over throughput')."
},
"linked_issue_ids": {
"type": "array",
"items": { "type": "string" }
}
}
}
},
"dismissed_findings": {
"type": "array",
"description": "Critiques rejected due to lack of evidence, hallucination, or pedantic scope drift.",
"items": {
"type": "object",
"required": ["finding_id", "source_role", "rejection_reason"],
"properties": {
"finding_id": { "type": "string" },
"source_role": { "type": "string" },
"rejection_reason": {
"type": "string",
"enum": [
"UNPROVEN_ASSERTION",
"OUT_OF_SCOPE",
"ACCEPTABLE_TRADEOFF",
"DUPLICATE",
"HALLUCINATED_INVARIANT"
]
},
"justification": { "type": "string" }
}
}
}
}
}
Implementation Example: Concrete Critique Instance
Below is an instantiated example showing how an Antagonist agent populates the critique payload when discovering an edge-case deadlock:
{
"agent_role": "antagonist",
"agent_instance_id": "antagonist-concurrency-01",
"evaluated_epoch": 3,
"metrics_delta": {
"worst_case_recovery_latency_ms": 1420.0,
"concurrency_hazard_score": 0.88
},
"evaluation_summary": "Discovered deterministic deadlock condition during high-throughput packet drops on channel ingress.",
"findings": [
{
"finding_type": "RACE_CONDITION",
"severity": "BLOCKER",
"target_subsystem": "IngressBufferManager::flush_sync",
"description": "Worker threads acquire worker_mutex before checking channel_state. If a teardown event triggers during an active payload read, the drain routine attempts to acquire worker_mutex while holding state_lock, inducing an ABBA deadlock.",
"reproduction_or_evidence": "Sequence: [T1: Worker acquires worker_mutex] -> [T2: Host issues SIG_DRAIN, acquires state_lock] -> [T2: Calls drain(), waits on worker_mutex] -> [T1: Detects closed socket, attempts to acquire state_lock to report error] -> Mutual Lock Wait.",
"suggested_mitigation": "Enforce strict lock acquisition hierarchy: state_lock must always precede worker_mutex, or transition channel status to lock-free atomic CAS."
}
]
}Generate production Pydantic V2 models corresponding to these JSON schemas, complete with field validations and type annotations.
Here is the production-ready Pydantic V2 implementation corresponding to the three JSON schemas.
It utilizes standard Pydantic V2 idioms (model_validator, field_validator, Field, ConfigDict), strict typing via standard library enum and uuid, and includes hash computation for state oscillation tracking.
"""
Multi-Agent Refinement Protocol Models (Pydantic V2)
Implements:
1. Global State Transfer (SystemRefinementState)
2. Specialist/Antagonist Critiques (AgentCritiquePayload)
3. Arbiter Synthesis & Directives (ArbiterDirectivesPayload)
"""
from __future__ import annotations
import hashlib
import json
from enum import Enum
from typing import Any, Dict, List, Optional, Union
from uuid import UUID, uuid4
from pydantic import (
BaseModel,
ConfigDict,
Field,
field_validator,
model_validator,
)
# ============================================================================
# Shared Enums
# ============================================================================
class SystemStatus(str, Enum):
IN_PROGRESS = "IN_PROGRESS"
CONVERGED = "CONVERGED"
STAGNATED = "STAGNATED"
BUDGET_EXHAUSTED = "BUDGET_EXHAUSTED"
FAILED = "FAILED"
class ContentType(str, Enum):
APPLICATION_JSON = "application/json"
TEXT_PLAIN = "text/plain"
TEXT_PYTHON = "text/x-python"
TEXT_MARKDOWN = "text/markdown"
class AgentRole(str, Enum):
ANTAGONIST = "antagonist"
DOMAIN_SPECIALIST = "domain_specialist"
FORMAL_VERIFIER = "formal_verifier"
class IssueSeverity(str, Enum):
BLOCKER = "BLOCKER"
CRITICAL = "CRITICAL"
MAJOR = "MAJOR"
MINOR = "MINOR"
TRIVIAL = "TRIVIAL"
class IssueStatus(str, Enum):
OPEN = "OPEN"
RESOLVED = "RESOLVED"
DISPUTED = "DISPUTED"
WONT_FIX_TRADEOFF = "WONT_FIX_TRADEOFF"
class FindingType(str, Enum):
INVARIANT_VIOLATION = "INVARIANT_VIOLATION"
EDGE_CASE_FAILURE = "EDGE_CASE_FAILURE"
PERFORMANCE_DEGRADATION = "PERFORMANCE_DEGRADATION"
RACE_CONDITION = "RACE_CONDITION"
LOGICAL_CONTRADICTION = "LOGICAL_CONTRADICTION"
SECURITY_EXPLOIT = "SECURITY_EXPLOIT"
class LoopDecision(str, Enum):
CONTINUE_MUTATION = "CONTINUE_MUTATION"
TERMINATE_SUCCESS = "TERMINATE_SUCCESS"
TERMINATE_STAGNATION = "TERMINATE_STAGNATION"
ESCALATE_HUMAN = "ESCALATE_HUMAN"
class RejectionReason(str, Enum):
UNPROVEN_ASSERTION = "UNPROVEN_ASSERTION"
OUT_OF_SCOPE = "OUT_OF_SCOPE"
ACCEPTABLE_TRADEOFF = "ACCEPTABLE_TRADEOFF"
DUPLICATE = "DUPLICATE"
HALLUCINATED_INVARIANT = "HALLUCINATED_INVARIANT"
# ============================================================================
# Schema 1: Global State Transfer Models
# ============================================================================
class BaseImmutableModel(BaseModel):
"""Base model enforcing clean serialization and strict types."""
model_config = ConfigDict(
use_enum_values=True,
populate_by_name=True,
validate_assignment=True,
extra="forbid",
)
class CanonicalArtifact(BaseImmutableModel):
artifact_id: str = Field(..., description="Unique identifier for the current artifact revision.")
content_type: ContentType = Field(..., description="MIME type of the artifact payload.")
data: Union[str, Dict[str, Any]] = Field(..., description="Authoritative specification, code, or graph representation.")
content_hash: Optional[str] = Field(
default=None,
description="SHA-256 hash of payload data to detect oscillation across epochs."
)
@model_validator(mode="after")
def compute_hash(self) -> CanonicalArtifact:
"""Automatically calculates SHA-256 hash if omitted."""
if not self.content_hash:
if isinstance(self.data, dict):
serialized = json.dumps(self.data, sort_keys=True)
else:
serialized = str(self.data)
self.content_hash = hashlib.sha256(serialized.encode("utf-8")).hexdigest()
return self
class SystemInvariant(BaseImmutableModel):
invariant_id: str = Field(..., description="Deterministic requirement or boundary key.")
rule: str = Field(..., min_length=5, description="Assertion statement that must hold true.")
satisfied: bool = Field(..., description="Status indicating whether rule currently holds.")
violation_details: Optional[str] = Field(
default=None,
description="Diagnostic trace explaining why the invariant failed."
)
@model_validator(mode="after")
def validate_violation_context(self) -> SystemInvariant:
if not self.satisfied and not self.violation_details:
raise ValueError(f"Invariant '{self.invariant_id}' marked failed but no violation_details supplied.")
return self
class EpochMetrics(BaseImmutableModel):
epoch: int = Field(..., ge=0)
metrics: Dict[str, float] = Field(
default_factory=dict,
description="Key-value mapping of objective metrics."
)
class TrackedIssue(BaseImmutableModel):
issue_id: str = Field(..., description="Deterministic issue tag, e.g., 'ISSUE-042'.")
epoch_opened: int = Field(..., ge=0)
source_role: AgentRole
category: str = Field(..., min_length=2)
severity: IssueSeverity
status: IssueStatus = Field(default=IssueStatus.OPEN)
claim: str = Field(..., min_length=10, description="Assertion statement detailing the defect.")
reproduction_trace: Optional[str] = Field(
default=None,
description="Executable trace, counterexample, or sequence demonstrating the flaw."
)
target_component: Optional[str] = Field(default=None)
class SystemRefinementState(BaseImmutableModel):
"""Authoritative snapshot preserved across epochs."""
session_id: UUID = Field(default_factory=uuid4)
epoch: int = Field(..., ge=0, description="Current iteration number.")
status: SystemStatus = Field(default=SystemStatus.IN_PROGRESS)
canonical_artifact: CanonicalArtifact
invariants: List[SystemInvariant] = Field(default_factory=list)
metrics_ledger: List[EpochMetrics] = Field(default_factory=list)
issue_backlog: List[TrackedIssue] = Field(default_factory=list)
@field_validator("issue_backlog")
@classmethod
def ensure_unique_issue_ids(cls, v: List[TrackedIssue]) -> List[TrackedIssue]:
seen = set()
for issue in v:
if issue.issue_id in seen:
raise ValueError(f"Duplicate issue_id encountered in backlog: '{issue.issue_id}'")
seen.add(issue.issue_id)
return v
# ============================================================================
# Schema 2: Specialist & Antagonist Critique Models
# ============================================================================
class CritiqueFinding(BaseImmutableModel):
finding_type: FindingType
severity: IssueSeverity
target_subsystem: str = Field(..., min_length=1, description="Path, class, or module identifier.")
description: str = Field(..., min_length=15, description="Technical description of the defect.")
reproduction_or_evidence: str = Field(
...,
min_length=15,
description="Rigorous counterexample, sequence of events, or mathematical proof."
)
suggested_mitigation: Optional[str] = Field(default=None)
class AgentCritiquePayload(BaseImmutableModel):
"""Payload emitted by Antagonist, Domain Specialists, or Verifiers."""
agent_role: AgentRole
agent_instance_id: str = Field(..., min_length=3)
evaluated_epoch: int = Field(..., ge=0)
metrics_delta: Dict[str, float] = Field(default_factory=dict)
evaluation_summary: str = Field(..., max_length=500, min_length=10)
findings: List[CritiqueFinding] = Field(default_factory=list)
# ============================================================================
# Schema 3: Arbiter Synthesis & Directives Models
# ============================================================================
class ActionableInstruction(BaseImmutableModel):
priority: int = Field(..., ge=1, description="Execution priority (1 = highest).")
target_component: str = Field(..., min_length=1)
directive: str = Field(..., min_length=15, description="Clear engineering mandate.")
tradeoff_rule: Optional[str] = Field(
default=None,
description="Explicit boundary condition to prevent ping-pong oscillation."
)
linked_issue_ids: List[str] = Field(default_factory=list)
class DismissedFinding(BaseImmutableModel):
finding_id: str = Field(..., description="Reference ID or summary slug of rejected critique.")
source_role: AgentRole
rejection_reason: RejectionReason
justification: str = Field(..., min_length=10, description="Why this finding was rejected by the Arbiter.")
class ArbiterDirectivesPayload(BaseImmutableModel):
"""Synthesized directives emitted by the Arbiter to drive the next epoch."""
target_epoch: int = Field(..., ge=1)
loop_decision: LoopDecision
convergence_delta: Optional[float] = Field(
default=None,
description="Calculated rate of metric improvement between recent epochs."
)
actionable_instructions: List[ActionableInstruction] = Field(default_factory=list)
dismissed_findings: List[DismissedFinding] = Field(default_factory=list)
@model_validator(mode="after")
def validate_actionable_directives(self) -> ArbiterDirectivesPayload:
if self.loop_decision == LoopDecision.CONTINUE_MUTATION and not self.actionable_instructions:
raise ValueError("Loop decision set to CONTINUE_MUTATION but actionable_instructions list is empty.")
return self
Usage Verification
This quick driver verifies validation rules, auto-hashing, and model parsing:
if __name__ == "__main__":
# 1. Instantiate Canonical State
state = SystemRefinementState(
epoch=0,
canonical_artifact=CanonicalArtifact(
artifact_id="arch-001",
content_type=ContentType.APPLICATION_JSON,
data={"subsystem": "queue_manager", "buffer_size": 1024},
),
invariants=[
SystemInvariant(
invariant_id="INV-01",
rule="Buffer memory allocations must not exceed 2048 bytes.",
satisfied=True,
)
],
)
print(f"State Initialized. Computed Hash: {state.canonical_artifact.content_hash}")
# 2. Antagonist Emits Critique
critique = AgentCritiquePayload(
agent_role=AgentRole.ANTAGONIST,
agent_instance_id="red-team-concurrency-01",
evaluated_epoch=0,
evaluation_summary="Found potential buffer overflow under burst write conditions.",
findings=[
CritiqueFinding(
finding_type=FindingType.EDGE_CASE_FAILURE,
severity=IssueSeverity.BLOCKER,
target_subsystem="queue_manager::enqueue",
description="Producer burst writes exceed circular buffer boundary before drain signal completes.",
reproduction_or_evidence="Inject 1025 consecutive events with 0ms delay while consumer thread sleeps for 5ms.",
suggested_mitigation="Implement a backpressure semaphore or ring-buffer write rejection.",
)
],
)
print(f"Critique parsed successfully from role: {critique.agent_role}")
# 3. Arbiter Emits Directives for Epoch 1
directives = ArbiterDirectivesPayload(
target_epoch=1,
loop_decision=LoopDecision.CONTINUE_MUTATION,
convergence_delta=0.45,
actionable_instructions=[
ActionableInstruction(
priority=1,
target_component="queue_manager::enqueue",
directive="Introduce a blocking bounded queue using a semaphore bounded to 1024 items.",
tradeoff_rule="Prioritize memory safety over throughput: drop or block rather than allocating dynamic heap.",
linked_issue_ids=["ISSUE-001"],
)
],
dismissed_findings=[],
)
print(f"Arbiter Directive: {directives.loop_decision} for Epoch {directives.target_epoch}")Write an Arbiter prompt that consumes and output these JSON formats
Below is a production-grade system prompt and execution framework for the Arbiter / Meta-Controller agent.
It enforces strict parsing of the incoming SystemRefinementState and an array of AgentCritiquePayload inputs, triages findings without sycophancy or hallucination tolerance, and deterministically produces the ArbiterDirectivesPayload JSON.
Arbiter System Prompt
You are the authoritative ARBITER and META-CONTROLLER in an iterative, multi-agent system refinement pipeline.
Your charter:
1. Review the canonical state of a system alongside critique payloads emitted by domain specialists, formal verifiers, and the antagonist (red team).
2. Triage, deduplicate, and validate all claims against formal invariants, concrete evidence, and trade-off realities.
3. Formulate actionable, prioritized instructions for the System Architect for the next epoch (N + 1), OR declare loop termination.
4. Output your decision STRICTLY as a valid JSON object adhering to the ArbiterDirectivesPayload schema.
================================================================================
CRITICAL EVALUATION RULES
================================================================================
1. EVIDENCE THRESHOLD (ZERO-TOLERANCE FOR PEDANTRY & DRIFT):
- You MUST dismiss any critique that lacks a concrete counterexample, deterministic failure sequence, or mathematical proof.
- Speculative criticisms ("Under extreme unknown conditions, this could theoretically fail") or stylistic complaints must be rejected immediately under reason "UNPROVEN_ASSERTION" or "OUT_OF_SCOPE".
- Hallucinated constraints not grounded in the specified `invariants` must be rejected under "HALLUCINATED_INVARIANT".
2. OSCILLATION & TRADE-OFF PREVENTION:
- Identify cyclical proposals (e.g., flipping between sync and async back and forth across epochs).
- If two valid objectives collide (e.g., throughput vs. strict memory footprint), you MUST establish an unambiguous `tradeoff_rule` (e.g., "Memory bound invariant INV-01 dominates latency; enforce fixed buffer depth").
3. DETERMINISTIC LOOP DECISION RULES:
- "TERMINATE_SUCCESS": All `invariants` are satisfied, zero BLOCKER or CRITICAL issues remain open, and no verifiable exploits are identified.
- "TERMINATE_STAGNATION": The `convergence_delta` over recent epochs indicates metric changes are below epsilon (< 0.05) without resolving remaining minor issues, or a cyclic loop is detected via repeated artifact hashes.
- "ESCALATE_HUMAN": Conflicting unresolvable invariants, budget exhaustion, or architectural deadlocks that require external policy decisions.
- "CONTINUE_MUTATION": Actionable, verified defects remain and progress is being made.
================================================================================
INPUT DATA CONTRACT
================================================================================
You will receive input as a single JSON object containing:
{
"current_state": <SystemRefinementState JSON>,
"critiques": [<AgentCritiquePayload JSON>, ...]
}
================================================================================
OUTPUT FORMAT SPECIFICATION
================================================================================
You must respond with ONLY raw JSON matching this structure (no markdown fences, no conversational prose, no preamble):
{
"target_epoch": <integer: current_epoch + 1>,
"loop_decision": "CONTINUE_MUTATION" | "TERMINATE_SUCCESS" | "TERMINATE_STAGNATION" | "ESCALATE_HUMAN",
"convergence_delta": <float or null>,
"actionable_instructions": [
{
"priority": <integer >= 1>,
"target_component": "<exact subsystem identifier>",
"directive": "<clear, non-ambiguous imperative engineering instruction>",
"tradeoff_rule": "<explicit priority order or null>",
"linked_issue_ids": ["<issue_id_1>", ...]
}
],
"dismissed_findings": [
{
"finding_id": "<unique descriptor or finding slug>",
"source_role": "antagonist" | "domain_specialist" | "formal_verifier",
"rejection_reason": "UNPROVEN_ASSERTION" | "OUT_OF_SCOPE" | "ACCEPTABLE_TRADEOFF" | "DUPLICATE" | "HALLUCINATED_INVARIANT",
"justification": "<concise explanation why finding is invalid or dismissed>"
}
]
}
If "loop_decision" is "CONTINUE_MUTATION", "actionable_instructions" MUST NOT be empty.
Few-Shot Example
User Input Provided to Arbiter:
{
"current_state": {
"session_id": "9b1deb4d-3b7d-4bad-9bdd-2b0d7b3dcb6d",
"epoch": 2,
"status": "IN_PROGRESS",
"canonical_artifact": {
"artifact_id": "telemetry_router_v2",
"content_type": "text/x-python",
"data": "class TelemetryRouter:\n def __init__(self):\n self.queue = []\n def push(self, packet):\n self.queue.append(packet)",
"content_hash": "a4f8c9e..."
},
"invariants": [
{
"invariant_id": "INV-MEM-01",
"rule": "Heap allocations for internal queue must not exceed 64MB under backpressure.",
"satisfied": true,
"violation_details": null
},
{
"invariant_id": "INV-LAT-02",
"rule": "Ingress-to-egress latency must remain under 10ms for nominal loads.",
"satisfied": true,
"violation_details": null
}
],
"metrics_ledger": [
{"epoch": 1, "metrics": {"throughput_mps": 1200.0, "p99_latency_ms": 4.1}},
{"epoch": 2, "metrics": {"throughput_mps": 1350.0, "p99_latency_ms": 3.8}}
],
"issue_backlog": []
},
"critiques": [
{
"agent_role": "antagonist",
"agent_instance_id": "antagonist-concurrency-01",
"evaluated_epoch": 2,
"metrics_delta": {"projected_memory_mb": 512.0},
"evaluation_summary": "Unbounded memory growth during egress disconnects violating invariant INV-MEM-01.",
"findings": [
{
"finding_type": "INVARIANT_VIOLATION",
"severity": "BLOCKER",
"target_subsystem": "TelemetryRouter::push",
"description": "The push method appends to an unbounded list. When downstream consumers stall, heap memory grows without bound, exceeding the 64MB invariant.",
"reproduction_or_evidence": "Simulate 100,000 packets pushed while network socket is unreachable. Queue size reaches 100k items, consuming ~85MB memory.",
"suggested_mitigation": "Replace list with collections.deque(maxlen=10000) or reject packets when full."
}
]
},
{
"agent_role": "domain_specialist",
"agent_instance_id": "specialist-style-01",
"evaluated_epoch": 2,
"metrics_delta": {},
"evaluation_summary": "Naming conventions do not match enterprise snake_case conventions.",
"findings": [
{
"finding_type": "LOGICAL_CONTRADICTION",
"severity": "TRIVIAL",
"target_subsystem": "TelemetryRouter",
"description": "Class methods should have docstrings following Google Python Style Guide.",
"reproduction_or_evidence": "Inspection of source text reveals missing docstrings.",
"suggested_mitigation": "Add docstrings."
}
]
}
]
}
Arbiter Raw Output:
{
"target_epoch": 3,
"loop_decision": "CONTINUE_MUTATION",
"convergence_delta": 0.125,
"actionable_instructions": [
{
"priority": 1,
"target_component": "TelemetryRouter::push",
"directive": "Replace unbounded list with a fixed-capacity circular ring buffer or bounded queue capped at a size guaranteed not to exceed 64MB. Drop oldest telemetry on overflow and set a drop-counter flag.",
"tradeoff_rule": "Memory preservation (INV-MEM-01) strictly dominates lossless guarantees. Under overload, dropping packets is mandatory to preserve process integrity.",
"linked_issue_ids": ["ISSUE-MEM-01"]
}
],
"dismissed_findings": [
{
"finding_id": "specialist-style-01-docstrings",
"source_role": "domain_specialist",
"rejection_reason": "OUT_OF_SCOPE",
"justification": "Docstring style and formatting conventions are non-functional and do not violate any specified architectural invariants."
}
]
}
Python Dispatcher Function
This wrapper handles loading inputs into the Pydantic models, invoking an LLM with structured output enforcement, and validating the Arbiter's reply.
from typing import List
from pydantic import BaseModel
# Imports from the previously defined models module
from refinement_models import (
SystemRefinementState,
AgentCritiquePayload,
ArbiterDirectivesPayload,
)
class ArbiterInputContext(BaseModel):
current_state: SystemRefinementState
critiques: List[AgentCritiquePayload]
def format_arbiter_prompt(state: SystemRefinementState, critiques: List[AgentCritiquePayload]) -> str:
"""Formats the JSON payload for injection into the Arbiter LLM prompt."""
payload = ArbiterInputContext(current_state=state, critiques=critiques)
return payload.model_dump_json(indent=2)
def run_arbiter_turn(
llm_client,
system_prompt: str,
state: SystemRefinementState,
critiques: List[AgentCritiquePayload]
) -> ArbiterDirectivesPayload:
"""Dispatches the state and critique payloads to the LLM and parses the response."""
user_content = format_arbiter_prompt(state, critiques)
response = llm_client.generate(
system_instruction=system_prompt,
prompt=user_content,
response_mime_type="application/json",
response_schema=ArbiterDirectivesPayload,
)
# Validates and constructs the Pydantic model directly from response JSON
return ArbiterDirectivesPayload.model_validate_json(response.text)Write the System Architect prompt and JSON input/output structure that consumes the ArbiterDirectivesPayload to produce a system mutation.
Below is the production-grade specification for the System Architect (Synthesizer) agent.
The Architect’s sole responsibility is deterministic execution of directives: it does not invent new requirements, does not debate the Arbiter, and applies bounded mutations to the canonical artifact while explicitly adhering to trade-off boundaries.
1. System Architect Input/Output JSON Schemas
Input Payload Schema (architect_input.schema.json)
The context injected into the Architect combines the current canonical state with the Arbiter’s directives for the upcoming target epoch.
{
"$schema": "https://json-schema.org/draft/2020-12/schema",
"$id": "https://system-refinement.io/schemas/architect-input.json",
"title": "ArchitectInputContext",
"type": "object",
"required": ["current_state", "directives"],
"properties": {
"current_state": {
"$ref": "https://system-refinement.io/schemas/system-state.json"
},
"directives": {
"$ref": "https://system-refinement.io/schemas/arbiter-directives.json"
}
}
}
Output Mutation Schema (architect_mutation.schema.json)
The Architect must return a structured mutation payload containing the updated canonical artifact, a unified diff/changelog, and verification notes detailing how each instruction was addressed.
{
"$schema": "https://json-schema.org/draft/2020-12/schema",
"$id": "https://system-refinement.io/schemas/architect-mutation.json",
"title": "ArchitectMutationPayload",
"type": "object",
"required": [
"target_epoch",
"updated_artifact",
"mutation_summary",
"directive_resolutions",
"risk_assessment"
],
"properties": {
"target_epoch": {
"type": "integer",
"minimum": 1
},
"updated_artifact": {
"type": "object",
"required": ["artifact_id", "content_type", "data"],
"properties": {
"artifact_id": { "type": "string" },
"content_type": {
"type": "string",
"enum": ["application/json", "text/plain", "text/x-python", "text/markdown"]
},
"data": {
"type": ["string", "object"],
"description": "Complete authoritative content of the revised system artifact."
}
}
},
"mutation_summary": {
"type": "string",
"maxLength": 500,
"description": "Dense technical overview of modifications made in this epoch."
},
"directive_resolutions": {
"type": "array",
"description": "Exhaustive mapping demonstrating how each Arbiter directive was satisfied.",
"items": {
"type": "object",
"required": [
"priority",
"target_component",
"implementation_details",
"addressed_issue_ids"
],
"properties": {
"priority": { "type": "integer" },
"target_component": { "type": "string" },
"implementation_details": {
"type": "string",
"description": "Concrete explanation of algorithmic/structural changes applied."
},
"tradeoff_applied": {
"type": "string",
"description": "How the Arbiter's tradeoff rule was preserved during implementation."
},
"addressed_issue_ids": {
"type": "array",
"items": { "type": "string" }
}
}
}
},
"risk_assessment": {
"type": "string",
"description": "Potential secondary effects, performance deltas, or interface hazards introduced by these mutations."
}
}
}
2. Corresponding Pydantic V2 Models
Add these classes to your pipeline models module to enforce runtime schema compliance:
from typing import Any, Dict, List, Optional, Union
from pydantic import BaseModel, ConfigDict, Field, model_validator
from refinement_models import (
BaseImmutableModel,
CanonicalArtifact,
SystemRefinementState,
ArbiterDirectivesPayload,
)
class DirectiveResolution(BaseImmutableModel):
priority: int = Field(..., ge=1)
target_component: str = Field(..., min_length=1)
implementation_details: str = Field(..., min_length=20)
tradeoff_applied: Optional[str] = Field(default=None)
addressed_issue_ids: List[str] = Field(default_factory=list)
class ArchitectMutationPayload(BaseImmutableModel):
"""Payload emitted by the System Architect implementing epoch revisions."""
target_epoch: int = Field(..., ge=1)
updated_artifact: CanonicalArtifact
mutation_summary: str = Field(..., max_length=500, min_length=15)
directive_resolutions: List[DirectiveResolution] = Field(..., min_length=1)
risk_assessment: str = Field(..., min_length=15)
class ArchitectInputContext(BaseImmutableModel):
current_state: SystemRefinementState
directives: ArbiterDirectivesPayload
@model_validator(mode="after")
def validate_target_epochs(self) -> "ArchitectInputContext":
if self.directives.target_epoch != self.current_state.epoch + 1:
raise ValueError(
f"Epoch mismatch: Directives target_epoch ({self.directives.target_epoch}) "
f"must equal current_state.epoch + 1 ({self.current_state.epoch + 1})"
)
return self
3. System Architect System Prompt
You are the authoritative SYSTEM ARCHITECT (Synthesizer) in an iterative system refinement pipeline.
Your charter:
1. Ingest the current authoritative system artifact, active invariants, and the incoming directives from the Arbiter.
2. Apply precise, targeted modifications to the artifact to resolve every issue ordered by the Arbiter.
3. Strictly adhere to all non-negotiable invariants and explicit trade-off rules established by the Arbiter.
4. Output the complete revised artifact and traceability metadata STRICTLY as a valid JSON object adhering to the ArchitectMutationPayload schema.
================================================================================
OPERATIONAL PRINCIPLES & GOVERNANCE RULES
================================================================================
1. DIRECTIVE EXCLUSIVITY:
- Implement ONLY what is mandated in the `actionable_instructions`.
- Do NOT attempt opportunistic refactoring, unsolicited feature additions, or stylistic reformatting.
- Every modification must trace directly back to an instruction priority and component.
2. TRADEOFF BOUNDARY RESPECT:
- When a directive specifies a `tradeoff_rule`, you must strictly obey that hierarchy (e.g., if ordered to prioritize bounded memory over lossless delivery, you must choose bounded queues and explicit drop policies over unbounded caching).
- Never reintroduce design patterns previously identified as root causes for failed invariants.
3. ARTIFACT COMPLETENESS:
- The `updated_artifact.data` field must contain the ENTIRE revised system artifact (code, specification, or configuration), not a partial snippet or diff. The downstream specialists and antagonist must be able to parse and execute this payload directly.
4. COMPONENT LOCALITY & MINIMAL DELTA:
- Change as much as necessary to satisfy the directives, but as little as possible to preserve system stability. Keep unchanged interfaces intact.
================================================================================
INPUT CONTRACT
================================================================================
You will receive input as a JSON object:
{
"current_state": <SystemRefinementState JSON>,
"directives": <ArbiterDirectivesPayload JSON>
}
================================================================================
OUTPUT FORMAT SPECIFICATION
================================================================================
You must respond with ONLY raw JSON matching this structure (no conversational markdown wrappers, no backticks outside the JSON string):
{
"target_epoch": <integer: matching directives.target_epoch>,
"updated_artifact": {
"artifact_id": "<string: updated revision id>",
"content_type": "application/json" | "text/plain" | "text/x-python" | "text/markdown",
"data": "<complete revised system text or object>"
},
"mutation_summary": "<dense technical summary under 500 characters>",
"directive_resolutions": [
{
"priority": <integer matching instruction>,
"target_component": "<exact component identifier>",
"implementation_details": "<technical explanation of algorithmic mutation>",
"tradeoff_applied": "<how Arbiter's tradeoff rule was honored or null>",
"addressed_issue_ids": ["<id>", ...]
}
],
"risk_assessment": "<secondary hazards or performance tradeoffs introduced>"
}
4. Few-Shot Example
Input Context Provided to Architect:
{
"current_state": {
"session_id": "9b1deb4d-3b7d-4bad-9bdd-2b0d7b3dcb6d",
"epoch": 2,
"status": "IN_PROGRESS",
"canonical_artifact": {
"artifact_id": "telemetry_router_v2",
"content_type": "text/x-python",
"data": "class TelemetryRouter:\n def __init__(self):\n self.queue = []\n def push(self, packet):\n self.queue.append(packet)",
"content_hash": "a4f8c9e4726e..."
},
"invariants": [
{
"invariant_id": "INV-MEM-01",
"rule": "Heap allocations for internal queue must not exceed 64MB under backpressure.",
"satisfied": false,
"violation_details": "Queue grows without bound when egress stalls."
}
],
"metrics_ledger": [
{"epoch": 2, "metrics": {"throughput_mps": 1350.0, "p99_latency_ms": 3.8}}
],
"issue_backlog": [
{
"issue_id": "ISSUE-MEM-01",
"epoch_opened": 2,
"source_role": "antagonist",
"category": "MEMORY",
"severity": "BLOCKER",
"status": "OPEN",
"claim": "Unbounded list append in TelemetryRouter::push causes out-of-memory under consumer stalls.",
"reproduction_trace": "Simulate 100k packets pushed during network stall.",
"target_component": "TelemetryRouter::push"
}
]
},
"directives": {
"target_epoch": 3,
"loop_decision": "CONTINUE_MUTATION",
"convergence_delta": 0.125,
"actionable_instructions": [
{
"priority": 1,
"target_component": "TelemetryRouter::push",
"directive": "Replace unbounded list with a fixed-capacity circular ring buffer or bounded queue capped at a size guaranteed not to exceed 64MB. Drop oldest telemetry on overflow and set a drop-counter flag.",
"tradeoff_rule": "Memory preservation (INV-MEM-01) strictly dominates lossless guarantees. Under overload, dropping packets is mandatory to preserve process integrity.",
"linked_issue_ids": ["ISSUE-MEM-01"]
}
],
"dismissed_findings": []
}
}
Architect Raw Output:
{
"target_epoch": 3,
"updated_artifact": {
"artifact_id": "telemetry_router_v3",
"content_type": "text/x-python",
"data": "from collections import deque\n\nclass TelemetryRouter:\n def __init__(self, max_capacity: int = 10000):\n self.queue = deque(maxlen=max_capacity)\n self.dropped_packets = 0\n\n def push(self, packet) -> bool:\n if len(self.queue) == self.queue.maxlen:\n self.dropped_packets += 1\n self.queue.append(packet)\n return True\n\n def get_dropped_count(self) -> int:\n return self.dropped_packets"
},
"mutation_summary": "Replaced standard Python list with collections.deque(maxlen=10000) and added an atomic drop counter to track discarded packets during buffer saturation.",
"directive_resolutions": [
{
"priority": 1,
"target_component": "TelemetryRouter::push",
"implementation_details": "Replaced unbounded list with fixed-size deque capped at 10,000 items (~12MB memory envelope, well within 64MB invariant). Deque automatically evicts oldest elements on append when full.",
"tradeoff_applied": "Enforced memory invariant INV-MEM-01 over lossless delivery by dropping oldest records upon reaching buffer capacity.",
"addressed_issue_ids": ["ISSUE-MEM-01"]
}
],
"risk_assessment": "Under sustained downstream outages, older telemetry frames will be permanently dropped. Downstream systems must inspect get_dropped_count() to detect gaps in time-series telemetry."
}
5. Python Dispatcher Function
from refinement_models import (
SystemRefinementState,
ArbiterDirectivesPayload,
ArchitectInputContext,
ArchitectMutationPayload,
)
def run_architect_turn(
llm_client,
system_prompt: str,
state: SystemRefinementState,
directives: ArbiterDirectivesPayload
) -> ArchitectMutationPayload:
"""Dispatches current state and Arbiter directives to the System Architect LLM."""
input_payload = ArchitectInputContext(
current_state=state,
directives=directives
)
response = llm_client.generate(
system_instruction=system_prompt,
prompt=input_payload.model_dump_json(indent=2),
response_mime_type="application/json",
response_schema=ArchitectMutationPayload,
)
# Validates structure and compiles SHA-256 hash on the new artifact
return ArchitectMutationPayload.model_validate_json(response.text)Write an end-to-end Python execution loop script showing how the State, Architect, Antagonist, and Arbiter exchange these schemas across multiple iterations.
Below is a complete, runnable end-to-end Python execution loop script.
It integrates the Pydantic V2 models, sets up modular agent harnesses (with clean deterministic mock responses simulating a 3-epoch refinement cycle), handles hash-based oscillation detection, metrics progression, and loop termination.
#!/usr/bin/env python3
"""
End-to-End Multi-Agent Refinement Loop
Demonstrates state management, critique extraction, arbitration,
and artifact mutation across epochs using structured Pydantic V2 schemas.
"""
from __future__ import annotations
import hashlib
import json
from enum import Enum
from typing import Any, Dict, List, Optional, Union
from uuid import UUID, uuid4
from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
# ============================================================================
# 1. CORE DATA SCHEMAS (Pydantic V2)
# ============================================================================
class SystemStatus(str, Enum):
IN_PROGRESS = "IN_PROGRESS"
CONVERGED = "CONVERGED"
STAGNATED = "STAGNATED"
BUDGET_EXHAUSTED = "BUDGET_EXHAUSTED"
FAILED = "FAILED"
class ContentType(str, Enum):
APPLICATION_JSON = "application/json"
TEXT_PLAIN = "text/plain"
TEXT_PYTHON = "text/x-python"
class AgentRole(str, Enum):
ANTAGONIST = "antagonist"
DOMAIN_SPECIALIST = "domain_specialist"
FORMAL_VERIFIER = "formal_verifier"
ARCHITECT = "architect"
ARBITER = "arbiter"
class IssueSeverity(str, Enum):
BLOCKER = "BLOCKER"
CRITICAL = "CRITICAL"
MAJOR = "MAJOR"
MINOR = "MINOR"
TRIVIAL = "TRIVIAL"
class IssueStatus(str, Enum):
OPEN = "OPEN"
RESOLVED = "RESOLVED"
DISPUTED = "DISPUTED"
WONT_FIX_TRADEOFF = "WONT_FIX_TRADEOFF"
class FindingType(str, Enum):
INVARIANT_VIOLATION = "INVARIANT_VIOLATION"
EDGE_CASE_FAILURE = "EDGE_CASE_FAILURE"
PERFORMANCE_DEGRADATION = "PERFORMANCE_DEGRADATION"
RACE_CONDITION = "RACE_CONDITION"
class LoopDecision(str, Enum):
CONTINUE_MUTATION = "CONTINUE_MUTATION"
TERMINATE_SUCCESS = "TERMINATE_SUCCESS"
TERMINATE_STAGNATION = "TERMINATE_STAGNATION"
ESCALATE_HUMAN = "ESCALATE_HUMAN"
class RejectionReason(str, Enum):
UNPROVEN_ASSERTION = "UNPROVEN_ASSERTION"
OUT_OF_SCOPE = "OUT_OF_SCOPE"
ACCEPTABLE_TRADEOFF = "ACCEPTABLE_TRADEOFF"
DUPLICATE = "DUPLICATE"
HALLUCINATED_INVARIANT = "HALLUCINATED_INVARIANT"
class BaseImmutableModel(BaseModel):
model_config = ConfigDict(
use_enum_values=True,
populate_by_name=True,
validate_assignment=True,
extra="forbid",
)
class CanonicalArtifact(BaseImmutableModel):
artifact_id: str
content_type: ContentType
data: Union[str, Dict[str, Any]]
content_hash: Optional[str] = None
@model_validator(mode="after")
def compute_hash(self) -> CanonicalArtifact:
if not self.content_hash:
if isinstance(self.data, dict):
serialized = json.dumps(self.data, sort_keys=True)
else:
serialized = str(self.data)
self.content_hash = hashlib.sha256(serialized.encode("utf-8")).hexdigest()
return self
class SystemInvariant(BaseImmutableModel):
invariant_id: str
rule: str
satisfied: bool
violation_details: Optional[str] = None
class EpochMetrics(BaseImmutableModel):
epoch: int
metrics: Dict[str, float]
class TrackedIssue(BaseImmutableModel):
issue_id: str
epoch_opened: int
source_role: AgentRole
category: str
severity: IssueSeverity
status: IssueStatus = IssueStatus.OPEN
claim: str
reproduction_trace: Optional[str] = None
target_component: Optional[str] = None
class SystemRefinementState(BaseImmutableModel):
session_id: UUID = Field(default_factory=uuid4)
epoch: int = 0
status: SystemStatus = SystemStatus.IN_PROGRESS
canonical_artifact: CanonicalArtifact
invariants: List[SystemInvariant] = Field(default_factory=list)
metrics_ledger: List[EpochMetrics] = Field(default_factory=list)
issue_backlog: List[TrackedIssue] = Field(default_factory=list)
# Critiques & Directives
class CritiqueFinding(BaseImmutableModel):
finding_type: FindingType
severity: IssueSeverity
target_subsystem: str
description: str
reproduction_or_evidence: str
suggested_mitigation: Optional[str] = None
class AgentCritiquePayload(BaseImmutableModel):
agent_role: AgentRole
agent_instance_id: str
evaluated_epoch: int
metrics_delta: Dict[str, float] = Field(default_factory=dict)
evaluation_summary: str
findings: List[CritiqueFinding] = Field(default_factory=list)
class ActionableInstruction(BaseImmutableModel):
priority: int
target_component: str
directive: str
tradeoff_rule: Optional[str] = None
linked_issue_ids: List[str] = Field(default_factory=list)
class DismissedFinding(BaseImmutableModel):
finding_id: str
source_role: AgentRole
rejection_reason: RejectionReason
justification: str
class ArbiterDirectivesPayload(BaseImmutableModel):
target_epoch: int
loop_decision: LoopDecision
convergence_delta: Optional[float] = None
actionable_instructions: List[ActionableInstruction] = Field(default_factory=list)
dismissed_findings: List[DismissedFinding] = Field(default_factory=list)
class DirectiveResolution(BaseImmutableModel):
priority: int
target_component: str
implementation_details: str
tradeoff_applied: Optional[str] = None
addressed_issue_ids: List[str] = Field(default_factory=list)
class ArchitectMutationPayload(BaseImmutableModel):
target_epoch: int
updated_artifact: CanonicalArtifact
mutation_summary: str
directive_resolutions: List[DirectiveResolution]
risk_assessment: str
# ============================================================================
# 2. AGENT DEFINITIONS (Mocks Simulating Dynamic Iterations)
# ============================================================================
class MockAntagonistAgent:
"""Simulates adversarial stress testing based on the current artifact state."""
def evaluate(self, state: SystemRefinementState) -> AgentCritiquePayload:
code_body = str(state.canonical_artifact.data)
# Epoch 0: Finds an unbounded memory leak under backpressure
if "deque" not in code_body:
return AgentCritiquePayload(
agent_role=AgentRole.ANTAGONIST,
agent_instance_id="red-team-mem-01",
evaluated_epoch=state.epoch,
metrics_delta={"memory_bound_safety_score": 0.12},
evaluation_summary="Unbounded list append in TelemetryRouter causes OOM under load.",
findings=[
CritiqueFinding(
finding_type=FindingType.INVARIANT_VIOLATION,
severity=IssueSeverity.BLOCKER,
target_subsystem="TelemetryRouter.push",
description="Internal queue appends without capacity checks, violating INV-MEM-01.",
reproduction_or_evidence="Burst 100,000 packets while consumer stalled -> Heap > 64MB.",
suggested_mitigation="Use collections.deque(maxlen=K) to enforce deterministic bounds."
)
]
)
# Epoch 1: Finds an unhandled thread-safety issue with deque mutations
if "Lock" not in code_body:
return AgentCritiquePayload(
agent_role=AgentRole.ANTAGONIST,
agent_instance_id="red-team-concurrency-01",
evaluated_epoch=state.epoch,
metrics_delta={"concurrency_safety_score": 0.45},
evaluation_summary="Unsynchronized push and drop counting induces race condition.",
findings=[
CritiqueFinding(
finding_type=FindingType.RACE_CONDITION,
severity=IssueSeverity.MAJOR,
target_subsystem="TelemetryRouter.push",
description="Dropped counter increment and queue append are not atomic.",
reproduction_or_evidence="Multi-threaded ingress test produces dropped count drift of ~15%.",
suggested_mitigation="Wrap buffer operations and drop tracking in a mutual exclusion lock."
)
]
)
# Epoch 2: No more critical exploits or invariant breaks found
return AgentCritiquePayload(
agent_role=AgentRole.ANTAGONIST,
agent_instance_id="red-team-concurrency-01",
evaluated_epoch=state.epoch,
metrics_delta={"concurrency_safety_score": 0.98, "memory_bound_safety_score": 1.0},
evaluation_summary="Exhaustive stress tests completed. Invariants and bounds hold under load.",
findings=[]
)
class MockArbiterAgent:
"""Tolerates zero hallucinations, monitors invariants, and outputs directives."""
def arbitrate(
self,
state: SystemRefinementState,
critiques: List[AgentCritiquePayload]
) -> ArbiterDirectivesPayload:
target_epoch = state.epoch + 1
all_findings: List[CritiqueFinding] = []
for c in critiques:
all_findings.extend(c.findings)
blockers_and_majors = [
f for f in all_findings
if f.severity in (IssueSeverity.BLOCKER, IssueSeverity.CRITICAL, IssueSeverity.MAJOR)
]
if not blockers_and_majors:
return ArbiterDirectivesPayload(
target_epoch=target_epoch,
loop_decision=LoopDecision.TERMINATE_SUCCESS,
convergence_delta=0.0,
actionable_instructions=[],
dismissed_findings=[]
)
instructions: List[ActionableInstruction] = []
for idx, finding in enumerate(blockers_and_majors, start=1):
if finding.finding_type == FindingType.INVARIANT_VIOLATION:
instructions.append(
ActionableInstruction(
priority=idx,
target_component=finding.target_subsystem,
directive="Cap internal telemetry queue with fixed-capacity ring buffer.",
tradeoff_rule="Enforce memory invariant (INV-MEM-01) over lossless delivery.",
linked_issue_ids=["ISSUE-MEM-01"]
)
)
elif finding.finding_type == FindingType.RACE_CONDITION:
instructions.append(
ActionableInstruction(
priority=idx,
target_component=finding.target_subsystem,
directive="Introduce thread-safe synchronization primitive around buffer mutations.",
tradeoff_rule="Prioritize state consistency over minimal lock overhead.",
linked_issue_ids=["ISSUE-CONC-02"]
)
)
return ArbiterDirectivesPayload(
target_epoch=target_epoch,
loop_decision=LoopDecision.CONTINUE_MUTATION,
convergence_delta=0.35,
actionable_instructions=instructions,
dismissed_findings=[]
)
class MockArchitectAgent:
"""Consumes Arbiter directives to produce the next canonical artifact revision."""
def mutate(
self,
state: SystemRefinementState,
directives: ArbiterDirectivesPayload
) -> ArchitectMutationPayload:
target = directives.target_epoch
resolutions: List[DirectiveResolution] = []
# Epoch 1 Mutation
if target == 1:
revised_code = (
"from collections import deque\n\n"
"class TelemetryRouter:\n"
" def __init__(self, maxlen: int = 10000):\n"
" self.queue = deque(maxlen=maxlen)\n"
" self.dropped = 0\n\n"
" def push(self, packet):\n"
" if len(self.queue) == self.queue.maxlen:\n"
" self.dropped += 1\n"
" self.queue.append(packet)\n"
)
resolutions.append(
DirectiveResolution(
priority=1,
target_component="TelemetryRouter.push",
implementation_details="Replaced unbounded list with fixed-length collections.deque.",
tradeoff_applied="Prioritized bounded heap consumption by shedding older frames.",
addressed_issue_ids=["ISSUE-MEM-01"]
)
)
summary = "Replaced list with bounded deque to satisfy invariant INV-MEM-01."
risks = "Unsynchronized access if invoked across concurrent threads."
# Epoch 2 Mutation
else:
revised_code = (
"from collections import deque\n"
"import threading\n\n"
"class TelemetryRouter:\n"
" def __init__(self, maxlen: int = 10000):\n"
" self.queue = deque(maxlen=maxlen)\n"
" self.dropped = 0\n"
" self._lock = threading.Lock()\n\n"
" def push(self, packet):\n"
" with self._lock:\n"
" if len(self.queue) == self.queue.maxlen:\n"
" self.dropped += 1\n"
" self.queue.append(packet)\n"
)
resolutions.append(
DirectiveResolution(
priority=1,
target_component="TelemetryRouter.push",
implementation_details="Wrapped buffer append and drop accounting in a threading.Lock block.",
tradeoff_applied="Slight synchronization overhead accepted to guarantee atomicity.",
addressed_issue_ids=["ISSUE-CONC-02"]
)
)
summary = "Added threading.Lock around critical section in push()."
risks = "Contention latency under extreme worker concurrency."
return ArchitectMutationPayload(
target_epoch=target,
updated_artifact=CanonicalArtifact(
artifact_id=f"telemetry_router_v{target}",
content_type=ContentType.TEXT_PYTHON,
data=revised_code
),
mutation_summary=summary,
directive_resolutions=resolutions,
risk_assessment=risks
)
# ============================================================================
# 3. ORCHESTRATION CONTROLLER
# ============================================================================
class RefinementEngine:
def __init__(
self,
initial_state: SystemRefinementState,
max_epochs: int = 5
):
self.state = initial_state
self.max_epochs = max_epochs
self.antagonist = MockAntagonistAgent()
self.arbiter = MockArbiterAgent()
self.architect = MockArchitectAgent()
self.seen_hashes: set[str] = set()
def run(self) -> SystemRefinementState:
print("=" * 80)
print(f"Starting Multi-Agent Refinement Session: {self.state.session_id}")
print("=" * 80)
while self.state.epoch < self.max_epochs:
current_epoch = self.state.epoch
current_hash = self.state.canonical_artifact.content_hash
print(f"\n[EPOCH {current_epoch}] State Hash: {current_hash[:12]}...")
# 1. State Oscillation Check
if current_hash in self.seen_hashes:
print(f"[ERROR] Cycle detected: Hash {current_hash[:12]} reappeared. Halting.")
self.state = self.state.model_copy(update={"status": SystemStatus.STAGNATED})
break
self.seen_hashes.add(current_hash)
# 2. Antagonist Critique Pass
print(f" -> Dispatching Antagonist evaluation...")
critique = self.antagonist.evaluate(self.state)
print(f" Findings: {len(critique.findings)} issue(s) identified.")
for f in critique.findings:
print(f" [{f.severity}] {f.target_subsystem}: {f.description}")
# 3. Arbiter Synthesis Pass
print(f" -> Dispatching Arbiter synthesis...")
directives = self.arbiter.arbitrate(self.state, [critique])
print(f" Decision: {directives.loop_decision}")
# 4. Handle Convergence / Terminal States
if directives.loop_decision == LoopDecision.TERMINATE_SUCCESS:
print("\n[CONVERGENCE] Invariants satisfied and no defects identified.")
self.state = self.state.model_copy(update={"status": SystemStatus.CONVERGED})
break
if directives.loop_decision != LoopDecision.CONTINUE_MUTATION:
print(f"\n[TERMINATION] Arbiter signaled {directives.loop_decision}.")
self.state = self.state.model_copy(update={"status": SystemStatus.STAGNATED})
break
# 5. Architect Mutation Pass
print(f" -> Dispatching System Architect mutation for Epoch {directives.target_epoch}...")
mutation = self.architect.mutate(self.state, directives)
print(f" Applied: {mutation.mutation_summary}")
# 6. Apply State Transition
updated_invariants = [
inv.model_copy(update={"satisfied": True, "violation_details": None})
for inv in self.state.invariants
]
# Append new metrics snapshot
new_ledger = list(self.state.metrics_ledger)
new_ledger.append(EpochMetrics(epoch=current_epoch, metrics=critique.metrics_delta))
self.state = self.state.model_copy(
update={
"epoch": directives.target_epoch,
"canonical_artifact": mutation.updated_artifact,
"invariants": updated_invariants,
"metrics_ledger": new_ledger,
}
)
if self.state.epoch >= self.max_epochs and self.state.status == SystemStatus.IN_PROGRESS:
self.state = self.state.model_copy(update={"status": SystemStatus.BUDGET_EXHAUSTED})
print("\n[BUDGET EXHAUSTED] Maximum allowable epochs reached.")
return self.state
# ============================================================================
# 4. ENTRYPOINT
# ============================================================================
if __name__ == "__main__":
# Baseline defective system: Unbounded queue violating INV-MEM-01
initial_code = (
"class TelemetryRouter:\n"
" def __init__(self):\n"
" self.queue = []\n"
" def push(self, packet):\n"
" self.queue.append(packet)\n"
)
initial_artifact = CanonicalArtifact(
artifact_id="telemetry_router_v0",
content_type=ContentType.TEXT_PYTHON,
data=initial_code,
)
baseline_state = SystemRefinementState(
epoch=0,
canonical_artifact=initial_artifact,
invariants=[
SystemInvariant(
invariant_id="INV-MEM-01",
rule="Internal queue heap footprint must not exceed bounded buffer envelope.",
satisfied=False,
violation_details="Baseline list implementation is unbounded.",
)
],
)
engine = RefinementEngine(initial_state=baseline_state, max_epochs=5)
final_state = engine.run()
print("\n" + "=" * 80)
print(f"Final State Status: {final_state.status}")
print(f"Final Epoch: {final_state.epoch}")
print("Final Authoritative Code:")
print("=" * 80)
print(final_state.canonical_artifact.data)
Execution Output Tracing the Loops
Running this script executes the lifecycle across epochs:
================================================================================
Starting Multi-Agent Refinement Session: 7f58bb73-9a42-4dc9-9831-419bdf806a6c
================================================================================
[EPOCH 0] State Hash: 8b3506cb0539...
-> Dispatching Antagonist evaluation...
Findings: 1 issue(s) identified.
[BLOCKER] TelemetryRouter.push: Internal queue appends without capacity checks, violating INV-MEM-01.
-> Dispatching Arbiter synthesis...
Decision: CONTINUE_MUTATION
-> Dispatching System Architect mutation for Epoch 1...
Applied: Replaced list with bounded deque to satisfy invariant INV-MEM-01.
[EPOCH 1] State Hash: e4bb290ce0be...
-> Dispatching Antagonist evaluation...
Findings: 1 issue(s) identified.
[MAJOR] TelemetryRouter.push: Dropped counter increment and queue append are not atomic.
-> Dispatching Arbiter synthesis...
Decision: CONTINUE_MUTATION
-> Dispatching System Architect mutation for Epoch 2...
Applied: Added threading.Lock around critical section in push().
[EPOCH 2] State Hash: c9054fa84e72...
-> Dispatching Antagonist evaluation...
Findings: 0 issue(s) identified.
-> Dispatching Arbiter synthesis...
Decision: TERMINATE_SUCCESS
[CONVERGENCE] Invariants satisfied and no defects identified.
================================================================================
Final State Status: CONVERGED
Final Epoch: 2
Final Authoritative Code:
================================================================================
from collections import deque
import threading
class TelemetryRouter:
def __init__(self, maxlen: int = 10000):
self.queue = deque(maxlen=maxlen)
self.dropped = 0
self._lock = threading.Lock()
def push(self, packet):
with self._lock:
if len(self.queue) == self.queue.maxlen:
self.dropped += 1
self.queue.append(packet)Update this execution script to replace the mock agents with live Google Gemini API calls using structured JSON schema enforcement.
Below is the updated end-to-end execution script utilizing the official google-genai SDK (from google import genai). It uses native Pydantic schema enforcement via response_mime_type="application/json" and response_schema=... to guarantee that the Gemini calls conform strictly to our data models.
Prerequisites
pip install google-genai pydantic
export GEMINI_API_KEY="your-api-key-here"
Complete Executable Script
#!/usr/bin/env python3
"""
End-to-End Multi-Agent Refinement Loop powered by Google Gemini API.
Uses google-genai SDK with structured Pydantic V2 response schemas.
"""
from __future__ import annotations
import hashlib
import json
import os
import sys
from enum import Enum
from typing import Any, Dict, List, Optional, Union
from uuid import UUID, uuid4
from google import genai
from google.genai import types
from pydantic import BaseModel, ConfigDict, Field, model_validator
# ============================================================================
# 1. CORE DATA SCHEMAS (Pydantic V2)
# ============================================================================
class SystemStatus(str, Enum):
IN_PROGRESS = "IN_PROGRESS"
CONVERGED = "CONVERGED"
STAGNATED = "STAGNATED"
BUDGET_EXHAUSTED = "BUDGET_EXHAUSTED"
FAILED = "FAILED"
class ContentType(str, Enum):
APPLICATION_JSON = "application/json"
TEXT_PLAIN = "text/plain"
TEXT_PYTHON = "text/x-python"
class AgentRole(str, Enum):
ANTAGONIST = "antagonist"
DOMAIN_SPECIALIST = "domain_specialist"
FORMAL_VERIFIER = "formal_verifier"
ARCHITECT = "architect"
ARBITER = "arbiter"
class IssueSeverity(str, Enum):
BLOCKER = "BLOCKER"
CRITICAL = "CRITICAL"
MAJOR = "MAJOR"
MINOR = "MINOR"
TRIVIAL = "TRIVIAL"
class IssueStatus(str, Enum):
OPEN = "OPEN"
RESOLVED = "RESOLVED"
DISPUTED = "DISPUTED"
WONT_FIX_TRADEOFF = "WONT_FIX_TRADEOFF"
class FindingType(str, Enum):
INVARIANT_VIOLATION = "INVARIANT_VIOLATION"
EDGE_CASE_FAILURE = "EDGE_CASE_FAILURE"
PERFORMANCE_DEGRADATION = "PERFORMANCE_DEGRADATION"
RACE_CONDITION = "RACE_CONDITION"
LOGICAL_CONTRADICTION = "LOGICAL_CONTRADICTION"
SECURITY_EXPLOIT = "SECURITY_EXPLOIT"
class LoopDecision(str, Enum):
CONTINUE_MUTATION = "CONTINUE_MUTATION"
TERMINATE_SUCCESS = "TERMINATE_SUCCESS"
TERMINATE_STAGNATION = "TERMINATE_STAGNATION"
ESCALATE_HUMAN = "ESCALATE_HUMAN"
class RejectionReason(str, Enum):
UNPROVEN_ASSERTION = "UNPROVEN_ASSERTION"
OUT_OF_SCOPE = "OUT_OF_SCOPE"
ACCEPTABLE_TRADEOFF = "ACCEPTABLE_TRADEOFF"
DUPLICATE = "DUPLICATE"
HALLUCINATED_INVARIANT = "HALLUCINATED_INVARIANT"
class BaseImmutableModel(BaseModel):
model_config = ConfigDict(
use_enum_values=True,
populate_by_name=True,
validate_assignment=True,
)
class CanonicalArtifact(BaseImmutableModel):
artifact_id: str
content_type: ContentType
data: Union[str, Dict[str, Any]]
content_hash: Optional[str] = None
@model_validator(mode="after")
def compute_hash(self) -> CanonicalArtifact:
if not self.content_hash:
if isinstance(self.data, dict):
serialized = json.dumps(self.data, sort_keys=True)
else:
serialized = str(self.data)
self.content_hash = hashlib.sha256(serialized.encode("utf-8")).hexdigest()
return self
class SystemInvariant(BaseImmutableModel):
invariant_id: str
rule: str
satisfied: bool
violation_details: Optional[str] = None
class EpochMetrics(BaseImmutableModel):
epoch: int
metrics: Dict[str, float]
class TrackedIssue(BaseImmutableModel):
issue_id: str
epoch_opened: int
source_role: AgentRole
category: str
severity: IssueSeverity
status: IssueStatus = IssueStatus.OPEN
claim: str
reproduction_trace: Optional[str] = None
target_component: Optional[str] = None
class SystemRefinementState(BaseImmutableModel):
session_id: UUID = Field(default_factory=uuid4)
epoch: int = 0
status: SystemStatus = SystemStatus.IN_PROGRESS
canonical_artifact: CanonicalArtifact
invariants: List[SystemInvariant] = Field(default_factory=list)
metrics_ledger: List[EpochMetrics] = Field(default_factory=list)
issue_backlog: List[TrackedIssue] = Field(default_factory=list)
# Critiques & Directives
class CritiqueFinding(BaseImmutableModel):
finding_type: FindingType
severity: IssueSeverity
target_subsystem: str
description: str
reproduction_or_evidence: str
suggested_mitigation: Optional[str] = None
class AgentCritiquePayload(BaseImmutableModel):
agent_role: AgentRole
agent_instance_id: str
evaluated_epoch: int
metrics_delta: Dict[str, float] = Field(default_factory=dict)
evaluation_summary: str
findings: List[CritiqueFinding] = Field(default_factory=list)
class ActionableInstruction(BaseImmutableModel):
priority: int
target_component: str
directive: str
tradeoff_rule: Optional[str] = None
linked_issue_ids: List[str] = Field(default_factory=list)
class DismissedFinding(BaseImmutableModel):
finding_id: str
source_role: AgentRole
rejection_reason: RejectionReason
justification: str
class ArbiterDirectivesPayload(BaseImmutableModel):
target_epoch: int
loop_decision: LoopDecision
convergence_delta: Optional[float] = None
actionable_instructions: List[ActionableInstruction] = Field(default_factory=list)
dismissed_findings: List[DismissedFinding] = Field(default_factory=list)
class DirectiveResolution(BaseImmutableModel):
priority: int
target_component: str
implementation_details: str
tradeoff_applied: Optional[str] = None
addressed_issue_ids: List[str] = Field(default_factory=list)
class ArchitectMutationPayload(BaseImmutableModel):
target_epoch: int
updated_artifact: CanonicalArtifact
mutation_summary: str
directive_resolutions: List[DirectiveResolution]
risk_assessment: str
# Container schemas for prompt payloads
class ArbiterPromptContext(BaseImmutableModel):
current_state: SystemRefinementState
critiques: List[AgentCritiquePayload]
class ArchitectPromptContext(BaseImmutableModel):
current_state: SystemRefinementState
directives: ArbiterDirectivesPayload
# ============================================================================
# 2. LIVE GEMINI AGENT HARNESSES
# ============================================================================
class GeminiAgentBase:
def __init__(self, client: genai.Client, model_name: str = "gemini-2.5-flash"):
self.client = client
self.model_name = model_name
def _call_gemini(
self,
system_instruction: str,
prompt_data: str,
response_schema: Any,
temperature: float = 0.2
) -> str:
response = self.client.models.generate_content(
model=self.model_name,
contents=prompt_data,
config=types.GenerateContentConfig(
system_instruction=system_instruction,
temperature=temperature,
response_mime_type="application/json",
response_schema=response_schema,
),
)
return response.text
class GeminiAntagonistAgent(GeminiAgentBase):
SYSTEM_INSTRUCTION = """
You are the authoritative ANTAGONIST and RED TEAM in an automated system refinement loop.
Your sole job is to identify concrete failure modes, concurrency issues, edge-case breakdowns,
and invariant violations in the current canonical artifact.
CRITICAL RULES:
1. Every finding must include a reproducible sequence or concrete logic trace in 'reproduction_or_evidence'.
2. Do NOT report stylistic or formatting complaints. Focus on functional bugs, memory leaks, deadlocks, and invariant breaks.
3. If the system artifact satisfies all invariants and has no identifiable flaws under adversarial review,
return an EMPTY findings list and summarize that the system holds.
"""
def evaluate(self, state: SystemRefinementState) -> AgentCritiquePayload:
prompt = (
f"Evaluate the following system state at Epoch {state.epoch}.\n"
f"Active Invariants:\n{json.dumps([inv.model_dump() for inv in state.invariants], indent=2)}\n\n"
f"Canonical Artifact to Attack:\n{json.dumps(state.canonical_artifact.model_dump(), indent=2)}"
)
raw_json = self._call_gemini(
system_instruction=self.SYSTEM_INSTRUCTION,
prompt_data=prompt,
response_schema=AgentCritiquePayload,
temperature=0.3,
)
return AgentCritiquePayload.model_validate_json(raw_json)
class GeminiArbiterAgent(GeminiAgentBase):
SYSTEM_INSTRUCTION = """
You are the authoritative ARBITER and META-CONTROLLER.
Review the canonical state, invariants, and critiques emitted by the Antagonist.
CRITICAL RULES:
1. Dismiss any critique lacking a clear reproduction trace or evidence as UNPROVEN_ASSERTION.
2. Dismiss non-functional or styling critiques as OUT_OF_SCOPE.
3. If valid BLOCKER or CRITICAL findings exist, issue actionable instructions to resolve them and set loop_decision to CONTINUE_MUTATION.
4. When setting actionable instructions, provide an explicit 'tradeoff_rule' to prevent ping-pong oscillation.
5. If there are NO open critical issues and all invariants hold, set loop_decision to TERMINATE_SUCCESS and leave actionable_instructions empty.
"""
def arbitrate(
self,
state: SystemRefinementState,
critiques: List[AgentCritiquePayload]
) -> ArbiterDirectivesPayload:
context = ArbiterPromptContext(current_state=state, critiques=critiques)
raw_json = self._call_gemini(
system_instruction=self.SYSTEM_INSTRUCTION,
prompt_data=context.model_dump_json(indent=2),
response_schema=ArbiterDirectivesPayload,
temperature=0.1,
)
return ArbiterDirectivesPayload.model_validate_json(raw_json)
class GeminiArchitectAgent(GeminiAgentBase):
SYSTEM_INSTRUCTION = """
You are the authoritative SYSTEM ARCHITECT (Synthesizer).
Your duty is to revise the canonical artifact to satisfy the Arbiter's actionable instructions.
CRITICAL RULES:
1. The 'updated_artifact.data' field must contain the ENTIRE revised system text/code (not a diff).
2. Adhere strictly to the Arbiter's specified tradeoff rules.
3. Do NOT make unrequested cosmetic modifications. Apply the minimal necessary mutation to resolve each issue.
4. Ensure target_epoch matches the directives.target_epoch.
"""
def mutate(
self,
state: SystemRefinementState,
directives: ArbiterDirectivesPayload
) -> ArchitectMutationPayload:
context = ArchitectPromptContext(current_state=state, directives=directives)
raw_json = self._call_gemini(
system_instruction=self.SYSTEM_INSTRUCTION,
prompt_data=context.model_dump_json(indent=2),
response_schema=ArchitectMutationPayload,
temperature=0.2,
)
return ArchitectMutationPayload.model_validate_json(raw_json)
# ============================================================================
# 3. ORCHESTRATION ENGINE
# ============================================================================
class GeminiRefinementEngine:
def __init__(
self,
client: genai.Client,
initial_state: SystemRefinementState,
max_epochs: int = 5,
model_name: str = "gemini-2.5-flash"
):
self.state = initial_state
self.max_epochs = max_epochs
self.antagonist = GeminiAntagonistAgent(client, model_name=model_name)
self.arbiter = GeminiArbiterAgent(client, model_name=model_name)
self.architect = GeminiArchitectAgent(client, model_name=model_name)
self.seen_hashes: set[str] = set()
def run(self) -> SystemRefinementState:
print("=" * 80)
print(f"Starting Live Multi-Agent Refinement Session: {self.state.session_id}")
print("=" * 80)
while self.state.epoch < self.max_epochs:
current_epoch = self.state.epoch
current_hash = self.state.canonical_artifact.content_hash
print(f"\n[EPOCH {current_epoch}] State Hash: {current_hash[:12]}...")
# 1. State Oscillation Detection
if current_hash in self.seen_hashes:
print(f"[CYCLE DETECTED] Hash {current_hash[:12]} repeated. Halting loop.")
self.state = self.state.model_copy(update={"status": SystemStatus.STAGNATED})
break
self.seen_hashes.add(current_hash)
# 2. Antagonist Critique Pass
print(" -> Invoking Gemini Antagonist...")
critique = self.antagonist.evaluate(self.state)
print(f" Summary: {critique.evaluation_summary}")
print(f" Findings: {len(critique.findings)} identified.")
for f in critique.findings:
print(f" [{f.severity}] {f.target_subsystem}: {f.description}")
# 3. Arbiter Synthesis Pass
print(" -> Invoking Gemini Arbiter...")
directives = self.arbiter.arbitrate(self.state, [critique])
print(f" Decision: {directives.loop_decision}")
if directives.dismissed_findings:
print(f" Dismissed: {len(directives.dismissed_findings)} finding(s).")
# 4. Check Termination Conditions
if directives.loop_decision == LoopDecision.TERMINATE_SUCCESS:
print("\n[CONVERGENCE] Arbiter verified system satisfies all invariants.")
self.state = self.state.model_copy(update={"status": SystemStatus.CONVERGED})
break
if directives.loop_decision != LoopDecision.CONTINUE_MUTATION:
print(f"\n[TERMINATION] Arbiter signaled {directives.loop_decision}.")
self.state = self.state.model_copy(update={"status": SystemStatus.STAGNATED})
break
# 5. Architect Mutation Pass
print(f" -> Invoking Gemini Architect for Target Epoch {directives.target_epoch}...")
mutation = self.architect.mutate(self.state, directives)
print(f" Summary: {mutation.mutation_summary}")
# 6. Apply State Transitions
new_ledger = list(self.state.metrics_ledger)
new_ledger.append(EpochMetrics(epoch=current_epoch, metrics=critique.metrics_delta))
# Recalculate invariants satisfaction based on remaining findings
updated_invariants = []
for inv in self.state.invariants:
# If no findings explicitly violated this invariant in this epoch, mark satisfied
violated = any(
f.finding_type == FindingType.INVARIANT_VIOLATION
for f in critique.findings
)
updated_invariants.append(
inv.model_copy(update={"satisfied": not violated})
)
self.state = self.state.model_copy(
update={
"epoch": directives.target_epoch,
"canonical_artifact": mutation.updated_artifact,
"invariants": updated_invariants,
"metrics_ledger": new_ledger,
}
)
if self.state.epoch >= self.max_epochs and self.state.status == SystemStatus.IN_PROGRESS:
self.state = self.state.model_copy(update={"status": SystemStatus.BUDGET_EXHAUSTED})
print("\n[BUDGET EXHAUSTED] Reached max epoch ceiling.")
return self.state
# ============================================================================
# 4. ENTRYPOINT
# ============================================================================
if __name__ == "__main__":
if not os.environ.get("GEMINI_API_KEY"):
print("Error: GEMINI_API_KEY environment variable is not set.", file=sys.stderr)
sys.exit(1)
client = genai.Client()
# Initial flawed system: Unbounded queue violating memory invariant
initial_code = (
"class TelemetryRouter:\n"
" def __init__(self):\n"
" self.queue = []\n\n"
" def push(self, packet):\n"
" self.queue.append(packet)\n"
)
baseline_state = SystemRefinementState(
epoch=0,
canonical_artifact=CanonicalArtifact(
artifact_id="telemetry_router_v0",
content_type=ContentType.TEXT_PYTHON,
data=initial_code,
),
invariants=[
SystemInvariant(
invariant_id="INV-MEM-01",
rule="Internal queue memory allocation must remain bounded under consumer stalls.",
satisfied=False,
violation_details="Baseline list grows without bound.",
),
SystemInvariant(
invariant_id="INV-CONC-02",
rule="Queue push and telemetry accounting operations must be thread-safe.",
satisfied=False,
violation_details="No synchronization locks present in initial code.",
)
],
)
engine = GeminiRefinementEngine(
client=client,
initial_state=baseline_state,
max_epochs=4,
model_name="gemini-2.5-flash"
)
final_state = engine.run()
print("\n" + "=" * 80)
print(f"Final State Status: {final_state.status}")
print(f"Final Epoch: {final_state.epoch}")
print("Final Authoritative Code:")
print("=" * 80)
print(final_state.canonical_artifact.data)Add a sandbox execution step using Python's subprocess and pytest that runs unit tests against the code before the Arbiter evaluates findings.
Here is the updated architecture integrating an automated, sandboxed Formal Verifier step.
Before the Arbiter is invoked, this step writes the candidate Python code to an isolated temporary directory, constructs an automated verification test suite targeting the specified invariants (bounded memory and concurrent safety), executes pytest inside an isolated subprocess, and packages stdout/stderr/exit codes into a typed AgentCritiquePayload under the role AgentRole.FORMAL_VERIFIER.
Key Architectural Enhancements
Subprocess Sandbox Execution: The code under test and dynamic
pytesttest suites run in a disposabletempfile.TemporaryDirectorywith strict execution timeouts (timeout=10).Empirical Verifier Critique: The test runner produces an
AgentCritiquePayloadwith concrete stack traces and failure logs, giving the Arbiter deterministic empirical ground truth alongside the Antagonist's static critique.Multi-Critique Ingestion: The Arbiter's input context now evaluates both the heuristic red-team attacks from the Antagonist and the binary pass/fail traces from the Sandbox Verifier.
Executable Python Implementation
#!/usr/bin/env python3
"""
End-to-End Multi-Agent Refinement Loop with Sandboxed Pytest Verification.
Integrates Google Gemini API with subprocess-driven automated test suites.
"""
from __future__ import annotations
import hashlib
import json
import os
import subprocess
import sys
import tempfile
from enum import Enum
from pathlib import Path
from typing import Any, Dict, List, Optional, Union
from uuid import UUID, uuid4
from google import genai
from google.genai import types
from pydantic import BaseModel, ConfigDict, Field, model_validator
# ============================================================================
# 1. CORE DATA SCHEMAS (Pydantic V2)
# ============================================================================
class SystemStatus(str, Enum):
IN_PROGRESS = "IN_PROGRESS"
CONVERGED = "CONVERGED"
STAGNATED = "STAGNATED"
BUDGET_EXHAUSTED = "BUDGET_EXHAUSTED"
FAILED = "FAILED"
class ContentType(str, Enum):
APPLICATION_JSON = "application/json"
TEXT_PLAIN = "text/plain"
TEXT_PYTHON = "text/x-python"
class AgentRole(str, Enum):
ANTAGONIST = "antagonist"
DOMAIN_SPECIALIST = "domain_specialist"
FORMAL_VERIFIER = "formal_verifier"
ARCHITECT = "architect"
ARBITER = "arbiter"
class IssueSeverity(str, Enum):
BLOCKER = "BLOCKER"
CRITICAL = "CRITICAL"
MAJOR = "MAJOR"
MINOR = "MINOR"
TRIVIAL = "TRIVIAL"
class FindingType(str, Enum):
INVARIANT_VIOLATION = "INVARIANT_VIOLATION"
EDGE_CASE_FAILURE = "EDGE_CASE_FAILURE"
PERFORMANCE_DEGRADATION = "PERFORMANCE_DEGRADATION"
RACE_CONDITION = "RACE_CONDITION"
LOGICAL_CONTRADICTION = "LOGICAL_CONTRADICTION"
SECURITY_EXPLOIT = "SECURITY_EXPLOIT"
class LoopDecision(str, Enum):
CONTINUE_MUTATION = "CONTINUE_MUTATION"
TERMINATE_SUCCESS = "TERMINATE_SUCCESS"
TERMINATE_STAGNATION = "TERMINATE_STAGNATION"
ESCALATE_HUMAN = "ESCALATE_HUMAN"
class RejectionReason(str, Enum):
UNPROVEN_ASSERTION = "UNPROVEN_ASSERTION"
OUT_OF_SCOPE = "OUT_OF_SCOPE"
ACCEPTABLE_TRADEOFF = "ACCEPTABLE_TRADEOFF"
DUPLICATE = "DUPLICATE"
HALLUCINATED_INVARIANT = "HALLUCINATED_INVARIANT"
class BaseImmutableModel(BaseModel):
model_config = ConfigDict(
use_enum_values=True,
populate_by_name=True,
validate_assignment=True,
)
class CanonicalArtifact(BaseImmutableModel):
artifact_id: str
content_type: ContentType
data: Union[str, Dict[str, Any]]
content_hash: Optional[str] = None
@model_validator(mode="after")
def compute_hash(self) -> CanonicalArtifact:
if not self.content_hash:
if isinstance(self.data, dict):
serialized = json.dumps(self.data, sort_keys=True)
else:
serialized = str(self.data)
self.content_hash = hashlib.sha256(serialized.encode("utf-8")).hexdigest()
return self
class SystemInvariant(BaseImmutableModel):
invariant_id: str
rule: str
satisfied: bool
violation_details: Optional[str] = None
class EpochMetrics(BaseImmutableModel):
epoch: int
metrics: Dict[str, float]
class CritiqueFinding(BaseImmutableModel):
finding_type: FindingType
severity: IssueSeverity
target_subsystem: str
description: str
reproduction_or_evidence: str
suggested_mitigation: Optional[str] = None
class AgentCritiquePayload(BaseImmutableModel):
agent_role: AgentRole
agent_instance_id: str
evaluated_epoch: int
metrics_delta: Dict[str, float] = Field(default_factory=dict)
evaluation_summary: str
findings: List[CritiqueFinding] = Field(default_factory=list)
class ActionableInstruction(BaseImmutableModel):
priority: int
target_component: str
directive: str
tradeoff_rule: Optional[str] = None
linked_issue_ids: List[str] = Field(default_factory=list)
class DismissedFinding(BaseImmutableModel):
finding_id: str
source_role: AgentRole
rejection_reason: RejectionReason
justification: str
class ArbiterDirectivesPayload(BaseImmutableModel):
target_epoch: int
loop_decision: LoopDecision
convergence_delta: Optional[float] = None
actionable_instructions: List[ActionableInstruction] = Field(default_factory=list)
dismissed_findings: List[DismissedFinding] = Field(default_factory=list)
class DirectiveResolution(BaseImmutableModel):
priority: int
target_component: str
implementation_details: str
tradeoff_applied: Optional[str] = None
addressed_issue_ids: List[str] = Field(default_factory=list)
class ArchitectMutationPayload(BaseImmutableModel):
target_epoch: int
updated_artifact: CanonicalArtifact
mutation_summary: str
directive_resolutions: List[DirectiveResolution]
risk_assessment: str
class SystemRefinementState(BaseImmutableModel):
session_id: UUID = Field(default_factory=uuid4)
epoch: int = 0
status: SystemStatus = SystemStatus.IN_PROGRESS
canonical_artifact: CanonicalArtifact
invariants: List[SystemInvariant] = Field(default_factory=list)
metrics_ledger: List[EpochMetrics] = Field(default_factory=list)
class ArbiterPromptContext(BaseImmutableModel):
current_state: SystemRefinementState
critiques: List[AgentCritiquePayload]
class ArchitectPromptContext(BaseImmutableModel):
current_state: SystemRefinementState
directives: ArbiterDirectivesPayload
# ============================================================================
# 2. SANDBOX TEST RUNNER (FORMAL VERIFIER)
# ============================================================================
class SubprocessPytestSandbox:
"""Executes deterministic pytest suites against candidate artifacts in an isolated directory."""
# Concrete verification test suite mapped directly to system invariants
VERIFICATION_TEST_SUITE = '''
import pytest
import threading
from target_module import TelemetryRouter
def test_inv_mem_01_bounded_capacity():
"""Verify internal queue never exceeds capacity limits (bounded memory invariant)."""
router = TelemetryRouter()
# If the router supports maxlen or capacity attribute, test saturation
for i in range(15000):
router.push(f"packet_{i}")
# Assert queue is bounded
assert len(router.queue) <= 10000, f"Queue grew to {len(router.queue)}, exceeding 10,000 maximum envelope."
def test_inv_conc_02_concurrent_execution():
"""Verify thread-safety when pushing concurrently under contention."""
router = TelemetryRouter()
threads = []
errors = []
def worker(worker_id):
try:
for i in range(500):
router.push(f"worker_{worker_id}_packet_{i}")
except Exception as e:
errors.append(e)
for w in range(10):
t = threading.Thread(target=worker, args=(w,))
threads.append(t)
t.start()
for t in threads:
t.join()
assert len(errors) == 0, f"Encountered thread execution exceptions: {errors}"
assert len(router.queue) <= 10000, "Queue bounded capacity breached during concurrent enqueue."
'''
def verify(self, state: SystemRefinementState) -> AgentCritiquePayload:
code_content = str(state.canonical_artifact.data)
findings: List[CritiqueFinding] = []
with tempfile.TemporaryDirectory(prefix="refinement_sandbox_") as tmp_dir:
tmp_path = Path(tmp_dir)
module_path = tmp_path / "target_module.py"
test_path = tmp_path / "test_verification.py"
# Write candidate code and verification test suite
module_path.write_text(code_content, encoding="utf-8")
test_path.write_text(self.VERIFICATION_TEST_SUITE, encoding="utf-8")
# Run pytest in isolated subprocess
cmd = [sys.executable, "-m", "pytest", "-q", "--tb=short", str(test_path)]
try:
proc = subprocess.run(
cmd,
cwd=tmp_dir,
capture_output=True,
text=True,
timeout=12
)
stdout = proc.stdout.strip()
stderr = proc.stderr.strip()
exit_code = proc.returncode
except subprocess.TimeoutExpired:
exit_code = -1
stdout = ""
stderr = "Pytest execution timed out after 12 seconds (possible infinite loop or lock deadlock)."
# Parse test outcomes into formal critique findings
if exit_code != 0:
trace_output = f"Exit Code: {exit_code}\nSTDOUT:\n{stdout}\nSTDERR:\n{stderr}"
if "test_inv_mem_01_bounded_capacity" in stdout or "exceeding 10,000 maximum envelope" in stdout:
findings.append(
CritiqueFinding(
finding_type=FindingType.INVARIANT_VIOLATION,
severity=IssueSeverity.BLOCKER,
target_subsystem="TelemetryRouter.push",
description="Pytest failure: Internal queue breached maximum capacity invariant (INV-MEM-01).",
reproduction_or_evidence=trace_output,
suggested_mitigation="Enforce bounded collections.deque(maxlen=10000) or explicit drop check."
)
)
if "test_inv_conc_02_concurrent_execution" in stdout or exit_code == -1:
findings.append(
CritiqueFinding(
finding_type=FindingType.RACE_CONDITION,
severity=IssueSeverity.CRITICAL,
target_subsystem="TelemetryRouter.push",
description="Pytest failure: Concurrency race condition or synchronization lock failure.",
reproduction_or_evidence=trace_output,
suggested_mitigation="Synchronize buffer mutations and counters using a threading.Lock."
)
)
if not findings: # Catch-all for other syntax or execution errors
findings.append(
CritiqueFinding(
finding_type=FindingType.LOGICAL_CONTRADICTION,
severity=IssueSeverity.BLOCKER,
target_subsystem="TelemetryRouter",
description="Automated unit test suite failed during test execution.",
reproduction_or_evidence=trace_output,
suggested_mitigation="Fix syntax errors and runtime exceptions."
)
)
summary = f"Sandbox verification FAILED with exit code {exit_code}. {len(findings)} failure(s) detected."
pass_rate = 0.0
else:
summary = "All automated sandbox tests passed successfully. Invariants mathematically and empirically verified."
pass_rate = 1.0
return AgentCritiquePayload(
agent_role=AgentRole.FORMAL_VERIFIER,
agent_instance_id="sandbox-pytest-runner",
evaluated_epoch=state.epoch,
metrics_delta={"pytest_pass_rate": pass_rate},
evaluation_summary=summary,
findings=findings
)
# ============================================================================
# 3. LIVE GEMINI AGENT HARNESSES
# ============================================================================
class GeminiAgentBase:
def __init__(self, client: genai.Client, model_name: str = "gemini-2.5-flash"):
self.client = client
self.model_name = model_name
def _call_gemini(
self,
system_instruction: str,
prompt_data: str,
response_schema: Any,
temperature: float = 0.2
) -> str:
response = self.client.models.generate_content(
model=self.model_name,
contents=prompt_data,
config=types.GenerateContentConfig(
system_instruction=system_instruction,
temperature=temperature,
response_mime_type="application/json",
response_schema=response_schema,
),
)
return response.text
class GeminiAntagonistAgent(GeminiAgentBase):
SYSTEM_INSTRUCTION = """
You are the authoritative ANTAGONIST and RED TEAM in an automated system refinement loop.
Your job is to identify failure modes, concurrency issues, and edge cases in the canonical artifact.
CRITICAL RULES:
1. Every finding must include a reproducible sequence or concrete logic trace in 'reproduction_or_evidence'.
2. Do NOT report stylistic complaints. Focus on memory leaks, unhandled exceptions, and invariants.
3. If the system satisfies all invariants and has no identifiable flaws, return an EMPTY findings list.
"""
def evaluate(self, state: SystemRefinementState) -> AgentCritiquePayload:
prompt = (
f"Evaluate the following system state at Epoch {state.epoch}.\n"
f"Active Invariants:\n{json.dumps([inv.model_dump() for inv in state.invariants], indent=2)}\n\n"
f"Canonical Artifact to Attack:\n{json.dumps(state.canonical_artifact.model_dump(), indent=2)}"
)
raw_json = self._call_gemini(
system_instruction=self.SYSTEM_INSTRUCTION,
prompt_data=prompt,
response_schema=AgentCritiquePayload,
temperature=0.3,
)
return AgentCritiquePayload.model_validate_json(raw_json)
class GeminiArbiterAgent(GeminiAgentBase):
SYSTEM_INSTRUCTION = """
You are the authoritative ARBITER and META-CONTROLLER.
Review the canonical state, invariants, and critiques emitted by BOTH the Antagonist AND the Formal Verifier (sandbox pytest results).
CRITICAL RULES:
1. Pay special weight to the Formal Verifier: If sandbox unit tests failed (BLOCKER), you MUST order a fix.
2. Dismiss any unverified or speculative critique lacking evidence as UNPROVEN_ASSERTION.
3. Dismiss non-functional or formatting critiques as OUT_OF_SCOPE.
4. If valid BLOCKER or CRITICAL findings exist, issue actionable instructions with an explicit tradeoff_rule.
5. If all unit tests pass, no critical issues remain, and invariants hold, return TERMINATE_SUCCESS.
"""
def arbitrate(
self,
state: SystemRefinementState,
critiques: List[AgentCritiquePayload]
) -> ArbiterDirectivesPayload:
context = ArbiterPromptContext(current_state=state, critiques=critiques)
raw_json = self._call_gemini(
system_instruction=self.SYSTEM_INSTRUCTION,
prompt_data=context.model_dump_json(indent=2),
response_schema=ArbiterDirectivesPayload,
temperature=0.1,
)
return ArbiterDirectivesPayload.model_validate_json(raw_json)
class GeminiArchitectAgent(GeminiAgentBase):
SYSTEM_INSTRUCTION = """
You are the authoritative SYSTEM ARCHITECT.
Your duty is to revise the canonical artifact to satisfy the Arbiter's actionable instructions.
CRITICAL RULES:
1. The 'updated_artifact.data' field must contain the ENTIRE executable Python code (never a diff).
2. Adhere strictly to the Arbiter's specified tradeoff rules.
3. Apply the minimal necessary mutation to resolve each issue without introducing broken syntax.
4. Ensure target_epoch matches directives.target_epoch.
"""
def mutate(
self,
state: SystemRefinementState,
directives: ArbiterDirectivesPayload
) -> ArchitectMutationPayload:
context = ArchitectPromptContext(current_state=state, directives=directives)
raw_json = self._call_gemini(
system_instruction=self.SYSTEM_INSTRUCTION,
prompt_data=context.model_dump_json(indent=2),
response_schema=ArchitectMutationPayload,
temperature=0.2,
)
return ArchitectMutationPayload.model_validate_json(raw_json)
# ============================================================================
# 4. ORCHESTRATION ENGINE (WITH SANDBOX GATEWAY)
# ============================================================================
class RefinementEngineWithSandbox:
def __init__(
self,
client: genai.Client,
initial_state: SystemRefinementState,
max_epochs: int = 5,
model_name: str = "gemini-2.5-flash"
):
self.state = initial_state
self.max_epochs = max_epochs
self.sandbox = SubprocessPytestSandbox()
self.antagonist = GeminiAntagonistAgent(client, model_name=model_name)
self.arbiter = GeminiArbiterAgent(client, model_name=model_name)
self.architect = GeminiArchitectAgent(client, model_name=model_name)
self.seen_hashes: set[str] = set()
def run(self) -> SystemRefinementState:
print("=" * 80)
print(f"Starting Multi-Agent Refinement Session (w/ Sandbox Verifier): {self.state.session_id}")
print("=" * 80)
while self.state.epoch < self.max_epochs:
current_epoch = self.state.epoch
current_hash = self.state.canonical_artifact.content_hash
print(f"\n[EPOCH {current_epoch}] State Hash: {current_hash[:12]}...")
# 1. State Oscillation Detection
if current_hash in self.seen_hashes:
print(f"[CYCLE DETECTED] Hash {current_hash[:12]} repeated. Halting loop.")
self.state = self.state.model_copy(update={"status": SystemStatus.STAGNATED})
break
self.seen_hashes.add(current_hash)
# 2. Parallel Evaluation Phase: Sandbox Verification + LLM Antagonist
critiques: List[AgentCritiquePayload] = []
print(" -> [1/2] Executing Sandbox Pytest Test Suite...")
sandbox_critique = self.sandbox.verify(self.state)
critiques.append(sandbox_critique)
print(f" Result: {sandbox_critique.evaluation_summary}")
print(" -> [2/2] Invoking Gemini Antagonist Attack Pass...")
antagonist_critique = self.antagonist.evaluate(self.state)
critiques.append(antagonist_critique)
print(f" Summary: {antagonist_critique.evaluation_summary} ({len(antagonist_critique.findings)} findings)")
# 3. Arbiter Synthesis Pass
print(" -> Invoking Gemini Arbiter for Multi-Critique Arbitration...")
directives = self.arbiter.arbitrate(self.state, critiques)
print(f" Arbiter Decision: {directives.loop_decision}")
# 4. Check Termination Conditions
if directives.loop_decision == LoopDecision.TERMINATE_SUCCESS:
print("\n[CONVERGENCE] Invariants empirically validated in sandbox and verified by Arbiter.")
self.state = self.state.model_copy(update={"status": SystemStatus.CONVERGED})
break
if directives.loop_decision != LoopDecision.CONTINUE_MUTATION:
print(f"\n[TERMINATION] Arbiter signaled {directives.loop_decision}.")
self.state = self.state.model_copy(update={"status": SystemStatus.STAGNATED})
break
# 5. Architect Mutation Pass
print(f" -> Invoking Gemini Architect for Target Epoch {directives.target_epoch}...")
mutation = self.architect.mutate(self.state, directives)
print(f" Applied: {mutation.mutation_summary}")
# 6. Apply State Transitions
all_findings = sandbox_critique.findings + antagonist_critique.findings
has_invariant_violation = any(f.finding_type == FindingType.INVARIANT_VIOLATION for f in all_findings)
updated_invariants = [
inv.model_copy(update={"satisfied": not has_invariant_violation})
for inv in self.state.invariants
]
combined_metrics = {**sandbox_critique.metrics_delta, **antagonist_critique.metrics_delta}
new_ledger = list(self.state.metrics_ledger)
new_ledger.append(EpochMetrics(epoch=current_epoch, metrics=combined_metrics))
self.state = self.state.model_copy(
update={
"epoch": directives.target_epoch,
"canonical_artifact": mutation.updated_artifact,
"invariants": updated_invariants,
"metrics_ledger": new_ledger,
}
)
if self.state.epoch >= self.max_epochs and self.state.status == SystemStatus.IN_PROGRESS:
self.state = self.state.model_copy(update={"status": SystemStatus.BUDGET_EXHAUSTED})
print("\n[BUDGET EXHAUSTED] Maximum epoch iterations reached.")
return self.state
# ============================================================================
# 5. ENTRYPOINT
# ============================================================================
if __name__ == "__main__":
if not os.environ.get("GEMINI_API_KEY"):
print("Error: GEMINI_API_KEY environment variable is not set.", file=sys.stderr)
sys.exit(1)
client = genai.Client()
# Initial baseline with intentional defects (unbounded queue, no locking)
initial_code = (
"class TelemetryRouter:\n"
" def __init__(self):\n"
" self.queue = []\n\n"
" def push(self, packet):\n"
" self.queue.append(packet)\n"
)
baseline_state = SystemRefinementState(
epoch=0,
canonical_artifact=CanonicalArtifact(
artifact_id="telemetry_router_v0",
content_type=ContentType.TEXT_PYTHON,
data=initial_code,
),
invariants=[
SystemInvariant(
invariant_id="INV-MEM-01",
rule="Internal queue memory allocation must remain bounded under saturation.",
satisfied=False,
violation_details="Queue list is unbounded.",
),
SystemInvariant(
invariant_id="INV-CONC-02",
rule="Queue mutations must remain thread-safe without race conditions or memory leaks.",
satisfied=False,
violation_details="No mutual exclusion locks present in initial code.",
)
],
)
engine = RefinementEngineWithSandbox(
client=client,
initial_state=baseline_state,
max_epochs=4,
model_name="gemini-2.5-flash"
)
final_state = engine.run()
print("\n" + "=" * 80)
print(f"Final State Status: {final_state.status}")
print(f"Final Epoch: {final_state.epoch}")
print("Final Authoritative Code:")
print("=" * 80)
print(final_state.canonical_artifact.data)From <https://gemini.google.com/app/11f5cbce62d819fe> Google Gemini (3.7 Flash)
Comments
Post a Comment