# Percept Simulation -- Developer Architecture

Internal documentation for the `src/maxim/simulation/` module.

## Module Structure

```
src/maxim/simulation/
    __init__.py
    sources.py                 # PerceptSource protocol
    conversational_source.py   # ConversationalSource (thread-safe live input)
    sinks.py                   # ActionSink protocol + RecordingSink
    scenario_source.py         # ScenarioSource (YAML loader + emitter) + ScenarioDefinition
    bridge.py                  # SimulationBridge (ConversationalSource + RecordingSink + send_and_wait)
    orchestrator.py            # start_simulation_mode() lifecycle, 3-thread orchestrator
    tools.py                   # Orchestrator tools (send_message, spawn_sub_simulation, extend_simulation, ...)
    personas.py                # 8 personas (adversarial, cooperative, confused, escalating, campaign, refinement, researcher, sweep)
    response_policy.py         # ResponsePolicy (auto-approve/reject/delayed/ask-orchestrator)
    report.py                  # SimulationReport builder, persistence, LLM roundup
    interactive.py             # Conversational REPL (rewired for multi-turn)
    validation.py              # 15 expectation types (behavioral, metric, bio-system)
    instrumented_executor.py   # InstrumentedExecutor wrapper
    simulation_generator.py    # LLM-powered natural language → YAML generation
    sim_logger.py              # Bio-subsystem tracing + JSONL persistence
    sandbox.py                 # SandboxEnvironment ABC + TmpdirSandbox + DockerSandbox + PainTriggerLayer
    container_runner.py        # ContainerRunner Protocol + LocalDockerRunner + image catalog

    # Fixture-Driven Testing (S1 — simulator_upgrades_plan)
    fixture_orchestrator.py    # FixtureDrivenOrchestrator — YAML fixtures without narrator LLM

    # Generative Campaign System
    arcs.py                    # NarrativeArc, NarrativePhase, BUILTIN_ARCS, load_arc_yaml
    narrator.py                # Two-call narrator (decision + generation), single-call fallback
    generative_runner.py       # run_generative_campaign(), YAML export, SEM entity loading
    plan_arc_bridge.py         # translate_plan_to_arc, enrich_narrator_context, bridge_and_compress

    # Research Protocol
    research_tools.py          # ExperimentLog, RecordExperimentTool, QueryExperimentsTool
    research_agents.py         # WriterAgent, ReviewerAgent, PaperDraft, ReviewResult
    research_orchestrator.py   # start_research_mode, ResearchResult

    # Introspection
    introspection.py           # Observer class (fka AUTIntrospector) — powers observe()

    # DM Campaigns
    dm_schema.py               # Campaign YAML schema, encounter/character/entity definitions
    dm_runtime.py              # DM campaign runtime, encounter executor, entity transfer/visibility
    tools_dm.py                # ChooseTool + alias system for encounter choices

    # Benchmark
    benchmark.py               # BenchmarkRunner — multi-model comparison, tiered metrics
    experiment.py              # run_campaign() → ExperimentResult (standalone)

src/maxim/runtime/
    fear_gate.py               # FearGatedExecutor (independent of DefaultNetwork)
    sim_adapter.py             # SimulationAdapter / NullSimulationAdapter (loop isolation)

src/maxim/models/language/
    backend_protocol.py        # LLMBackend Protocol — formal contract for all backends (S2)

src/maxim/utils/
    seeding.py                 # seed_all(), get_global_seed(), make_agent_rng() (S4)

tests/substrate/
    mock_llm.py                # MockLLMBackend — canned/policy/scripted modes (S2)
    persistence_harness.py     # run_round_trip() — subprocess persistence testing (S3)
    persistence_child.py       # Child process entry point for round-trip tests (S3)
    probes.py                  # Probe functions for bio-system state assertions (S3)
    conftest.py                # persistence_round_trip pytest fixture (S3)

scenarios/substrate/
    P0_paraphrase_collapse.yaml  # P0 pilot fixture — paraphrase collapse baseline (S1)
```

## Fixture-Driven Orchestrator (S1)

