extropy/tests/test_reasoning_execution.py at b29573a12997d6e98cf42cc1eba6ed63c856f18b · exaforge/extropy · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
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
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
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
"""Execution tests for reasoning internals that affect memory/accountability."""

import asyncio
from datetime import datetime
from unittest.mock import patch

from extropy.core.llm import TokenUsage
from extropy.core.models import ExposureRecord, ReasoningContext, SimulationRunConfig
from extropy.core.models.scenario import (
    Event,
    EventType,
    ExposureChannel,
    ExposureRule,
    InteractionConfig,
    InteractionType,
    OutcomeConfig,
    OutcomeDefinition,
    OutcomeType,
    ScenarioMeta,
    ScenarioSpec,
    SeedExposure,
    SimulationConfig,
    SpreadConfig,
    TimestepUnit,
)
from extropy.simulation.reasoning import _reason_agent_two_pass_async


def _make_context() -> ReasoningContext:
    return ReasoningContext(
        agent_id="a1",
        persona="I am a parent with limited time.",
        event_content="School board policy update",
        exposure_history=[
            ExposureRecord(
                timestep=0,
                channel="broadcast",
                content="Policy update",
                credibility=0.9,
                source_agent_id=None,
            )
        ],
        agent_name="Taylor",
    )


def _make_scenario_no_pass2() -> ScenarioSpec:
    return ScenarioSpec(
        meta=ScenarioMeta(
            name="reasoning_exec",
            description="Reasoning execution test",
            population_spec="test.yaml",
            study_db="study.db",
            population_id="default",
            network_id="default",
            created_at=datetime(2024, 1, 1),
        ),
        event=Event(
            type=EventType.NEWS,
            content="Policy update",
            source="Board",
            credibility=0.8,
            ambiguity=0.3,
            emotional_valence=0.0,
        ),
        seed_exposure=SeedExposure(
            channels=[
                ExposureChannel(
                    name="broadcast",
                    description="Broadcast",
                    reach="broadcast",
                )
            ],
            rules=[
                ExposureRule(
                    channel="broadcast",
                    timestep=0,
                    when="true",
                    probability=1.0,
                )
            ],
        ),
        interaction=InteractionConfig(
            primary_model=InteractionType.PASSIVE_OBSERVATION,
            description="Observe updates",
        ),
        spread=SpreadConfig(share_probability=0.4),
        outcomes=OutcomeConfig(
            suggested_outcomes=[
                OutcomeDefinition(
                    name="notes",
                    description="Open notes only",
                    type=OutcomeType.OPEN_ENDED,
                    required=False,
                )
            ]
        ),
        simulation=SimulationConfig(max_timesteps=2, timestep_unit=TimestepUnit.DAY),
    )


def test_action_intent_is_captured_without_pass2_outcome():
    context = _make_context()
    scenario = _make_scenario_no_pass2()
    config = SimulationRunConfig(
        scenario_path="test.yaml",
        output_dir="test/",
        max_retries=1,
    )

    async def _mock_simple_call_async(*args, **kwargs):
        return (
            {
                "reasoning": "I need to show up and speak.",
                "private_thought": "If I don't go, nothing changes.",
                "public_statement": "I should attend the meeting.",
                "reasoning_summary": "I need to attend and speak.",
                "action_intent": "Attend the next board meeting",
                "sentiment": -0.2,
                "conviction": 75,
                "will_share": True,
                "actions": [],
            },
            TokenUsage(),
        )

    with patch(
        "extropy.simulation.reasoning.simple_call_async",
        new=_mock_simple_call_async,
    ):
        response = asyncio.run(_reason_agent_two_pass_async(context, scenario, config))

    assert response is not None
    assert response.action_intent == "Attend the next board meeting"
    assert response.outcomes.get("action_intent") == "Attend the next board meeting"