# core


<!-- WARNING: THIS FILE WAS AUTOGENERATED! DO NOT EDIT! -->

## Overview

This module implements the RLM (Recursive Language Model) protocol from
[rlmpaper](https://github.com/alexzhang13/rlm) using
[claudette](https://claudette.answer.ai) as the LLM backend.

### Design Principles

- **Namespace-explicit**: All functions take `ns: dict` parameter (no
  frame walking)
- **Protocol-faithful**: Uses prompts and types from
  `_rlmpaper_compat.py`
- **Solveit-independent**: Core works anywhere; Solveit integration is
  separate
- **Return everything useful**: Functions return
  `(answer, iterations, ns)` for inspection

## Imports

## Environment Detection

------------------------------------------------------------------------

### in_solveit

``` python

def in_solveit(
    
)->bool:

```

*Check if running in Solveit environment.*

Solveit injects `__msg_id` into the call stack. This function tests for
that.

``` python
# Test environment detection
is_solveit = in_solveit()
print(f"Running in Solveit: {is_solveit}")
assert isinstance(is_solveit, bool)
```

    Running in Solveit: False

## Core LLM Functions

These functions provide the REPL environment’s `llm_query` and
`llm_query_batched` capabilities. They require an explicit namespace
dict and store results there.

------------------------------------------------------------------------

### llm_query

``` python

def llm_query(
    prompt:str, ns:dict, name:str='llm_res', model:str='claude-sonnet-4-5'
)->str:

```

*Query a sub-LLM and store the result in namespace.*

Args: prompt: The prompt to send to the LLM ns: Namespace dict where
result will be stored name: Variable name for storing the result model:
Claude model to use

Returns: The LLM’s response (also stored in ns\[name\])

``` python
# Test llm_query with explicit namespace
test_ns = {}
# Note: Commented out to avoid API calls during CI
# result = llm_query("Say 'hello' and nothing else", test_ns, name='greeting')
# assert 'greeting' in test_ns
# assert len(test_ns['greeting']) > 0
print("✓ llm_query signature test passed")
```

    ✓ llm_query signature test passed

------------------------------------------------------------------------

### llm_query_batched

``` python

def llm_query_batched(
    prompts:list, ns:dict, name:str='batch_res', model:str='claude-sonnet-4-5'
)->list:

```

*Query LLM with multiple prompts concurrently.*

NOTE: Currently executes sequentially due to claudette’s synchronous
API. True concurrency pending claudette async API support. The async
scaffolding is in place for when that becomes available.

Args: prompts: List of prompt strings ns: Namespace dict where results
will be stored name: Variable name for storing the list of results
model: Claude model to use

Returns: List of LLM responses (also stored in ns\[name\])

``` python
# Test llm_query_batched signature
test_ns = {}
# Note: Commented out to avoid API calls during CI
# prompts = ["Say 'one'", "Say 'two'"]
# result = llm_query_batched(prompts, test_ns, name='batch')
# assert 'batch' in test_ns
# assert len(test_ns['batch']) == 2
print("✓ llm_query_batched signature test passed")
```

    ✓ llm_query_batched signature test passed

## REPL Execution

Execute Python code in a namespace and capture stdout/stderr.

------------------------------------------------------------------------

### exec_code

``` python

def exec_code(
    code:str, ns:dict
)->REPLResult:

```

*Execute code in namespace and return result.*

Captures stdout, stderr, and any exceptions. The namespace is mutated
with any variables created during execution.

Args: code: Python code to execute ns: Namespace dict for execution

Returns: REPLResult with stdout, stderr, locals snapshot, execution_time

``` python
# Test exec_code with explicit namespace
test_ns = {}
result = exec_code("x = 2 + 2\nprint(x)", test_ns)
assert test_ns['x'] == 4
assert '4' in result.stdout
assert result.execution_time > 0
print("✓ exec_code works")

# Test error handling
test_ns = {}
result = exec_code("raise ValueError('test error')", test_ns)
assert 'ValueError: test error' in result.stderr
print("✓ exec_code error handling works")
```

    ✓ exec_code works
    ✓ exec_code error handling works

## RLM Loop

The main RLM iteration loop. Follows the rlmpaper protocol: 1. Build
system prompt with metadata 2. Add first-iteration safeguard 3. Execute
REPL code blocks 4. Check for FINAL/FINAL_VAR 5. Repeat until answer or
max iterations

------------------------------------------------------------------------

### rlm_run

``` python

def rlm_run(
    query:str, context, ns:dict=None, model:str='claude-sonnet-4-5', max_iters:int=10, logger:NoneType=None,
    verbose:bool=False
)->tuple:

```

*Run RLM loop until FINAL or max iterations.*

This implements the RLM protocol: the root LLM emits `repl` code blocks
which are executed in a namespace with `context`, `llm_query`,
`llm_query_batched`, and `FINAL_VAR` available. The loop continues until
the model returns FINAL(…) or FINAL_VAR(…).

Args: query: User’s question to answer context: Context data (str, list
of str, or dict) ns: Namespace dict (if None, creates fresh namespace)
model: Claude model to use max_iters: Maximum iterations before giving
up logger: RLMLogger instance for JSON-lines logging (optional) verbose:
Enable Rich console output (default: False)

Returns: (answer, iterations, namespace) tuple where: - answer: Final
answer string (or fallback if max_iters reached) - iterations: List of
RLMIteration objects - namespace: The dict containing all REPL variables

## Tests

``` python
# Test QueryMetadata (from _rlmpaper_compat)
meta = QueryMetadata(["chunk1", "chunk2", "chunk3"])
assert meta.context_type == "list"
assert len(meta.context_lengths) == 3
assert meta.context_total_length == sum(len(c) for c in ["chunk1", "chunk2", "chunk3"])
print("✓ QueryMetadata works")
```

    ✓ QueryMetadata works

```` python
# Test find_code_blocks (from _rlmpaper_compat)
text = """Here's some code:
```repl
x = 1 + 1
print(x)
```
And more text."""
blocks = find_code_blocks(text)
assert len(blocks) == 1
assert 'x = 1 + 1' in blocks[0]
print("✓ find_code_blocks works")
````

    ✓ find_code_blocks works

``` python
# Test find_final_answer (from _rlmpaper_compat)
assert find_final_answer("FINAL(42)") == "42"
assert find_final_answer("FINAL(The answer is 42)") == "The answer is 42"

# Test FINAL_VAR
test_ns = {'result': 'hello world'}
assert find_final_answer("FINAL_VAR(result)", ns=test_ns) == "hello world"

# Test no final
assert find_final_answer("Just some text") is None
print("✓ find_final_answer works")
```

    ✓ find_final_answer works

``` python
# Test rlm_run with simple mock scenario
# Note: This doesn't call LLM APIs, just tests the structure
context = ["The capital of France is Paris."]
test_ns = {}

# We can't easily test without API calls, but we can verify the function signature
# and that it sets up the namespace correctly
meta = QueryMetadata(context)
test_ns['context'] = context
test_ns['llm_query'] = partial(llm_query, ns=test_ns, model='claude-sonnet-4-5')
test_ns['llm_query_batched'] = partial(llm_query_batched, ns=test_ns, model='claude-sonnet-4-5')

# Test that FINAL_VAR is now set up as a callable function
def _test_final_var(variable_name: str) -> str:
    variable_name = variable_name.strip().strip('"').strip("'")
    if variable_name in test_ns:
        return str(test_ns[variable_name])
    return f"Error: Variable '{variable_name}' not found in namespace"

test_ns['FINAL_VAR'] = _test_final_var

assert 'context' in test_ns
assert 'llm_query' in test_ns
assert 'llm_query_batched' in test_ns
assert 'FINAL_VAR' in test_ns
assert callable(test_ns['llm_query'])
assert callable(test_ns['llm_query_batched'])
assert callable(test_ns['FINAL_VAR'])

# Test FINAL_VAR function behavior
test_ns['my_var'] = 'test value'
assert test_ns['FINAL_VAR']('my_var') == 'test value'
assert 'Error' in test_ns['FINAL_VAR']('nonexistent')
print("✓ rlm_run namespace setup works")
print("✓ FINAL_VAR executable function works")
```

    ✓ rlm_run namespace setup works
    ✓ FINAL_VAR executable function works

## Usage Examples

``` python
# Simple usage (requires API key)
context = ["The speed of light is 299,792,458 m/s."]
answer, iterations, ns = rlm_run("What is the speed of light?", context)
print(f"Answer: {answer}")
print(f"Iterations: {len(iterations)}")
print(f"Variables in namespace: {[k for k in ns.keys() if not k.startswith('_')]}")
```

``` python
# Persistent namespace across runs
ns = {}

# Define example contexts
context1 = {'prompt': 'Query ontology for X', 'tools': [...]}
context2 = {'prompt': 'Query ontology for Y', 'tools': [...]}

# Run multiple queries, reusing namespace
answer1, iters1, ns = rlm_run("What is X?", context1, ns=ns)
answer2, iters2, ns = rlm_run("What about Y?", context2, ns=ns)
# ns now contains variables from both runs
```