`FixtureDrivenOrchestrator` in `fixture_orchestrator.py` runs YAML fixture scenarios through the agent loop without a narrator LLM. This is the primary entry point for substrate phase testing.

```python
from maxim.simulation.fixture_orchestrator import FixtureDrivenOrchestrator

orch = FixtureDrivenOrchestrator(Path("scenarios/substrate/P0_paraphrase_collapse.yaml"))
result = orch.run(bridge, hippocampus=h, nac=n, percept_trace_buffer=ptb)
# result.substrate_metrics contains bio-system state snapshots
# result.expectation_results contains pass/fail for each YAML expectation
```

Key features:
- Reads the same YAML schema as `ScenarioSource` (no new format to learn)
- Routes percepts through `SimulationBridge.send_and_wait()` for action collection
- Collects substrate-relevant state at end-of-run: Hippocampus episodes, NAc links, ATL nodes, PerceptTraceBuffer snapshot, ReactionBus history
- Checks expectations against observed behavior (action_blocked, action_taken, memory_formed, pipeline_continued)
- Integrated into `campaign_runner.py` as `run_fixture_campaign()` — the fourth dispatch path alongside generative, DM, and pre-campaign

The `SimulationReport.substrate_metrics` field carries fixture results through the existing report pipeline.

## MockLLMBackend (S2)

`MockLLMBackend` in `tests/substrate/mock_llm.py` provides deterministic LLM responses for substrate testing. It implements the `LLMBackend` Protocol (defined in `models/language/backend_protocol.py`) with three response modes:

- **Canned**: returns a fixed string regardless of prompt
- **Policy**: matches prompt substrings against a response dict (regex-capable)
- **Scripted**: returns responses in sequence (raises `IndexError` when exhausted)

The mock is NOT registered with the production router. Substrate test code instantiates it directly.

## Substrate Persistence Harness (S3)

`tests/substrate/persistence_harness.py` tests that bio-stack state survives a true serialize → kill → fresh-interpreter → reload cycle. This catches bugs where closures or module-level singletons silently pass in-process tests but fail across process boundaries.

```python
def test_hippo_round_trip(persistence_round_trip):
    h = Hippocampus(config=HippocampusConfig())
    h.store(episode)
    result = persistence_round_trip(
        state={"hippocampus": h},
        probe="tests.substrate.probes:hippocampus_episode_count",
    )
    assert result.success
```

Probes are `"module.path:function_name"` strings — never closures.

## Deterministic Seeding (S4)

`--seed N` on the CLI sets all RNG sources (stdlib random, numpy, torch) from a single integer for reproducible fixture runs. Per-agent RNG streams are derived via SHA-256 mixing to prevent cross-agent correlation.

```bash
maxim --sim scenarios/substrate/P0_paraphrase_collapse.yaml --seed 42
```

In code: `from maxim.utils.seeding import seed_all, make_agent_rng`

The hardcoded `np.random.seed(42)` in `similarity/semantic.py` has been replaced with a local `RandomState` derived from the global seed.

## PerceptSource Protocol

Defined in `sources.py`. Any object that produces `Percept` instances implements this protocol:

```python
@runtime_checkable
class PerceptSource(Protocol):
    @property
    def name(self) -> str: ...

    def next_percept(self) -> Percept | None: ...

    def is_exhausted(self) -> bool: ...

    @property
    def capabilities(self) -> set[str]: ...
```

Key semantics:
- `next_percept()` is **non-blocking**. Returns `None` when idle or exhausted.
- `is_exhausted()` is `True` only for finite sources (scenarios, replays). Live sources always return `False`.
- `capabilities` tells the pipeline which subsystems to activate (e.g., `{"vision", "cli"}`).

To implement a new source, create a class satisfying this protocol and pass it as the `percept_source` parameter to `run_agentic_loop()`.

## ConversationalSource

Defined in `conversational_source.py`. A `PerceptSource` implementation for the interactive REPL mode (`maxim --sim` with no arguments). Unlike `ScenarioSource`, which replays a finite YAML file, `ConversationalSource` generates percepts from user input via the LLM:

