diff --git a/backend/app/api/sentiment.py b/backend/app/api/sentiment.py index 99d89b4..b305610 100644 --- a/backend/app/api/sentiment.py +++ b/backend/app/api/sentiment.py @@ -263,6 +263,12 @@ def inject_event(campaign_id: str): custom_prompt=body.get("custom_prompt"), timeout=float(body.get("timeout", 60.0)), ) + # `result` carries its own success flag (e.g. the sim environment was + # unreachable) without raising — propagate it instead of always + # reporting HTTP success, which previously let the UI show "Event + # injected successfully" for injections that silently did nothing. + if not result.get("success"): + return jsonify({"success": False, "result": result, "error": result.get("error")}), 502 return jsonify({"success": True, "result": result}) except ValueError as e: return jsonify({"success": False, "error": str(e)}), 404 diff --git a/backend/app/services/campaign_manager.py b/backend/app/services/campaign_manager.py index c720ebe..d03b6d5 100644 --- a/backend/app/services/campaign_manager.py +++ b/backend/app/services/campaign_manager.py @@ -596,11 +596,17 @@ def inject_scenario_event( prompt=prompt, timeout=timeout, ) - if scenario.lower() == "b": - campaign.scenario_b_injected = True - else: - campaign.scenario_c_injected = True - self._save(campaign) + # Only mark the scenario as injected if the interview actually + # reached agents. `interview_all_agents` doesn't raise when the + # environment is unreachable — it returns success=False — so + # checking only for exceptions here previously let a silent no-op + # be recorded as a successful injection. + if result.get("success"): + if scenario.lower() == "b": + campaign.scenario_b_injected = True + else: + campaign.scenario_c_injected = True + self._save(campaign) return result except Exception as e: logger.error(f"Event injection failed for Scenario {scenario}: {e}") @@ -809,16 +815,27 @@ def _execute_pending_injections( f"Executing staggered injection tier={tier} round={current_round} " f"sim={sim_id}: {prompt[:60]}..." ) + succeeded = False try: - SimulationRunner.interview_all_agents( + result = SimulationRunner.interview_all_agents( simulation_id=sim_id, prompt=prompt, timeout=30.0, ) + succeeded = bool(result.get("success")) + if not succeeded: + logger.warning( + f"Staggered injection tier={tier} reported failure (non-fatal): " + f"{result.get('error')}" + ) except Exception as e: logger.warning(f"Staggered injection tier={tier} failed (non-fatal): {e}") + # Mark this tier attempted either way so we don't retry it forever, + # but only a *successful* tier counts toward the scenario being + # considered "injected". inj["done"] = True + inj["succeeded"] = succeeded executed_any = True executed_scenarios.add(inj.get("scenario")) @@ -826,7 +843,8 @@ def _execute_pending_injections( for scenario in executed_scenarios: if not scenario: continue - if all(i["done"] for i in campaign.pending_injections if i.get("scenario") == scenario): + scenario_injections = [i for i in campaign.pending_injections if i.get("scenario") == scenario] + if all(i["done"] for i in scenario_injections) and any(i.get("succeeded") for i in scenario_injections): if scenario == "b": campaign.scenario_b_injected = True elif scenario == "c": diff --git a/backend/scripts/run_parallel_simulation.py b/backend/scripts/run_parallel_simulation.py index 2a627ff..cddb08a 100644 --- a/backend/scripts/run_parallel_simulation.py +++ b/backend/scripts/run_parallel_simulation.py @@ -234,15 +234,39 @@ def __init__( self.twitter_agent_graph = twitter_agent_graph self.reddit_env = reddit_env self.reddit_agent_graph = reddit_agent_graph - + + # Per-platform locks. These serialize an interview-triggered env.step() + # against the round loop's own env.step() on the same env, since OASIS + # environments aren't safe to step() concurrently from two coroutines. + # Set via register_twitter()/register_reddit() once each platform's env + # is constructed (round loop and interview dispatcher share the lock). + self.twitter_lock: Optional[asyncio.Lock] = None + self.reddit_lock: Optional[asyncio.Lock] = None + self.commands_dir = os.path.join(simulation_dir, IPC_COMMANDS_DIR) self.responses_dir = os.path.join(simulation_dir, IPC_RESPONSES_DIR) self.status_file = os.path.join(simulation_dir, ENV_STATUS_FILE) - + # 确保目录存在 os.makedirs(self.commands_dir, exist_ok=True) os.makedirs(self.responses_dir, exist_ok=True) - + + def register_twitter(self, env, agent_graph, lock: asyncio.Lock): + """Attach the Twitter env/graph once constructed, so mid-run interview + commands can reach it while the round loop is still going.""" + self.twitter_env = env + self.twitter_agent_graph = agent_graph + self.twitter_lock = lock + self.update_status("alive") + + def register_reddit(self, env, agent_graph, lock: asyncio.Lock): + """Attach the Reddit env/graph once constructed, so mid-run interview + commands can reach it while the round loop is still going.""" + self.reddit_env = env + self.reddit_agent_graph = agent_graph + self.reddit_lock = lock + self.update_status("alive") + def update_status(self, status: str): """更新环境状态""" with open(self.status_file, 'w', encoding='utf-8') as f: @@ -322,10 +346,12 @@ async def _interview_single_platform(self, agent_id: int, prompt: str, platform: 包含结果的字典，或包含error的字典 """ env, agent_graph, actual_platform = self._get_env_and_graph(platform) - + if not env or not agent_graph: return {"platform": platform, "error": f"{platform}平台不可用"} - + + lock = self.twitter_lock if actual_platform == "twitter" else self.reddit_lock + try: agent = agent_graph.get_agent(agent_id) interview_action = ManualAction( @@ -333,8 +359,12 @@ async def _interview_single_platform(self, agent_id: int, prompt: str, platform: action_args={"prompt": prompt} ) actions = {agent: interview_action} - await env.step(actions) - + if lock is not None: + async with lock: + await env.step(actions) + else: + await env.step(actions) + result = self._get_interview_result(agent_id, actual_platform) result["platform"] = actual_platform return result @@ -466,8 +496,12 @@ async def handle_batch_interview(self, command_id: str, interviews: List[Dict], print(f" 警告: 无法获取Twitter Agent {agent_id}: {e}") if twitter_actions: - await self.twitter_env.step(twitter_actions) - + if self.twitter_lock is not None: + async with self.twitter_lock: + await self.twitter_env.step(twitter_actions) + else: + await self.twitter_env.step(twitter_actions) + for interview in twitter_interviews: agent_id = interview.get("agent_id") result = self._get_interview_result(agent_id, "twitter") @@ -493,8 +527,12 @@ async def handle_batch_interview(self, command_id: str, interviews: List[Dict], print(f" 警告: 无法获取Reddit Agent {agent_id}: {e}") if reddit_actions: - await self.reddit_env.step(reddit_actions) - + if self.reddit_lock is not None: + async with self.reddit_lock: + await self.reddit_env.step(reddit_actions) + else: + await self.reddit_env.step(reddit_actions) + for interview in reddit_interviews: agent_id = interview.get("agent_id") result = self._get_interview_result(agent_id, "reddit") @@ -1099,24 +1137,29 @@ def __init__(self): async def run_twitter_simulation( - config: Dict[str, Any], + config: Dict[str, Any], simulation_dir: str, action_logger: Optional[PlatformActionLogger] = None, main_logger: Optional[SimulationLogManager] = None, - max_rounds: Optional[int] = None + max_rounds: Optional[int] = None, + ipc_handler: Optional["ParallelIPCHandler"] = None, ) -> PlatformSimulation: """运行Twitter模拟 - + Args: config: 模拟配置 simulation_dir: 模拟目录 action_logger: 动作日志记录器 main_logger: 主日志管理器 max_rounds: 最大模拟轮数（可选，用于截断过长的模拟） - + ipc_handler: 可选的IPC处理器；一旦env就绪就会注册进去， + 使得Interview命令可以在轮次循环运行期间被处理，而不必等到 + 全部轮次结束 + Returns: PlatformSimulation: 包含env和agent_graph的结果对象 """ + twitter_lock = asyncio.Lock() result = PlatformSimulation() def log_info(msg): @@ -1161,21 +1204,24 @@ def log_info(msg): await result.env.reset() log_info("环境已启动") - + + if ipc_handler is not None: + ipc_handler.register_twitter(result.env, result.agent_graph, twitter_lock) + if action_logger: action_logger.log_simulation_start(config) - + total_actions = 0 last_rowid = 0 # 跟踪数据库中最后处理的行号（使用 rowid 避免 created_at 格式差异） - + # 执行初始事件 event_config = config.get("event_config", {}) initial_posts = event_config.get("initial_posts", []) - + # 记录 round 0 开始（初始事件阶段） if action_logger: action_logger.log_round_start(0, 0) # round 0, simulated_hour 0 - + initial_action_count = 0 if initial_posts: initial_actions = {} @@ -1188,7 +1234,7 @@ def log_info(msg): action_type=ActionType.CREATE_POST, action_args={"content": content} ) - + if action_logger: action_logger.log_action( round_num=0, @@ -1201,9 +1247,10 @@ def log_info(msg): initial_action_count += 1 except Exception: pass - + if initial_actions: - await result.env.step(initial_actions) + async with twitter_lock: + await result.env.step(initial_actions) log_info(f"已发布 {len(initial_actions)} 条初始帖子") # 记录 round 0 结束 @@ -1251,13 +1298,14 @@ def log_info(msg): continue actions = {agent: LLMAction() for _, agent in active_agents} - await result.env.step(actions) - + async with twitter_lock: + await result.env.step(actions) + # 从数据库获取实际执行的动作并记录 actual_actions, last_rowid = fetch_new_actions_from_db( db_path, last_rowid, agent_names ) - + round_action_count = 0 for action_data in actual_actions: if action_logger: @@ -1270,45 +1318,50 @@ def log_info(msg): ) total_actions += 1 round_action_count += 1 - + if action_logger: action_logger.log_round_end(round_num + 1, round_action_count) - + if (round_num + 1) % 20 == 0: progress = (round_num + 1) / total_rounds * 100 log_info(f"Day {simulated_day}, {simulated_hour:02d}:00 - Round {round_num + 1}/{total_rounds} ({progress:.1f}%)") - + # 注意：不关闭环境，保留给Interview使用 - + if action_logger: action_logger.log_simulation_end(total_rounds, total_actions) - + result.total_actions = total_actions elapsed = (datetime.now() - start_time).total_seconds() log_info(f"模拟循环完成! 耗时: {elapsed:.1f}秒, 总动作: {total_actions}") - + return result async def run_reddit_simulation( - config: Dict[str, Any], + config: Dict[str, Any], simulation_dir: str, action_logger: Optional[PlatformActionLogger] = None, main_logger: Optional[SimulationLogManager] = None, - max_rounds: Optional[int] = None + max_rounds: Optional[int] = None, + ipc_handler: Optional["ParallelIPCHandler"] = None, ) -> PlatformSimulation: """运行Reddit模拟 - + Args: config: 模拟配置 simulation_dir: 模拟目录 action_logger: 动作日志记录器 main_logger: 主日志管理器 max_rounds: 最大模拟轮数（可选，用于截断过长的模拟） - + ipc_handler: 可选的IPC处理器；一旦env就绪就会注册进去， + 使得Interview命令可以在轮次循环运行期间被处理，而不必等到 + 全部轮次结束 + Returns: PlatformSimulation: 包含env和agent_graph的结果对象 """ + reddit_lock = asyncio.Lock() result = PlatformSimulation() def log_info(msg): @@ -1352,21 +1405,24 @@ def log_info(msg): await result.env.reset() log_info("环境已启动") - + + if ipc_handler is not None: + ipc_handler.register_reddit(result.env, result.agent_graph, reddit_lock) + if action_logger: action_logger.log_simulation_start(config) - + total_actions = 0 last_rowid = 0 # 跟踪数据库中最后处理的行号（使用 rowid 避免 created_at 格式差异） - + # 执行初始事件 event_config = config.get("event_config", {}) initial_posts = event_config.get("initial_posts", []) - + # 记录 round 0 开始（初始事件阶段） if action_logger: action_logger.log_round_start(0, 0) # round 0, simulated_hour 0 - + initial_action_count = 0 if initial_posts: initial_actions = {} @@ -1402,61 +1458,63 @@ def log_info(msg): pass if initial_actions: - await result.env.step(initial_actions) + async with reddit_lock: + await result.env.step(initial_actions) log_info(f"已发布 {len(initial_actions)} 条初始帖子") - + # 记录 round 0 结束 if action_logger: action_logger.log_round_end(0, initial_action_count) - + # 主模拟循环 time_config = config.get("time_config", {}) total_hours = time_config.get("total_simulation_hours", 72) minutes_per_round = time_config.get("minutes_per_round", 30) total_rounds = (total_hours * 60) // minutes_per_round - + # 如果指定了最大轮数，则截断 if max_rounds is not None and max_rounds > 0: original_rounds = total_rounds total_rounds = min(total_rounds, max_rounds) if total_rounds < original_rounds: log_info(f"轮数已截断: {original_rounds} -> {total_rounds} (max_rounds={max_rounds})") - + start_time = datetime.now() - + for round_num in range(total_rounds): # 检查是否收到退出信号 if _shutdown_event and _shutdown_event.is_set(): if main_logger: main_logger.info(f"收到退出信号，在第 {round_num + 1} 轮停止模拟") break - + simulated_minutes = round_num * minutes_per_round simulated_hour = (simulated_minutes // 60) % 24 simulated_day = simulated_minutes // (60 * 24) + 1 - + active_agents = get_active_agents_for_round( result.env, config, simulated_hour, round_num ) - + # 无论是否有活跃agent，都记录round开始 if action_logger: action_logger.log_round_start(round_num + 1, simulated_hour) - + if not active_agents: # 没有活跃agent时也记录round结束（actions_count=0） if action_logger: action_logger.log_round_end(round_num + 1, 0) continue - + actions = {agent: LLMAction() for _, agent in active_agents} - await result.env.step(actions) - + async with reddit_lock: + await result.env.step(actions) + # 从数据库获取实际执行的动作并记录 actual_actions, last_rowid = fetch_new_actions_from_db( db_path, last_rowid, agent_names ) - + round_action_count = 0 for action_data in actual_actions: if action_logger: @@ -1469,16 +1527,16 @@ def log_info(msg): ) total_actions += 1 round_action_count += 1 - + if action_logger: action_logger.log_round_end(round_num + 1, round_action_count) - + if (round_num + 1) % 20 == 0: progress = (round_num + 1) / total_rounds * 100 log_info(f"Day {simulated_day}, {simulated_hour:02d}:00 - Round {round_num + 1}/{total_rounds} ({progress:.1f}%)") - + # 注意：不关闭环境，保留给Interview使用 - + if action_logger: action_logger.log_simulation_end(total_rounds, total_actions) @@ -1571,67 +1629,85 @@ async def main(): log_manager.info("=" * 60) start_time = datetime.now() - + # 存储两个平台的模拟结果 twitter_result: Optional[PlatformSimulation] = None reddit_result: Optional[PlatformSimulation] = None - + + # 如果启用等待命令模式，在轮次循环开始*之前*就创建IPC处理器并启动一个 + # 独立的调度协程持续轮询命令。这样Interview/close_env命令在模拟运行期间 + # 就能被处理，而不必等到所有轮次跑完（此前的实现只在跑完之后才服务命令， + # 导致运行期间的事件注入永远失败: "environment is not running or has closed"）。 + # 各平台的env在construct后通过 register_twitter()/register_reddit() 挂载进 + # 同一个handler；handler和每个平台round loop共享一把锁，避免round loop + # 自身的env.step()与interview触发的env.step()在同一个env上并发执行。 + ipc_handler: Optional[ParallelIPCHandler] = None + dispatcher_task: Optional[asyncio.Task] = None + + if wait_for_commands: + ipc_handler = ParallelIPCHandler(simulation_dir=simulation_dir) + ipc_handler.update_status("alive") + + async def _dispatch_commands(): + try: + while not _shutdown_event.is_set(): + should_continue = await ipc_handler.process_commands() + if not should_continue: + break + try: + await asyncio.wait_for(_shutdown_event.wait(), timeout=0.5) + break # 收到退出信号 + except asyncio.TimeoutError: + pass # 超时继续循环 + except asyncio.CancelledError: + raise + except Exception as e: + print(f"\n命令处理出错: {e}") + + log_manager.info("IPC 命令调度已启动 - 支持在模拟运行期间注入事件 (interview/batch_interview/close_env)") + dispatcher_task = asyncio.create_task(_dispatch_commands()) + if args.twitter_only: - twitter_result = await run_twitter_simulation(config, simulation_dir, twitter_logger, log_manager, args.max_rounds) + twitter_result = await run_twitter_simulation( + config, simulation_dir, twitter_logger, log_manager, args.max_rounds, ipc_handler=ipc_handler + ) elif args.reddit_only: - reddit_result = await run_reddit_simulation(config, simulation_dir, reddit_logger, log_manager, args.max_rounds) + reddit_result = await run_reddit_simulation( + config, simulation_dir, reddit_logger, log_manager, args.max_rounds, ipc_handler=ipc_handler + ) else: # 并行运行（每个平台使用独立的日志记录器） results = await asyncio.gather( - run_twitter_simulation(config, simulation_dir, twitter_logger, log_manager, args.max_rounds), - run_reddit_simulation(config, simulation_dir, reddit_logger, log_manager, args.max_rounds), + run_twitter_simulation( + config, simulation_dir, twitter_logger, log_manager, args.max_rounds, ipc_handler=ipc_handler + ), + run_reddit_simulation( + config, simulation_dir, reddit_logger, log_manager, args.max_rounds, ipc_handler=ipc_handler + ), ) twitter_result, reddit_result = results - + total_elapsed = (datetime.now() - start_time).total_seconds() log_manager.info("=" * 60) log_manager.info(f"模拟循环完成! 总耗时: {total_elapsed:.1f}秒") - - # 是否进入等待命令模式 - if wait_for_commands: + + # 轮次循环已经跑完；调度协程从一开始就在运行，此时仍继续等待 + # Interview/close_env命令，直到收到 close_env 或 shutdown 信号 + if dispatcher_task is not None: log_manager.info("") log_manager.info("=" * 60) log_manager.info("进入等待命令模式 - 环境保持运行") log_manager.info("支持的命令: interview, batch_interview, close_env") log_manager.info("=" * 60) - - # 创建IPC处理器 - ipc_handler = ParallelIPCHandler( - simulation_dir=simulation_dir, - twitter_env=twitter_result.env if twitter_result else None, - twitter_agent_graph=twitter_result.agent_graph if twitter_result else None, - reddit_env=reddit_result.env if reddit_result else None, - reddit_agent_graph=reddit_result.agent_graph if reddit_result else None - ) - ipc_handler.update_status("alive") - - # 等待命令循环（使用全局 _shutdown_event） + try: - while not _shutdown_event.is_set(): - should_continue = await ipc_handler.process_commands() - if not should_continue: - break - # 使用 wait_for 替代 sleep，这样可以响应 shutdown_event - try: - await asyncio.wait_for(_shutdown_event.wait(), timeout=0.5) - break # 收到退出信号 - except asyncio.TimeoutError: - pass # 超时继续循环 - except KeyboardInterrupt: - print("\n收到中断信号") + await dispatcher_task except asyncio.CancelledError: - print("\n任务被取消") - except Exception as e: - print(f"\n命令处理出错: {e}") - + pass + log_manager.info("\n关闭环境...") ipc_handler.update_status("stopped") - + # 关闭环境 if twitter_result and twitter_result.env: await twitter_result.env.close() diff --git a/backend/tests/test_campaign_injection_success.py b/backend/tests/test_campaign_injection_success.py new file mode 100644 index 0000000..74f6d6b --- /dev/null +++ b/backend/tests/test_campaign_injection_success.py @@ -0,0 +1,111 @@ +"""Tests for issue #29 (A2): event injection must not be reported as +successful when the underlying interview call didn't actually reach agents. + +Covers both the manual injection path (inject_scenario_event) and the +auto/staggered injection path (_execute_pending_injections). +""" + +from unittest.mock import MagicMock + +from app.services import campaign_manager as cm_module +from app.services.campaign_manager import CampaignManager, CampaignState + + +def _state(**overrides): + base = dict( + campaign_id="c1", project_id="p1", graph_id="g1", + seed_data={"product_name": "Widget"}, sim_id_b="sim_b", sim_id_c="sim_c", + ) + base.update(overrides) + return CampaignState(**base) + + +def _manager(state): + mgr = CampaignManager.__new__(CampaignManager) + mgr._save = MagicMock() + mgr._load = MagicMock(return_value=state) + return mgr + + +# --- inject_scenario_event (manual injection) -------------------------------- + +def test_inject_scenario_event_marks_injected_on_success(monkeypatch): + state = _state() + mgr = _manager(state) + monkeypatch.setattr( + cm_module.SimulationRunner, "interview_all_agents", + MagicMock(return_value={"success": True, "interviews_count": 5}), + ) + + result = mgr.inject_scenario_event(campaign_id="c1", scenario="b", custom_prompt="hi") + + assert result["success"] is True + assert state.scenario_b_injected is True + mgr._save.assert_called_once() + + +def test_inject_scenario_event_does_not_mark_injected_on_reported_failure(monkeypatch): + state = _state() + mgr = _manager(state) + monkeypatch.setattr( + cm_module.SimulationRunner, "interview_all_agents", + MagicMock(return_value={"success": False, "error": "env not running"}), + ) + + result = mgr.inject_scenario_event(campaign_id="c1", scenario="b", custom_prompt="hi") + + assert result["success"] is False + assert state.scenario_b_injected is False + mgr._save.assert_not_called() + + +# --- _execute_pending_injections (auto/staggered injection) ------------------ + +def _pending(scenario="b", sim_id="sim_b", round_num=7): + return [{ + "round": round_num, "sim_id": sim_id, "scenario": scenario, + "prompt": "p1", "tier": "high", "done": False, + }] + + +def test_execute_pending_injections_marks_scenario_injected_on_success(monkeypatch): + state = _state(pending_injections=_pending()) + mgr = _manager(state) + monkeypatch.setattr( + cm_module.SimulationRunner, "interview_all_agents", + MagicMock(return_value={"success": True, "interviews_count": 5}), + ) + + mgr._execute_pending_injections(campaign=state, sim_id="sim_b", current_round=7) + + assert state.pending_injections[0]["done"] is True + assert state.scenario_b_injected is True + + +def test_execute_pending_injections_does_not_mark_injected_when_reported_failure(monkeypatch): + state = _state(pending_injections=_pending()) + mgr = _manager(state) + monkeypatch.setattr( + cm_module.SimulationRunner, "interview_all_agents", + MagicMock(return_value={"success": False, "error": "env not running"}), + ) + + mgr._execute_pending_injections(campaign=state, sim_id="sim_b", current_round=7) + + assert state.pending_injections[0]["done"] is True + assert state.scenario_b_injected is False + + +def test_execute_pending_injections_does_not_mark_injected_on_exception(monkeypatch): + state = _state(pending_injections=_pending()) + mgr = _manager(state) + + def _raise(*args, **kwargs): + raise ValueError("boom") + + monkeypatch.setattr(cm_module.SimulationRunner, "interview_all_agents", _raise) + + mgr._execute_pending_injections(campaign=state, sim_id="sim_b", current_round=7) + + assert state.pending_injections[0]["done"] is True + assert state.scenario_b_injected is False diff --git a/backend/tests/test_parallel_simulation_ipc.py b/backend/tests/test_parallel_simulation_ipc.py new file mode 100644 index 0000000..a08a56b --- /dev/null +++ b/backend/tests/test_parallel_simulation_ipc.py @@ -0,0 +1,144 @@ +"""Tests for issue #29 (A1): event injection must actually reach a running +simulation instead of silently failing with "environment is not running or +has closed". + +`run_parallel_simulation.py` is a standalone script (not part of the `app` +package), so it's loaded directly from its file path. These tests exercise +`ParallelIPCHandler` in isolation with fake envs — no real OASIS/LLM calls — +to verify: + 1. Registering a platform's env marks the environment "alive" immediately, + instead of only after the full round loop finishes. + 2. An interview-triggered env.step() and the round loop's own env.step() + never run concurrently on the same env (they'd otherwise race on OASIS's + internal state). +""" + +import asyncio +import importlib.util +import json +import os + +import pytest + +SCRIPTS_DIR = os.path.join(os.path.dirname(__file__), "..", "scripts") + + +def _load_module(): + spec = importlib.util.spec_from_file_location( + "run_parallel_simulation", os.path.join(SCRIPTS_DIR, "run_parallel_simulation.py") + ) + mod = importlib.util.module_from_spec(spec) + spec.loader.exec_module(mod) + return mod + + +@pytest.fixture(scope="module") +def rps(): + return _load_module() + + +class FakeEnv: + """Stand-in for an OASIS env: asserts the shared lock is held during + step(), so a lock-free regression fails loudly instead of racing.""" + + def __init__(self, lock): + self.lock = lock + self.step_calls = 0 + + async def step(self, actions): + assert self.lock.locked(), "env.step() called without holding the platform lock" + self.step_calls += 1 + await asyncio.sleep(0) # yield control, like a real awaited call would + + +class FakeAgentGraph: + def get_agent(self, agent_id): + return f"agent_{agent_id}" + + +def test_register_twitter_marks_env_alive_immediately(tmp_path, rps): + handler = rps.ParallelIPCHandler(simulation_dir=str(tmp_path)) + lock = asyncio.Lock() + + handler.register_twitter(env="env", agent_graph="graph", lock=lock) + + assert handler.twitter_env == "env" + assert handler.twitter_lock is lock + status_file = os.path.join(str(tmp_path), "env_status.json") + with open(status_file) as f: + status = json.load(f) + assert status["status"] == "alive" + assert status["twitter_available"] is True + + +def test_register_reddit_marks_env_alive_immediately(tmp_path, rps): + handler = rps.ParallelIPCHandler(simulation_dir=str(tmp_path)) + lock = asyncio.Lock() + + handler.register_reddit(env="env", agent_graph="graph", lock=lock) + + assert handler.reddit_env == "env" + assert handler.reddit_lock is lock + status_file = os.path.join(str(tmp_path), "env_status.json") + with open(status_file) as f: + status = json.load(f) + assert status["status"] == "alive" + assert status["reddit_available"] is True + + +def test_interview_step_and_round_step_never_overlap(tmp_path, rps, monkeypatch): + """The interview dispatcher and the round loop both call env.step() on the + same env from independent coroutines. Without the shared lock they could + interleave into the same OASIS env; with it, FakeEnv.step()'s assertion + that the lock is held would fail if they ever raced.""" + + async def scenario(): + handler = rps.ParallelIPCHandler(simulation_dir=str(tmp_path)) + lock = asyncio.Lock() + env = FakeEnv(lock) + graph = FakeAgentGraph() + handler.register_twitter(env, graph, lock) + monkeypatch.setattr( + handler, "_get_interview_result", lambda agent_id, platform: {"agent_id": agent_id} + ) + + async def round_loop_step(): + async with lock: + await env.step({"round": "action"}) + + interview = handler._interview_single_platform(agent_id=1, prompt="hi", platform="twitter") + round_step = round_loop_step() + + results = await asyncio.gather(interview, round_step) + assert env.step_calls == 2 + assert "error" not in results[0] + + asyncio.run(scenario()) + + +def test_batch_interview_acquires_lock_before_stepping(tmp_path, rps, monkeypatch): + async def scenario(): + handler = rps.ParallelIPCHandler(simulation_dir=str(tmp_path)) + lock = asyncio.Lock() + env = FakeEnv(lock) + graph = FakeAgentGraph() + handler.register_twitter(env, graph, lock) + monkeypatch.setattr( + handler, "_get_interview_result", lambda agent_id, platform: {"agent_id": agent_id} + ) + monkeypatch.setattr(handler, "send_response", lambda *a, **k: None) + + async def round_loop_step(): + async with lock: + await env.step({"round": "action"}) + + batch = handler.handle_batch_interview( + command_id="cmd1", + interviews=[{"agent_id": 1, "prompt": "hi", "platform": "twitter"}], + ) + round_step = round_loop_step() + + await asyncio.gather(batch, round_step) + assert env.step_calls == 2 + + asyncio.run(scenario()) diff --git a/backend/tests/test_sentiment_routes.py b/backend/tests/test_sentiment_routes.py index b401113..2387db7 100644 --- a/backend/tests/test_sentiment_routes.py +++ b/backend/tests/test_sentiment_routes.py @@ -19,6 +19,9 @@ def to_dict(self): class FakeCampaignManager: def __init__(self): self.campaigns = {} + # Test hook: when set, inject_scenario_event returns this instead of a + # synthesized success — used to exercise the failed-injection path. + self.next_inject_result = None def create_campaign(self, project_id, graph_id, seed_data, **kwargs): campaign_id = f'camp_{len(self.campaigns) + 1}' @@ -111,11 +114,13 @@ def list_campaigns(self): def inject_scenario_event(self, campaign_id, scenario, **kwargs): payload = self.campaigns[campaign_id] + if self.next_inject_result is not None: + return self.next_inject_result if scenario == 'b': payload['scenario_b_injected'] = True if scenario == 'c': payload['scenario_c_injected'] = True - return {'scenario': scenario, 'status': 'injected'} + return {'success': True, 'scenario': scenario, 'status': 'injected'} class FakeSentimentAnalyzer: @@ -227,6 +232,52 @@ def test_campaign_prepare_poll_start_and_sentiment_routes(client): assert compare_response.get_json()['comparison']['rounds'] == [1, 2] +def test_inject_event_route_reports_success(client): + from app.api import sentiment as sentiment_api + + create_response = client.post( + '/api/sentiment/campaign/create', + json={'project_id': 'proj_1', 'graph_id': 'graph_1', 'seed_data': _valid_seed()}, + ) + campaign_id = create_response.get_json()['campaign']['campaign_id'] + + inject_response = client.post( + f'/api/sentiment/campaign/{campaign_id}/inject', + json={'scenario': 'b', 'custom_prompt': 'Breaking news!'}, + ) + assert inject_response.status_code == 200 + payload = inject_response.get_json() + assert payload['success'] is True + assert payload['result']['success'] is True + + +def test_inject_event_route_propagates_failure(client): + """A no-op injection (e.g. the sim env wasn't reachable) must not be + reported as an HTTP success — this previously let the UI show "Event + injected successfully" for injections that silently did nothing.""" + from app.api import sentiment as sentiment_api + + create_response = client.post( + '/api/sentiment/campaign/create', + json={'project_id': 'proj_1', 'graph_id': 'graph_1', 'seed_data': _valid_seed()}, + ) + campaign_id = create_response.get_json()['campaign']['campaign_id'] + + sentiment_api._campaign_manager.next_inject_result = { + 'success': False, + 'error': 'Simulation environment is not running or has closed, cannot perform Interview', + } + + inject_response = client.post( + f'/api/sentiment/campaign/{campaign_id}/inject', + json={'scenario': 'b', 'custom_prompt': 'Breaking news!'}, + ) + assert inject_response.status_code == 502 + payload = inject_response.get_json() + assert payload['success'] is False + assert 'not running' in payload['error'] + + def test_report_status_route_accepts_get_query(client): task_id = TaskManager().create_task('report_generate', {'simulation_id': 'sim_1'}) TaskManager().update_task(