+------------------------------------------------------------------+
|                     CONFLUENCE LABS ARC-AGI-2                     |
|                        SYSTEM ARCHITECTURE                       |
+------------------------------------------------------------------+
|                                                                  |
|  +--------------------+    +-------------------------------+     |
|  |    run.sh           |    |     Configuration (.env)      |     |
|  |  Entry Point        |--->|  GEMINI_CLI_AGENTS=12         |     |
|  |  --smoke / --full   |    |  GEMINI_CLI_MAX_ITERATIONS=10 |     |
|  +--------------------+    |  GEMINI_CLI_CONCURRENCY=132    |     |
|           |                |  WALL_CLOCK_LIMIT=43200        |     |
|           v                +-------------------------------+     |
|  +---------------------------------------------------+          |
|  |            gemini-cli-solver/ (Core Engine)        |          |
|  |                                                    |          |
|  |  +----------+  +----------+      +----------+     |          |
|  |  | Agent  1 |  | Agent  2 | ...  | Agent 12 |     |          |
|  |  | (iter x10)| | (iter x10)|     | (iter x10)|    |          |
|  |  +----+-----+  +----+-----+      +----+-----+    |          |
|  |       |              |                 |           |          |
|  |       v              v                 v           |          |
|  |  +-------------------------------------------+    |          |
|  |  |        E2B Sandbox Pool (132 slots)        |   |          |
|  |  |  +------+ +------+ +------+     +------+  |   |          |
|  |  |  | Py 1 | | Py 2 | | Py 3 | ... |Py 132| |   |          |
|  |  |  +------+ +------+ +------+     +------+  |   |          |
|  |  +-------------------------------------------+    |          |
|  |       |                                            |          |
|  |       v                                            |          |
|  |  +-------------------------------------------+    |          |
|  |  |        Gemini API (LLM Backend)            |   |          |
|  |  |  Task Interpretation + Code Generation     |   |          |
|  |  +-------------------------------------------+    |          |
|  +---------------------------------------------------+          |
|           |                                                      |
|           v                                                      |
|  +---------------------------------------------------+          |
|  |        Ensemble Aggregation & Output Selection     |          |
|  |  Vote / Consensus among 12 agent solutions         |          |
|  +---------------------------------------------------+          |
|           |                                                      |
|           v                                                      |
|  +---------------------------------------------------+          |
|  |        Final Output (2 guesses per test input)     |          |
|  +---------------------------------------------------+          |
+------------------------------------------------------------------+

arc-agi-2/
  gemini-cli-solver/         # Core solver engine
    src/                     # Source code
      solver.py              # Main solver logic
      agent.py               # Individual agent implementation
      sandbox.py             # E2B sandbox management
      ensemble.py            # Ensemble aggregation
      prompts/               # Prompt templates
        task_prompt.py        # Task formatting
        refinement_prompt.py  # Iterative refinement prompts
    config/                  # Configuration files
    tests/                   # Unit and integration tests
  .env                       # Environment configuration
  run.sh                     # Entry point script
  pyproject.toml             # Python project configuration
  uv.lock                    # Dependency lock file
  README.md                  # Documentation

import asyncio
from dataclasses import dataclass, field
from typing import List, Optional, Dict, Any
import json

@dataclass
class ArcTask:
    """Represents a single ARC-AGI-2 task."""
    task_id: str
    train_pairs: List[Dict[str, List[List[int]]]]
    test_inputs: List[List[List[int]]]

    @classmethod
    def from_json(cls, task_id: str, data: Dict[str, Any]) -> "ArcTask":
        return cls(
            task_id=task_id,
            train_pairs=data["train"],
            test_inputs=[t["input"] for t in data["test"]]
        )

    def format_for_prompt(self) -> str:
        """Format task as a structured prompt for the LLM."""
        lines = ["# ARC Task: Transform input grids to output grids\n"]
        lines.append("## Training Examples\n")
        for i, pair in enumerate(self.train_pairs):
            lines.append(f"### Example {i + 1}")
            lines.append(f"Input:\n{self._format_grid(pair['input'])}")
            lines.append(f"Output:\n{self._format_grid(pair['output'])}")
            lines.append("")
        lines.append("## Task")
        lines.append("Write a Python function `transform(input_grid)` that takes")
        lines.append("a 2D list of integers and returns the transformed 2D list.")
        lines.append("The function should correctly transform ALL training inputs")
        lines.append("to their corresponding outputs.\n")
        return "\n".join(lines)

    @staticmethod
    def _format_grid(grid: List[List[int]]) -> str:
        return "\n".join(
            "[" + ", ".join(str(c) for c in row) + "]"
            for row in grid
        )