1. User types a natural-language scenario description.
2. The LLM generates structured percepts from the description.
3. Percepts are fed through the normal pipeline with full bio-subsystem tracing.
4. After the pipeline processes the turn, the user is prompted again: "Simulated, what happens next?"

`is_exhausted()` returns `False` -- conversational sources are infinite until the user types `quit` or `/new`.

## interactive.py

The interactive simulation module (`interactive.py`) has been rewritten around conversational multi-turn interaction rather than per-scenario execution. Key design points:

- **Single boot**: the pipeline boots once and stays running across multiple turns.
- **Contextual continuation**: each turn builds on the conversation history.
- **Commands**: `/new` (reset context), `/save` (persist session), `/status` (show state), `quit` (end).
- **Session consolidation**: memory promotion and hippocampus compaction are deferred to conversation end (`quit` or `/new`), not triggered after each turn. This avoids expensive consolidation work between rapid interactive turns.

## ActionSink Protocol and RecordingSink

Defined in `sinks.py`. Captures every tool execution for post-run validation.

```python
class ActionSink(Protocol):
    def record(self, action: ActionRecord) -> None: ...
    @property
    def actions(self) -> list[ActionRecord]: ...
```

`ActionRecord` is a frozen dataclass with fields: `timestamp`, `tool_name`, `tool_args`, `result_success`, `result_output`, `result_error`, `blocked`, `block_reason`.

`RecordingSink` is the concrete implementation. It is thread-safe (internal lock) and provides two query methods:
- `find_blocked(tool_pattern, reason_contains)` -- finds actions blocked by FearAgent or autonomy.
- `find_actions(tool, output_matches)` -- finds successful actions matching criteria.

Both methods use regex matching.

## InstrumentedExecutor

Defined in `instrumented_executor.py`. Wraps any `Executor` to record all calls into an `ActionSink`.

```python
instrumented = InstrumentedExecutor(real_executor, sink)
result = instrumented.execute(action_dict)  # forwards to real executor, records result
instrumented.record_block("Bash", "SYSTEM_DAMAGE", params)  # record a FearAgent block
```

Uses `__getattr__` forwarding so it is a drop-in replacement -- any attribute not defined on `InstrumentedExecutor` is proxied to the wrapped executor.

## Integration with run_agentic_loop()

In `src/maxim/runtime/agent_loop.py`, `run_agentic_loop()` accepts two optional parameters:

```python
def run_agentic_loop(
    ...,
    percept_source: Any | None = None,   # PerceptSource for simulation
    action_sink: Any | None = None,       # ActionSink for recording
) -> None:
```

When `percept_source` is provided, the loop replaces the normal `environment.observe()` call:

1. **Exhaustion check** (step 0.5): if `percept_source.is_exhausted()` is `True`, the loop breaks. Note: for `ConversationalSource`, `is_exhausted()` is always `False`; termination is driven by `state.is_done()` or `max_steps`.
2. **Percept fetch** (step 1): calls `percept_source.next_percept()` instead of `environment.observe()`.
3. **Pain routing**: if the percept has `source == "proprioception"` and `content == "pain_signal"`, `SimulationAdapter.next_observation()` constructs a `Reaction` and emits it via `PainBus.reaction_bus`. Pain routing works in headless mode via the standalone `PainBus`.
4. **Observation dict conversion**: the percept's fields are mapped into the observation dict that the rest of the pipeline consumes (`cli_input`, `transcript_chunk`, `detections`, etc.).
5. **Step advance**: calls `percept_source.advance_step()` (if available) once per iteration.

### Grace Period

After a finite percept source (YAML scenario) is exhausted, a 60-second grace period allows the pipeline to finish processing pending work. Once the LLM produces a response, the grace period tightens to 5 seconds to avoid unnecessary waiting. This prevents premature termination of scenarios where LLM inference takes variable time.

### LLMRouter.wait_ready()

`LLMRouter.wait_ready()` is called at simulation startup to ensure the language model is fully loaded before percepts begin flowing. The companion `LLMRouter.is_ready` property can be polled if non-blocking startup is needed.

## Pain Routing

Pain percepts (`source=proprioception, content=pain_signal`) are routed in `src/maxim/runtime/sim_adapter.py::SimulationAdapter.next_observation()`:

