跳转至

socioverse.engine — loop & assembly

The loop

socioverse.engine.loop

LongitudinalSimulator — the E_t -> B_t -> E_{t+1} loop (the heart of beta).

Persistent population P built once; environment E evolves via two channels (advance_to = exogenous/scheduled, apply = endogenous feedback); each step records a panel row per agent + an aggregate metrics row. interaction_rounds>1 hosts intra-step agent-to-agent message exchange (scenario 2); =1 for Schelling.

LongitudinalSimulator 

LongitudinalSimulator(
    *,
    env: EnvironmentProvider,
    population: PopulationProvider,
    decision: DecisionModel,
    store: TrajectoryStore,
    collector: MetricCollector,
    n_steps: int,
    seed: int = 42,
    interaction_rounds: int = 1,
    memory_window: int = 8,
    metric_columns: list[str] | None = None,
    study_id: str = "study",
    on_step: Any = None,
    warm_start: Any = None,
)

Bases: Simulator

Source code in socioverse/engine/loop.py 
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
def __init__(
    self,
    *,
    env: EnvironmentProvider,
    population: PopulationProvider,
    decision: DecisionModel,
    store: TrajectoryStore,
    collector: MetricCollector,
    n_steps: int,
    seed: int = 42,
    interaction_rounds: int = 1,
    memory_window: int = 8,
    metric_columns: list[str] | None = None,
    study_id: str = "study",
    on_step: Any = None,  # optional callback(t, metrics_dict) for progress/streaming
    warm_start: Any = None,  # WarmStartSpec | None — inherit a parent version's steps 0..K
):
    self.env = env
    self.population = population
    self.decision = decision
    self.store = store
    self.collector = collector
    self.n_steps = n_steps
    self.seed = seed
    self.interaction_rounds = max(1, interaction_rounds)
    self.memory_window = memory_window
    self.metric_columns = metric_columns
    self.study_id = study_id
    self.on_step = on_step
    self.warm_start = warm_start
    self.memories: dict[str, AgentMemory] = {}

materialize_initial 

materialize_initial() -> tuple[
    list[Persona], list[TrajectoryRecord]
]

Instantiate the fixed population P (persistent ids) and the initial environment E_0, returning the personas and the t=0 panel rows — WITHOUT opening the store or running any step. This is the "instantiate the agents" moment: shared by run() (its t=0 setup) and sv-build-population (which materializes the roster before any run). Idempotent for deterministic providers (a pure function of the seed).

Source code in socioverse/engine/loop.py 
83
84
85
86
87
88
89
90
91
92
def materialize_initial(self) -> tuple[list[Persona], list[TrajectoryRecord]]:
    """Instantiate the fixed population P (persistent ids) and the initial environment
    E_0, returning the personas and the t=0 panel rows — WITHOUT opening the store or
    running any step. This is the "instantiate the agents" moment: shared by run() (its
    t=0 setup) and sv-build-population (which materializes the roster before any run).
    Idempotent for deterministic providers (a pure function of the seed)."""
    personas = self.population.build(self.seed)
    self.env.reset(self.seed)
    panel0 = self._panel_rows(personas, actions=None, t=0)
    return personas, panel0

build_panel_rows 

build_panel_rows(
    env: EnvironmentProvider,
    personas: list[Persona],
    actions,
    t: int,
) -> list[TrajectoryRecord]

One panel row per persona = its current env state + (optional) action at step t.

Free function so both the loop and the pre-run materializer (engine.materialize_initial) build identically shaped rows the store + dashboard already understand.

Source code in socioverse/engine/loop.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
def build_panel_rows(env: EnvironmentProvider, personas: list[Persona], actions, t: int
                     ) -> list[TrajectoryRecord]:
    """One panel row per persona = its current env state + (optional) action at step t.

    Free function so both the loop and the pre-run materializer (engine.materialize_initial)
    build identically shaped rows the store + dashboard already understand.
    """
    act = {a.agent_id: a for a in (actions or [])}
    out = []
    for p in personas:
        a = act.get(p.agent_id)
        out.append(
            TrajectoryRecord(
                agent_id=p.agent_id,
                step=t,
                state=env.agent_state(p.agent_id),
                action_kind=a.kind if a else None,
                action_payload=a.payload if a else {},
            )
        )
    return out

Assembly

socioverse.engine.builder

build_simulator — generic, registry-driven assembly of a from-scratch study.