@dataclass
class SolverConfig:
    """Configuration for the solver engine."""
    num_agents: int = 12
    max_iterations: int = 10
    concurrency: int = 132
    wall_clock_limit: int = 43200  # 12 hours in seconds
    gemini_model: str = "gemini-2.5-pro"
    sandbox_timeout: int = 30  # seconds per execution


class Solver:
    """Main solver orchestrator."""

    def __init__(self, config: SolverConfig):
        self.config = config
        self.agents = [
            Agent(agent_id=i, config=config)
            for i in range(config.num_agents)
        ]

    async def solve(self, task: ArcTask) -> List[List[List[int]]]:
        """Solve an ARC task using the multi-agent ensemble."""
        # Launch all agents in parallel
        agent_tasks = [
            agent.solve_task(task)
            for agent in self.agents
        ]
        results = await asyncio.gather(*agent_tasks, return_exceptions=True)

        # Filter successful results
        valid_results = [
            r for r in results
            if isinstance(r, AgentResult) and r.success
        ]

        # Ensemble aggregation
        return self._aggregate_solutions(valid_results, task)

    def _aggregate_solutions(
        self,
        results: List["AgentResult"],
        task: ArcTask
    ) -> List[List[List[int]]]:
        """Select final outputs via majority voting."""
        from collections import Counter

        per_test_outputs = []
        for test_idx in range(len(task.test_inputs)):
            # Collect all candidate outputs for this test input
            candidates = []
            for result in results:
                if test_idx < len(result.test_outputs):
                    output = result.test_outputs[test_idx]
                    candidates.append(self._grid_to_hashable(output))

            # Majority vote
            if candidates:
                counter = Counter(candidates)
                top_two = counter.most_common(2)
                per_test_outputs.append([
                    self._hashable_to_grid(top_two[0][0]),
                    self._hashable_to_grid(top_two[1][0])
                    if len(top_two) > 1
                    else self._hashable_to_grid(top_two[0][0])
                ])

        return per_test_outputs

    @staticmethod
    def _grid_to_hashable(grid):
        return tuple(tuple(row) for row in grid)

    @staticmethod
    def _hashable_to_grid(hashable):
        return [list(row) for row in hashable]

@dataclass
class AgentResult:
    """Result from a single agent's attempt."""
    agent_id: int
    success: bool
    test_outputs: List[List[List[int]]]
    iterations_used: int
    final_code: str
    error_log: List[str] = field(default_factory=list)