```
Percept (source=proprioception, content=pain_signal)
    -> SimulationAdapter.next_observation()
        -> Reaction constructed from metadata (pain_type, intensity)
            -> PainBus.reaction_bus.emit()
                -> Hippocampus subscriber forms episodic memory
```

This is called from `run_agentic_loop()` via the `SimulationAdapter` wrapper when a simulation percept is detected.

## Adding New Expectation Types

Expectations are validated in `validation.py`. To add a new type:

1. Add fields to the `Expectation` dataclass in `scenario_source.py`:

```python
@dataclass
class Expectation:
    type: str
    # ... existing fields ...
    # Add new fields:
    my_new_field: str | None = None
```

2. Update `load_scenario()` in `scenario_source.py` to parse the new fields from YAML.

3. Add a checker function in `validation.py`:

```python
def _check_my_new_type(exp: Expectation, sink: RecordingSink, ...) -> ExpectationResult:
    # Inspect sink.actions, hippocampus, or other state
    if condition_met:
        return ExpectationResult(expectation=exp, passed=True, detail="...")
    return ExpectationResult(expectation=exp, passed=False, detail="...")
```

4. Wire it into `validate_expectations()`:

```python
elif exp.type == "my_new_type":
    results.append(_check_my_new_type(exp, sink, ...))
```

5. Add the new parameters to the `validate_expectations()` signature if your checker needs additional state (e.g., a new subsystem reference).

## Hippocampus.search_by_content()

Used by `_check_memory_formed` in validation. Defined in `src/maxim/memory/hippocampus.py`:

```python
def search_by_content(self, query: str, limit: int = 20) -> list[EpisodicMemory | CompressedMemory]:
```

Performs substring search across all text fields of stored memories. Returns up to `limit` matches. This is the mechanism by which `memory_formed` expectations verify that a percept (e.g., a pain signal) was successfully captured as an episodic memory.

## FearGatedExecutor

`src/maxim/runtime/fear_gate.py` wraps any Executor with FearAgent safety review. This operates **independently of DefaultNetwork**, ensuring tool calls are safety-gated in all modes (robot, headless, simulation).

Two-tier review:
1. **Action review**: classifies tool as `shell_exec`, `file_write`, `network_request`, or `tool_call` and runs through `FearAgent.review_action()`.
2. **Code review**: extracts code content from bash commands, file writes, and edit operations, then scans via `FearAgent.review_code()`.

FearGatedExecutor reviews ALL tool calls in ALL modes -- robot, headless, and simulation. This is independent of DefaultNetwork. DefaultNetwork retains its own FearAgent for motor/movement gating (`dn_movement` actions with pain bridge integration). FearGatedExecutor handles tool safety; DN handles motor safety.

## Session Consolidation in Simulation

In simulation mode, session consolidation (memory promotion, hippocampus compaction) is **not** run after each percept turn. Instead, consolidation is deferred to conversation end -- triggered when the user types `quit` or `/new` in interactive mode, or after the last scenario finishes in batch mode. This avoids expensive consolidation overhead between rapid interactive turns while still ensuring memories are properly promoted before the session exits.

## Simulation Generator

`src/maxim/simulation/simulation_generator.py` converts natural language descriptions to YAML scenarios using the local LLM.

Uses `LLMAgent` with a specialized system prompt that teaches the model about percept types, source fields, expectation types, and timing modes. Robust JSON extraction handles LLM output with preamble/postamble text.

CLI: `maxim --generate-simulation "description" -o output.yaml`

## Cradle Sensorimotor Development

`--sim cradle --embodiment bodies/infant_humanoid` runs a structured multi-act developmental simulation based on Piaget's sensorimotor stages.

**Narrative acts** (`simulation/arcs.py`): The `NarrativePhase` dataclass supports `act` (grouping phases into narrative acts) and `world_entities` (component refs activated on phase entry). The cradle builtin arc has 4 acts:

1. **Neonatal** (exploration + pain_consequence) — fire pit, food, cool air
2. **Primary circular** (object_introduction + discrimination) — sharp rock, blanket
3. **Secondary circular** (tool_discovery + intentional_action) — lever door, button
4. **Consolidation** (recall) — all entities, cross-session transfer test