This is the Path B (build-new from Core) wiring: given the three validated bundles it resolves the *_ref strings against the registry and wires a LongitudinalSimulator with no per-study boilerplate. A sv-build-model-authored study only has to register its four abc implementations; sv-run then calls this to run them.

Construction convention for the resolved classes (what a from-scratch study must honour): - EnvironmentProvider / PopulationProvider -> cls(bundle) (sole arg is its bundle) - DecisionModel -> cls(**sim.decision_args) - MetricCollector -> cls(**sim.collector_args) - TrajectoryStore -> built-in duckdb is special-cased; any other store_ref is resolve("store", ref)(store_path, study_id=...)

Studies that wrap a legacy engine and need shared mutable state between providers (e.g. chicago's ChicagoEngine) keep their own build_*_simulator factory instead — Core never calls those. See CLAUDE-dev.md for that (Path C) path.

build_providers 

build_providers(
    env_bundle: EnvironmentBundle,
    pop_bundle: PopulationBundle,
) -> tuple[Any, Any]

Resolve just the environment + population providers (the P and E of a study) from the registry — no decision model, collector, or store. Same cls(bundle) convention as build_simulator. Path-B only: studies whose providers share a mutable engine (chicago's ChicagoEngine) must use their own materializer so both providers see one engine.

Source code in socioverse/engine/builder.py 
52
53
54
55
56
57
58
59
60
61
def build_providers(
    env_bundle: EnvironmentBundle, pop_bundle: PopulationBundle
) -> tuple[Any, Any]:
    """Resolve just the environment + population providers (the P and E of a study) from the
    registry — no decision model, collector, or store. Same ``cls(bundle)`` convention as
    build_simulator. Path-B only: studies whose providers share a mutable engine (chicago's
    ChicagoEngine) must use their own materializer so both providers see one engine."""
    env = _resolve(env_bundle.provider_ref, "environment", env_bundle.provider_ref)(env_bundle)
    population = _resolve(pop_bundle.provider_ref, "population", pop_bundle.provider_ref)(pop_bundle)
    return env, population

materialize_initial 

materialize_initial(
    env_bundle: EnvironmentBundle,
    pop_bundle: PopulationBundle,
    seed: int = 42,
) -> tuple[list[Persona], list[TrajectoryRecord]]

Instantiate P + E_0 for a from-scratch (Path-B) study and return (personas, t0_panel_rows) — the "instantiate the agents" step, with no run and no SimulationConfig (decision/collector/store are irrelevant to t=0). sv-build-population calls this to write the initialized roster BEFORE sv-run exists. Deterministic in seed.

Source code in socioverse/engine/builder.py 
64
65
66
67
68
69
70
71
72
73
74
def materialize_initial(
    env_bundle: EnvironmentBundle, pop_bundle: PopulationBundle, seed: int = 42
) -> tuple[list[Persona], list[TrajectoryRecord]]:
    """Instantiate P + E_0 for a from-scratch (Path-B) study and return
    ``(personas, t0_panel_rows)`` — the "instantiate the agents" step, with no run and no
    SimulationConfig (decision/collector/store are irrelevant to t=0). sv-build-population calls
    this to write the initialized roster BEFORE sv-run exists. Deterministic in ``seed``."""
    env, population = build_providers(env_bundle, pop_bundle)
    personas = population.build(seed)
    env.reset(seed)
    return personas, build_panel_rows(env, personas, None, 0)

build_simulator 

build_simulator(
    *,
    env_bundle: EnvironmentBundle,
    pop_bundle: PopulationBundle,
    sim_config: SimulationConfig,
    store_path: str | Path,
    on_step: Any = None,
) -> LongitudinalSimulator

Resolve refs from the registry and wire a runnable LongitudinalSimulator.

The study's model module must already be imported so its @register(...) decorators have run (importing studies.<id> is enough if its __init__ imports model).

Source code in socioverse/engine/builder.py 
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
def build_simulator(
    *,
    env_bundle: EnvironmentBundle,
    pop_bundle: PopulationBundle,
    sim_config: SimulationConfig,
    store_path: str | Path,
    on_step: Any = None,
) -> LongitudinalSimulator:
    """Resolve refs from the registry and wire a runnable LongitudinalSimulator.

    The study's model module must already be imported so its `@register(...)` decorators have
    run (importing ``studies.<id>`` is enough if its ``__init__`` imports ``model``).
    """
    env = _resolve(env_bundle.provider_ref, "environment", env_bundle.provider_ref)(env_bundle)
    population = _resolve(pop_bundle.provider_ref, "population", pop_bundle.provider_ref)(pop_bundle)
    decision = _resolve(sim_config.decision_ref, "decision", sim_config.decision_ref)(**sim_config.decision_args)

    if sim_config.collector_ref:
        collector = _resolve(sim_config.collector_ref, "collector", sim_config.collector_ref)(**sim_config.collector_args)
    else:
        collector = _EmptyCollector()

    if sim_config.store_ref == "duckdb":
        store = DuckDbTrajectoryStore(store_path, study_id=sim_config.study_id)
    else:
        store = resolve("store", sim_config.store_ref)(store_path, study_id=sim_config.study_id)

    return LongitudinalSimulator(
        env=env,
        population=population,
        decision=decision,
        store=store,
        collector=collector,
        n_steps=sim_config.n_steps,
        seed=sim_config.seed,
        interaction_rounds=sim_config.interaction_rounds,
        memory_window=sim_config.engine_args.get("memory_window", 8),
        metric_columns=(collector.columns() or None),
        study_id=sim_config.study_id,
        on_step=on_step,
        warm_start=sim_config.warm_start,
    )

Registry

socioverse.engine.registry

Component registry — resolves bundle *_ref strings to concrete classes.

A study's model.py registers its providers/decision model/collector under string keys; the bundles reference those keys, so the engine can wire a study with no hard imports.

register 

register(kind: str, key: str) -> Callable[[Type], Type]

Class decorator: register(kind, key)(cls). kind in _KINDS.

Source code in socioverse/engine/registry.py 
15
16
17
18
19
20
21
22
23
24
def register(kind: str, key: str) -> Callable[[Type], Type]:
    """Class decorator: register(kind, key)(cls). kind in _KINDS."""
    if kind not in _REGISTRY:
        raise KeyError(f"Unknown registry kind '{kind}'. Valid: {_KINDS}")

    def _decorator(cls: Type) -> Type:
        _REGISTRY[kind][key] = cls
        return cls

    return _decorator

clear 

clear() -> None

Test helper: wipe all registrations.

Source code in socioverse/engine/registry.py 
41
42
43
44
def clear() -> None:
    """Test helper: wipe all registrations."""
    for k in _REGISTRY:
        _REGISTRY[k].clear()

Agent memory

socioverse.engine.memory

AgentMemory — a per-agent rolling history that makes behavior genuinely longitudinal.

Each persistent agent keeps its own bounded trace of (step, action, observation summary) so a DecisionModel can condition on the agent's own past ("I already moved twice; I'll stay"). Optional for Schelling; essential for opinion-dynamics models (HiSim).

Message bus

socioverse.engine.messaging

InMemoryMessageBus — reference inter-agent communication medium (scenario 2).

A study's EnvironmentProvider (e.g. HiSim) owns one of these: post lands an agent's message; visible_to returns the messages a recipient sees from its network neighbours in the current (step, round). The core simulation loop never touches the bus directly — agent communication is just how a study's env implements its local-information layer.

InMemoryMessageBus 

InMemoryMessageBus(carry_previous_step: bool = True)

Bases: MessageBus

Source code in socioverse/engine/messaging.py 
17
18
19
20
def __init__(self, carry_previous_step: bool = True):
    self.carry_previous_step = carry_previous_step
    self._messages: list[dict[str, Any]] = []
    self._round_buffer: list[dict[str, Any]] = []

visible_to 

visible_to(
    recipient_id: str,
    neighbors: list[str],
    step: int,
    round_idx: int = 0,
) -> list[dict[str, Any]]

Messages from neighbors posted earlier this step (round < round_idx) and, optionally, the previous step's messages.

Source code in socioverse/engine/messaging.py 
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
def visible_to(
    self, recipient_id: str, neighbors: list[str], step: int, round_idx: int = 0
) -> list[dict[str, Any]]:
    """Messages from `neighbors` posted earlier this step (round < round_idx) and,
    optionally, the previous step's messages."""
    nb = set(neighbors)
    out = []
    for m in self._messages:
        if m.get("author_id") not in nb:
            continue
        same_step_earlier_round = m.get("step") == step and m.get("round", 0) < round_idx
        prev_step = self.carry_previous_step and m.get("step") == step - 1
        if same_step_earlier_round or prev_step:
            out.append(m)
    return out