class Agent:
    """Individual solver agent with iterative refinement."""

    def __init__(self, agent_id: int, config: SolverConfig):
        self.agent_id = agent_id
        self.config = config
        self.llm_client = GeminiClient(model=config.gemini_model)
        self.sandbox = E2BSandbox(timeout=config.sandbox_timeout)

    async def solve_task(self, task: ArcTask) -> AgentResult:
        """Attempt to solve a task with iterative refinement."""
        prompt = task.format_for_prompt()
        code = None
        error_log = []

        for iteration in range(self.config.max_iterations):
            # Generate or refine code
            if iteration == 0:
                code = await self.llm_client.generate_code(prompt)
            else:
                refinement_prompt = self._build_refinement_prompt(
                    task, code, error_log[-1]
                )
                code = await self.llm_client.generate_code(refinement_prompt)

            # Test against training pairs
            test_result = await self.sandbox.execute_and_test(
                code, task.train_pairs
            )

            if test_result.all_passed:
                # Success: run on test inputs
                test_outputs = await self.sandbox.execute_on_inputs(
                    code, task.test_inputs
                )
                return AgentResult(
                    agent_id=self.agent_id,
                    success=True,
                    test_outputs=test_outputs,
                    iterations_used=iteration + 1,
                    final_code=code
                )
            else:
                error_log.append(test_result.error_summary)

        # Exhausted iterations -- return best effort
        return AgentResult(
            agent_id=self.agent_id,
            success=False,
            test_outputs=[],
            iterations_used=self.config.max_iterations,
            final_code=code or "",
            error_log=error_log
        )

    def _build_refinement_prompt(
        self,
        task: ArcTask,
        previous_code: str,
        error_summary: str
    ) -> str:
        """Build a refinement prompt incorporating error feedback."""
        return f"""
The following code was generated to solve an ARC task but produced
incorrect results. Please analyze the errors and generate an improved
version.

## Previous Code
```python
{previous_code}

## 7 Multi-Agent Ensemble

### 7.1 Ensemble Architecture

The Confluence Labs system deploys 12 independent agents per test input, configured via
      `GEMINI_CLI_AGENTS=12`. This ensemble architecture is motivated by several theoretical
      and practical considerations:

- **Diversity of hypotheses:** LLMs exhibit stochasticity in their outputs (controlled
          by temperature and sampling parameters). Running multiple agents increases the probability that at
          least one agent discovers the correct transformation rule, even if individual agents have a
          relatively low per-attempt success rate.
- **Robustness to initialization:** Different agents may explore different parts of
          the program space, leading to complementary solution strategies that cover a wider range of
          possible transformation rules.
- **Statistical reliability:** With 12 agents, the ensemble can use majority voting
          or consensus-based selection to filter out spurious solutions that happen to pass training
          examples but do not generalize.

### 7.2 Ensemble Aggregation Strategies

Given 12 agents each producing candidate solutions, the system must select the final two guesses
      to submit (ARC-AGI-2 allows 2 guesses per test input). Confluence Labs employs a consensus-based
      approach:

| Strategy | Description | Strengths | Weaknesses |
| --- | --- | --- | --- |
| Majority Vote | Select output with most agent agreement | Simple, robust to outliers | May fail if correct answer is rare |
| Weighted Consensus | Weight by agent confidence / iteration count | Incorporates quality signals | Confidence may not correlate with correctness |
| Diversity Selection | Select most common + most different output | Maximizes coverage with 2 guesses | Second guess may be noise |
| Verification-Based | Verify solutions against additional criteria | Highest precision when criteria are available | Additional criteria hard to define for novel tasks |

### 7.3 Mathematical Analysis of Ensemble Size

Let *p* denote the probability that a single agent solves a given task. With *n* = 12
      independent agents, the probability that at least one agent succeeds is:


$$
P(at least one success) = 1 - (1 - p)^n = 1 - (1 - p)^12
$$

For example, if a single agent has only a 30% chance of solving a task (p = 0.3), the ensemble
      probability rises to:


$$
P = 1 - (0.7)^12 = 1 - 0.0138 = 0.986 (98.6%)
$$

This dramatic amplification of success probability is the fundamental driver behind the 12-agent
      architecture. The marginal benefit of additional agents follows a diminishing returns curve,
      and 12 agents represents a carefully chosen balance between coverage and cost.

> [!info]- Ensemble Size vs. Success Probability (for p = 0.30)
> | Agents (n) | P(at least one success) | Marginal Gain |
> | --- | --- | --- |
> | 1 | 30.0% | -- |
> | 2 | 51.0% | +21.0% |
> | 4 | 76.0% | +12.5%/agent |
> | 6 | 88.2% | +6.1%/agent |
> | 8 | 94.2% | +3.0%/agent |
> | 10 | 97.2% | +1.5%/agent |
> | 12 | 98.6% | +0.7%/agent |
> | 16 | 99.5% | +0.2%/agent |
> | 20 | 99.8% | +0.1%/agent |

### 7.4 Agent Independence and Correlation

The analysis above assumes agent independence, which is not strictly true. All agents use the same
      LLM (Gemini) and receive the same task description, introducing positive correlation between agent
      outcomes. This correlation reduces the effective diversity of the ensemble and means the actual
      success probability is somewhat lower than the theoretical maximum. However, LLM output
      stochasticity (temperature > 0) and the iterative refinement process (where agents diverge
      based on different error paths) introduce meaningful variance.

> **Practical Consideration:** To maximize diversity, agents could be initialized with
>       different system prompts, reasoning strategies, or temperature settings. While the public
>       repository uses a uniform agent configuration, this represents an obvious avenue for further
>       optimization.

## 8 Infrastructure & Sandbox Environments

### 8.1 E2B Sandbox Architecture

Confluence Labs uses [E2B](https://e2b.dev) (short for "Environment to Binary")
      as its sandbox infrastructure. E2B provides ephemeral, isolated cloud environments for executing
      untrusted code -- a critical requirement when running LLM-generated programs that may contain
      bugs, infinite loops, or unexpected behavior.

The system maintains a pool of 132 concurrent sandboxes (`GEMINI_CLI_CONCURRENCY=132`),
      allowing massive parallelization of code execution. This concurrency level is designed to support
      the throughput requirements of 12 agents, each potentially running multiple iterations simultaneously
      across the full task set.

Python -- E2B Sandbox Integration (Conceptual)

def __init__(self, timeout: int = 30):
    self.timeout = timeout
    self.api_key = os.environ.get("E2B_API_KEY")

async def execute_and_test(
    self,
    code: str,
    train_pairs: List[Dict[str, Any]]
) -> "ExecutionResult":
    """Execute code and test against training pairs."""
    # Build test harness
    test_code = self._build_test_harness(code, train_pairs)

    # Execute in isolated sandbox
    try:
        result = await asyncio.wait_for(
            self._run_in_sandbox(test_code),
            timeout=self.timeout
        )
        return self._parse_execution_result(result)
    except asyncio.TimeoutError:
        return ExecutionResult(
            all_passed=False,
            error_summary="Execution timed out after "
                          f"{self.timeout} seconds"
        )

def _build_test_harness(
    self,
    code: str,
    train_pairs: List[Dict[str, Any]]
) -> str:
    """Wrap user code with test assertions."""
    harness = f"""