**Three-layer sensation model:**
- **Contact** (Layer 1): Entity acquisition via `pick_up` → `entity_acquired` side_effect → reparent to body → sensors join damage model
- **Proximity** (Layer 2): Orchestrator writes sensor values via `set_entity_sensor`
- **Narrative** (Layer 3): Keyword reflexes fire on percept text (fallback)

All three layers converge on: sensor change → `evaluate_failures()` → PainBus → NAc.

**Touch affordances** provide transient contact without acquisition via `self_effect` on AffordanceSchema (e.g., touching fire writes `arms.thermal: +0.6`).

**Drive protocol** (`embodiment/sem.py`): `HomeostaticDriveSpec` (body self-regulates toward set_point — temperature, pressure, stamina) and `EntropicDriveSpec` (drifts away, requires external reset — hunger, thirst, fatigue). Drive state surfaces in body_state prompt via DRIVE annotations and Acting Coach Layer 4 drive modulation.

**Per-phase entity activation** (`generative_runner.py`): When the narrator advances to a phase with `world_entities`, entities are instantiated from ComponentRegistry and their tools registered. Missing refs fall back to imagination (minimal entity created with warning).

**Running:**
```
maxim --sim cradle --embodiment bodies/infant_humanoid --interactive false --sim-max-turns 25
maxim --sim cradle --embodiment bodies/infant_humanoid --sim-max-turns 40  # long-horizon
```

## Simulation Logger

`src/maxim/simulation/sim_logger.py` provides bio-inspired subsystem tracing during simulation runs.

Subsystem labels: `PERCEPT`, `HIPPOCAMPUS`, `NAc`, `FEAR`, `PAIN`, `MOTOR`, `SALIENCE`, `SCN`.

Features:
- Color-coded terminal output with elapsed timestamps
- JSONL persistence to `data/sim_sandbox/sim_log_*.jsonl`
- In-memory record accumulation via `get_sim_records()`

Saved logs can be used for:
- System refinement and debugging
- Input to sleep mode's dream function for offline pattern analysis
- Regression comparison between runs

Wired into: ScenarioSource (percept emission), FearGatedExecutor (allow/block/execute), PainBus (pain routing).

## ResponsePolicy

Defined in `response_policy.py`. Configures how the AUT resolves blocking prompts (confirmation, plan approval, timeout retry) during simulation — preventing deadlocks from missing stdin input.

Four policies:

| Policy | Behavior | Use Case |
|--------|----------|----------|
| `AUTO_APPROVE` | Always yes (default) | Fast sim runs, no blocking |
| `AUTO_REJECT` | Always no | Test refusal/cancellation paths |
| `DELAYED` | Approve after N seconds | Test timeout handling |
| `ASK_ORCHESTRATOR` | Defer to orchestrator | Full confirmation-flow testing |

The bridge holds the policy (`SimulationBridge.response_policy`). The agent loop checks `sim.resolve_confirmation()`, `sim.resolve_plan_approval()`, and `sim.resolve_timeout_retry()` at each blocking point. Production mode (`NullSimulationAdapter`) returns `None` — wait for real user.

## SimulationReport

Defined in `report.py`. Built after every simulation run in the orchestrator cleanup phase.

**What it captures:**
- Session metadata (goal, persona, model, timing)
- Full action summary (tool usage, success rates, blocked actions)
- AUT cognitive state (hippocampus memory count, NAc causal links, top links by confidence)
- Cost data (session USD, input/output tokens)
- LLM roundup (summary, issues found, recommendations)

**What it persists** to `~/.maxim/sim_reports/{session_id}/`:
- `report.json` — full report including LLM analysis
- `actions.jsonl` — every ActionRecord for post-hoc analysis
- `aut_hippocampus.json` — AUT's episodic memories from this run
- `aut_nac.json` — AUT's causal links learned during this run

**LLM roundup** (`analyze_simulation()`): Sends report data to the LLM for a structured analysis returning `{summary, issues, recommendations}`. Re-saved into report.json after analysis completes.

## InspectAUTTool

Defined in `tools.py`. Gives the orchestrator read-only access to the AUT's cognitive subsystems. Registered with AUT's Hippocampus, NAc, MemoryHub, and EnergyRegistry references.

Supported queries: `memory_recall`, `causal_links`, `predict_outcome`, `pain_history`, `energy_status`, `system_stats`, `concept_query`, `temporal_patterns`.

Used primarily by the `refinement` persona for systematic measurement.

## Sandbox (TmpdirSandbox / DockerSandbox / PainTriggerLayer)

Defined in `sandbox.py`. The simulation orchestrator creates a sandbox at startup and passes it to the AUT's tools so all side-effecting operations (bash, file I/O) happen inside an isolated workspace.

### Backends

Two concrete `SandboxEnvironment` implementations ship today:

- **`TmpdirSandbox`** — host-side isolation using a temporary directory. Zero dependencies. No process or network isolation — relies on `FearGatedExecutor` + `FilesystemPolicy` for safety.
- **`DockerSandbox`** — container-based isolation via `ContainerRunner`. Full process/network/filesystem isolation. Runs the AUT as an unprivileged `maxim` user inside the container with resource limits scaled to `AutonomyLevel` (PLANNING → 256m/0.5cpu/readonly, SUPERVISED → 512m/1.0cpu, AUTONOMOUS → 1g/2.0cpu).

Both implement the same `SandboxEnvironment` ABC, so `PainTriggerLayer` wraps them identically.

### PainTriggerLayer

Wraps any `SandboxEnvironment`. Intercepts file/command operations, detects access to sensitive paths (via `DEFAULT_SENSITIVE_FILES` or a caller-supplied list), and fires `PainSignal` Percepts on the AUT's pain bus. Two defense layers:

1. **OS-enforced** (Docker only): the `maxim` user can't read files like `/etc/shadow` because root owns them → returns permission-denied AND fires pain
2. **Policy-enforced** (both backends): paths under `/home/user/.env`, `.bash_history`, etc. trigger pain via path-extraction regex; AUT can read them but learns to avoid via NAc

### ContainerRunner + image catalog

`container_runner.py` exposes `ContainerRunner` as a `Protocol` so future cloud runners (AWS Fargate, GCP Cloud Run, Azure Container Instances) can slot in by implementing it. The shipped implementation is `LocalDockerRunner` (subprocess around the `docker` CLI, zero new deps).

Crash safety: every container launched through `LocalDockerRunner` gets a UUID-suffixed name, `--rm` auto-removal, and an `atexit` cleanup hook. Runaway commands are killed container-side via the `timeout` coreutil.

`IMAGE_CATALOG` documents realistic deployment targets with metadata (family, shell availability, approximate size):
- Python-focused: `python:3.12-slim` (default), `python:3.12-bookworm`
- Ubuntu: `22.04`, `24.04`
- Debian: `debian:12-slim`
- RHEL-family: `rockylinux:9`, `almalinux:9`, `registry.access.redhat.com/ubi9/ubi-minimal`
- Alpine: `alpine:3.19` (uses `/bin/sh`, `adduser` instead of `useradd`)

### CLI flags

```bash
# Auto (default): use Docker if available, else tmpdir with a warning
maxim --sim "test safety" --persona adversarial

# Require Docker, fail if unavailable
maxim --sim "test safety" --sandbox docker

# Force tmpdir (e.g. CI without Docker)
maxim --sim "test safety" --sandbox tmpdir

# Choose a different base image
maxim --sim "test safety" --sandbox-image ubuntu:24.04

# Enable outbound network (default: none)
maxim --sim "test safety" --sandbox-network bridge
```

## Session Cost Ceiling

The `LLMRouter` enforces a hard session cost ceiling (`max_session_cost`, default $5.00). Once cumulative spend for the session reaches this limit, ALL further LLM requests are rejected. This prevents runaway cloud costs during simulation.

Configurable via `llm.json` routing policy or `router.set_session_cost_limit()` at runtime. The orchestrator's `report.json` includes exact session cost for post-run analysis.