async def _run_in_sandbox(self, code: str) -> Dict[str, Any]:
    """Execute code in an E2B sandbox instance."""
    # E2B API call (simplified)
    sandbox = await e2b.Sandbox.create(api_key=self.api_key)
    try:
        execution = await sandbox.run_code(code, language="python")
        return {
            "stdout": execution.stdout,
            "stderr": execution.stderr,
            "exit_code": execution.exit_code
        }
    finally:
        await sandbox.close()

async def execute_on_inputs(
    self,
    code: str,
    test_inputs: List[List[List[int]]]
) -> List[List[List[int]]]:
    """Execute validated code on test inputs."""
    results = []
    for test_input in test_inputs:
        exec_code = f"""

### 8.2 Sandbox Pool Management

With 132 concurrent sandboxes, pool management becomes a non-trivial engineering challenge.
      The system implements:

- **Semaphore-based concurrency control:** An asyncio semaphore limits the number
          of simultaneous sandbox executions to 132, preventing resource exhaustion
- **Timeout enforcement:** Each sandbox execution is wrapped in a timeout handler
          that forcefully terminates runaway processes
- **Cleanup guarantees:** Sandbox instances are closed in `finally` blocks
          to prevent resource leaks even when exceptions occur
- **Retry logic:** Transient E2B API failures trigger automatic retries with
          exponential backoff

### 8.3 Concurrency Architecture

## 9 Configuration & Environment

### 9.1 Environment Variables

The system is configured entirely through environment variables, loaded from a `.env`
      file at startup. This approach provides flexibility for different execution environments
      (development, testing, production) while keeping sensitive API keys out of the codebase.

.env -- Configuration File

### 9.2 Configuration Design Rationale

| Parameter | Value | Rationale |
| --- | --- | --- |
| `GEMINI_CLI_AGENTS` | 12 | Sweet spot on ensemble success probability curve (see Section 7.3) |
| `GEMINI_CLI_MAX_ITERATIONS` | 10 | Sufficient for most tasks; diminishing returns beyond 8 iterations empirically |
| `GEMINI_CLI_CONCURRENCY` | 132 | 12 agents x 10 iterations + 10% overhead buffer for retries |
| `WALL_CLOCK_LIMIT` | 43200s (12h) | Competition constraint; allows thorough exploration of task set |
| `SANDBOX_TIMEOUT` | 30s | ARC transformations should execute in milliseconds; 30s catches infinite loops |

### 9.3 Package Management with uv

The project uses [uv](https://github.com/astral-sh/uv) as its Python
      package manager, reflecting a modern approach to Python dependency management. uv offers significant
      advantages over traditional tools like pip:

- **Speed:** 10-100x faster than pip for dependency resolution and installation
- **Reproducibility:** Lock file (`uv.lock`) ensures deterministic builds
- **Virtual environment management:** Automatic venv creation and activation
- **Compatibility:** Full compatibility with pyproject.toml and existing Python tooling

## 10 Execution Pipeline

### 10.1 Entry Point: run.sh

The system provides two execution modes, controlled by the `run.sh` entry point:

Shell -- run.sh Execution Modes

### 10.2 Full Run Pipeline

A full run (`./run.sh`) executes the complete evaluation pipeline:

1. **Initialization:** Load configuration, validate API keys, initialize sandbox pool
2. **Task Loading:** Load all ARC-AGI-2 evaluation tasks from JSON files
3. **Parallel Dispatch:** Distribute tasks across the agent pool, respecting the
          concurrency limit
4. **Per-Task Solving:** For each task, 12 agents independently attempt to solve it
          with up to 10 iterations each
5. **Ensemble Aggregation:** Collect solutions from all agents and select 2 guesses
          per test input via majority voting
6. **Output Generation:** Write final predictions in the required submission format
7. **Cleanup:** Close all sandbox instances, log summary statistics

### 10.3 Smoke Test Pipeline

The smoke test mode (`./run.sh --smoke <task_id>`) provides a lightweight execution
      path for development and debugging:

- Single task execution (specified by ID)
- Single agent (instead of 12)
- Reduced iterations (3 instead of 10)
- Full logging and debug output enabled
- Typically completes in under 5 minutes

### 10.4 Wall Clock Management

The 12-hour wall clock limit (`WALL_CLOCK_LIMIT=43200`) is a hard constraint imposed by
      the ARC-AGI-2 competition. The system implements sophisticated time management to maximize the
      number of tasks solved within this window:

Python -- Wall Clock Manager

def __init__(self, limit_seconds: int = 43200):
    self.limit = limit_seconds
    self.start_time = time.monotonic()
    self.tasks_completed = 0
    self.tasks_remaining = 0

@property
def elapsed(self) -> float:
    return time.monotonic() - self.start_time

@property
def remaining(self) -> float:
    return max(0, self.limit - self.elapsed)

@property
def is_expired(self) -> bool:
    return self.remaining <= 0

def budget_per_task(self) -> float:
    """Dynamically allocate time budget per remaining task."""
    if self.tasks_remaining <= 0:
        return self.remaining
    # Reserve 10% buffer for final aggregation
    available = self.remaining * 0.9
    return available / self.tasks_remaining

def should_continue(self, min_time_per_task: float = 60.0) -> bool:
    """Check if there is enough time for another task."""
    return self.remaining > min_time_per_task

## 11 Program Synthesis via LLMs

### 11.1 The Program Synthesis Paradigm in Context

Program synthesis via LLMs represents a fundamental paradigm shift from traditional approaches to
      ARC-like tasks. Rather than training a neural network to directly predict output grids (as in
      pixel-level prediction approaches), or searching over a hand-crafted domain-specific language
      (as in traditional program synthesis), Confluence Labs uses the LLM as a code generator that
      produces general Python programs.

This approach offers several key advantages:

- **Expressiveness:** Python is a Turing-complete language, meaning any computable
          transformation can in principle be expressed. Unlike restricted DSLs, there are no artificial
          limits on the complexity of expressible transformations.
- **Interpretability:** Generated programs are human-readable, providing transparency
          into the system's reasoning process. Each solution explicitly encodes the discovered transformation
          rule in a form that can be inspected, debugged, and verified.
- **Verifiability:** Programs can be executed and their outputs compared against expected
          results, providing unambiguous correctness feedback.
- **Transfer:** Correct programs generalize perfectly to new inputs (within the scope
          of the inferred rule), as they encode the abstract transformation rather than memorizing specific
          examples.

### 11.2 Prompt Engineering for Code Generation

The quality of the prompt is critical to the success of the program synthesis approach. Confluence Labs
      employs carefully crafted prompt templates that leverage the first core principle (structural alignment
      with training distributions).

Python -- Prompt Template

def transform(input_grid: list[list[int]]) -> list[list[int]]:
    # Your implementation here
    pass

### 11.3 Code Extraction and Validation

LLM responses often contain explanatory text alongside the code. The system implements robust
      code extraction to isolate the Python function from the surrounding narrative:

Python -- Code Extraction

PYTHON_BLOCK_PATTERN = re.compile(
    r'```python\s*\n(.*?)```',
    re.DOTALL
)

@classmethod
def extract(cls, response: str) -> str:
    """Extract Python code from an LLM response."""
    matches = cls.PYTHON_BLOCK_PATTERN.findall(response)
    if not matches:
        # Fallback: try to find a function definition directly
        lines = response.split('\n')
        code_lines = []
        in_function = False
        for line in lines:
            if line.strip().startswith('def transform'):
                in_function = True
            if in_function:
                code_lines.append(line)
        if code_lines:
            return '\n'.join(code_lines)
        raise ValueError("No Python code found in LLM response")

    # If multiple code blocks, find the one with transform()
    for match in matches:
        if 'def transform' in match:
            return match.strip()

    # Fallback: return the longest code block
    return max(matches, key=len).strip()

@classmethod
def validate(cls, code: str) -> bool:
    """Validate that the code is syntactically correct Python."""
    try:
        ast.parse(code)
        return True
    except SyntaxError:
        return False

@classmethod
def has_transform_function(cls, code: str) -> bool:
    """Check that the code defines a transform() function."""
    tree = ast.parse(code)
    for node in ast.walk(tree):
        if isinstance(node, ast.FunctionDef):
            if node.name == 'transform':
                return True
    return False

### 11.4 Common Transformation Patterns

Through extensive experimentation with ARC tasks, Confluence Labs has identified recurring
      transformation patterns that their system handles effectively:

> [!info]- Spatial Transformations
> - **Rotation:** 90, 180, 270-degree rotations of entire grids or sub-regions
> - **Reflection:** Horizontal, vertical, or diagonal mirroring
> - **Translation:** Moving objects by fixed or variable offsets
> - **Scaling:** Enlarging or shrinking patterns by integer factors
> - **Tiling:** Repeating patterns to fill a larger grid
>
> ```
> # Example: 90-degree clockwise rotation
> def transform(input_grid):
>     rows = len(input_grid)
>     cols = len(input_grid[0])
>     output = [[0] * rows for _ in range(cols)]
>     for r in range(rows):
>         for c in range(cols):
>             output[c][rows - 1 - r] = input_grid[r][c]
>     return output
> ```

> [!info]- Color and Value Transformations
> - **Color mapping:** Systematic replacement of one color with another
> - **Conditional coloring:** Color assignment based on spatial relationships
> - **Flood fill:** Filling connected regions with a specified color
> - **Border detection:** Identifying and coloring object boundaries
>
> ```
> # Example: Color swap based on adjacency to background
> def transform(input_grid):
>     import copy
>     rows = len(input_grid)
>     cols = len(input_grid[0])
>     output = copy.deepcopy(input_grid)
>
>     for r in range(rows):
>         for c in range(cols):
>             if input_grid[r][c] != 0:
>                 # Check if adjacent to background
>                 adjacent_to_bg = False
>                 for dr, dc in [(-1,0),(1,0),(0,-1),(0,1)]:
>                     nr, nc = r + dr, c + dc
>                     if 0 <= nr < rows and 0 <= nc < cols:
>                         if input_grid[nr][nc] == 0:
>                             adjacent_to_bg = True
>                 if adjacent_to_bg:
>                     output[r][c] = 3  # Change border cells
>     return output
> ```

> [!info]- Object-Level Transformations
> - **Object detection:** Identifying connected components as discrete objects
> - **Sorting:** Arranging objects by size, color, or position
> - **Grouping:** Clustering objects by shared properties
> - **Compositing:** Combining or overlaying multiple objects
>
> ```
> # Example: Sort objects by size (descending)
> def transform(input_grid):
>     from collections import deque
>
>     rows = len(input_grid)
>     cols = len(input_grid[0])
>     visited = [[False] * cols for _ in range(rows)]
>     objects = []
>
>     # Find connected components (objects)
>     for r in range(rows):
>         for c in range(cols):
>             if input_grid[r][c] != 0 and not visited[r][c]:
>                 # BFS to find full object
>                 obj_cells = []
>                 queue = deque([(r, c)])
>                 visited[r][c] = True
>                 color = input_grid[r][c]
>                 while queue:
>                     cr, cc = queue.popleft()
>                     obj_cells.append((cr, cc))
>                     for dr, dc in [(-1,0),(1,0),(0,-1),(0,1)]:
>                         nr, nc = cr+dr, cc+dc
>                         if (0 <= nr < rows and 0 <= nc < cols
>                             and not visited[nr][nc]
>                             and input_grid[nr][nc] == color):
>                             visited[nr][nc] = True
>                             queue.append((nr, nc))
>                 objects.append((color, obj_cells))
>
>     # Sort by size (descending)
>     objects.sort(key=lambda x: len(x[1]), reverse=True)
>
>     # Rebuild grid with sorted placement
>     output = [[0] * cols for _ in range(rows)]
>     # ... placement logic depends on specific task
>     return output
> ```

## 12 Iterative Refinement Loop

### 12.1 Refinement Architecture

The iterative refinement loop is where the second and third core principles (extended reasoning
      horizons and measurable feedback) come together. Each agent can perform up to 10 iterations
      (configurable via `GEMINI_CLI_MAX_ITERATIONS=10`), with each iteration building on the
      feedback from the previous one.

### 12.2 Error Feedback Categories

The refinement prompt includes detailed error information categorized by type:

| Error Category | Information Provided | Refinement Strategy |
| --- | --- | --- |
| Syntax Error | Python traceback with line number | Fix syntax; often trivial |
| Runtime Error | Exception type, message, traceback | Fix logic errors (index bounds, type mismatches) |
| Wrong Output | Expected vs. actual grids, cell-level diff | Revise transformation rule |
| Partial Match | Percentage correct, specific mismatched regions | Adjust edge cases while preserving core logic |
| Timeout | Execution exceeded time limit | Optimize algorithm or fix infinite loop |

### 12.3 Refinement Prompt Construction

Python -- Refinement Prompt Builder

def build(
    self,
    task: ArcTask,
    previous_code: str,
    iteration: int,
    errors: List[Dict[str, Any]]
) -> str:
    sections = []

    # Context header
    sections.append(
        f"## Refinement Iteration {iteration + 1}\n"
        f"Your previous attempt failed on "
        f"{len(errors)} training examples."
    )

    # Previous code
    sections.append(
        f"## Previous Code\n```python\n{previous_code}\n```"
    )

    # Detailed error analysis
    sections.append("## Error Analysis")
    for error in errors:
        pair_idx = error['pair']
        if 'error' in error:
            sections.append(
                f"### Training Pair {pair_idx}: Runtime Error\n"
                f"```\n{error['error']}\n```"
            )
        elif not error['passed']:
            expected = error.get('expected', [])
            actual = error.get('actual', [])
            diff = self._compute_grid_diff(expected, actual)
            sections.append(
                f"### Training Pair {pair_idx}: Wrong Output\n"
                f"Expected:\n{self._format_grid(expected)}\n"
                f"Got:\n{self._format_grid(actual)}\n"
                f"Differences:\n{diff}"
            )

    # Original task for reference
    sections.append(
        f"## Original Task\n{task.format_for_prompt()}"
    )

    # Refinement instructions
    sections.append(
        "## Instructions\n"
        "1. Carefully analyze the errors above\n"
        "2. Identify what your transformation got wrong\n"
        "3. Consider alternative transformation rules\n"
        "4. Write a corrected transform() function\n"
        "5. Ensure it handles ALL training examples"
    )

    return "\n\n".join(sections)

@staticmethod
def _compute_grid_diff(
    expected: List[List[int]],
    actual: List[List[int]]
) -> str:
    """Compute cell-level diff between grids."""
    diffs = []
    for r in range(len(expected)):
        for c in range(len(expected[r])):
            exp_val = expected[r][c]
            act_val = (actual[r][c]
                       if r < len(actual) and c < len(actual[r])
                       else "MISSING")
            if exp_val != act_val:
                diffs.append(
                    f"  Cell ({r},{c}): expected {exp_val}, "
                    f"got {act_val}"
                )
    if not diffs:
        return "  No differences (dimension mismatch?)"
    return "\n".join(diffs[:20])  # Limit to 20 diffs

@staticmethod
def _format_grid(grid: List[List[int]]) -> str:
    return "\n".join(str(row) for row in grid)