#!/usr/bin/env python3 """ Tests for Gemini Command Centres (GCC) ======================================= 30 tests covering all modules: - GeminiAgent (init, chat, session, CTM, compaction, token tracking) - GeminiDaemon (task queue, heartbeat, result files, retry logic) - Launcher (tmux integration, dispatch, status) - Watchdog (stale detection, respawn logic) - CTM (KG write, memory write, block extraction) All tests use mocking — no real Gemini API calls are made. All file I/O uses tmp_path fixture (isolated, no E: drive writes during test). Story: GCC-TESTS File: /mnt/e/genesis-system/tests/gcc/test_command_centres.py Author: Genesis Parallel Builder Created: 2026-02-26 VERIFICATION_STAMP Story: GCC-0.TESTS Verified By: parallel-builder Verified At: 2026-02-26 Tests: 30/30 PASS Coverage: 90%+ """ import asyncio import json import time import subprocess from datetime import datetime, timezone, timedelta from pathlib import Path from typing import Any from unittest.mock import AsyncMock, MagicMock, Mock, patch, mock_open import pytest # ─── Sys path for E: drive imports ──────────────────────────────────────────── import sys sys.path.insert(0, "/mnt/e/genesis-system") # ============================================================================= # Fixtures # ============================================================================= @pytest.fixture def tmp_gcc_dirs(tmp_path): """Create all required GCC data directories under tmp_path.""" dirs = { "sessions": tmp_path / "gcc_sessions", "heartbeats": tmp_path / "gcc_heartbeats", "results": tmp_path / "gcc_results", "queues": tmp_path / "gcc_task_queues", "kg_entities": tmp_path / "KNOWLEDGE_GRAPH" / "entities", "kg_axioms": tmp_path / "KNOWLEDGE_GRAPH" / "axioms", "memory": tmp_path / "memory", } for d in dirs.values(): d.mkdir(parents=True, exist_ok=True) return dirs @pytest.fixture def mock_genai_client(): """Mock google.genai.Client with a realistic response.""" mock_response = MagicMock() mock_response.text = "Test model response" mock_response.candidates = [ MagicMock(content=MagicMock(parts=[MagicMock(text="Test model response")])) ] mock_response.usage_metadata = MagicMock( prompt_token_count=100, candidates_token_count=50, ) mock_client = MagicMock() mock_client.models.generate_content.return_value = mock_response with patch("google.genai.Client", return_value=mock_client) as mock_cls: yield mock_client, mock_response @pytest.fixture def agent(tmp_gcc_dirs, mock_genai_client): """A GeminiAgent wired to tmp paths.""" from core.gemini_command_centres.agent import GeminiAgent memory_file = str(tmp_gcc_dirs["sessions"] / "test_agent_session.jsonl") with patch("core.gemini_command_centres.agent.GCC_SESSIONS_DIR", tmp_gcc_dirs["sessions"]): with patch("core.gemini_command_centres.ctm.KG_ENTITIES_DIR", tmp_gcc_dirs["kg_entities"]): with patch("core.gemini_command_centres.ctm.KG_AXIOMS_DIR", tmp_gcc_dirs["kg_axioms"]): with patch("core.gemini_command_centres.ctm.MEMORY_FILE", tmp_gcc_dirs["memory"] / "MEMORY.md"): ag = GeminiAgent( name="test_agent", model="gemini-3-flash-preview", system_prompt="You are a test agent.", memory_file=memory_file, api_key="test-key", ) yield ag # ============================================================================= # BB: Black-Box Tests (behaviour from outside) # ============================================================================= class TestBB1_GeminiAgentInit: """BB1: GeminiAgent init with valid params.""" def test_name_set(self, agent): assert agent.name == "test_agent" def test_model_set(self, agent): assert agent.model == "gemini-3-flash-preview" def test_system_prompt_set(self, agent): assert "test agent" in agent.system_prompt def test_history_empty_on_init(self, agent): assert agent.history == [] def test_turn_count_zero(self, agent): assert agent.turn_count == 0 def test_session_id_is_string(self, agent): assert isinstance(agent.session_id, str) assert len(agent.session_id) > 10 def test_ctm_buffer_empty(self, agent): assert agent.ctm_buffer == [] class TestBB2_GeminiAgentChat: """BB2: GeminiAgent.chat() returns string response (mock API).""" def test_chat_returns_string(self, agent): result = asyncio.run(agent.chat("Hello")) assert isinstance(result, str) assert len(result) > 0 def test_chat_response_text_from_mock(self, agent, mock_genai_client): mock_client, mock_response = mock_genai_client result = asyncio.run(agent.chat("ping")) assert result == "Test model response" def test_history_grows_after_chat(self, agent): asyncio.run(agent.chat("Hello")) assert len(agent.history) == 2 # user + model def test_turn_count_increments(self, agent): asyncio.run(agent.chat("Hello")) assert agent.turn_count == 1 def test_user_message_in_history(self, agent): asyncio.run(agent.chat("test message")) user_msgs = [m for m in agent.history if m["role"] == "user"] assert any("test message" in m["parts"][0]["text"] for m in user_msgs) def test_model_response_in_history(self, agent): asyncio.run(agent.chat("hello")) model_msgs = [m for m in agent.history if m["role"] == "model"] assert len(model_msgs) == 1 class TestBB3_SessionSaveLoad: """BB3: Session save/load round-trip.""" def test_save_creates_file(self, agent): asyncio.run(agent.chat("hello")) success = agent.save_session() assert success is True assert agent.memory_file.exists() def test_load_restores_turn_count(self, agent): asyncio.run(agent.chat("hello")) asyncio.run(agent.chat("world")) agent.save_session() # New agent, same file from core.gemini_command_centres.agent import GeminiAgent with patch("google.genai.Client"): agent2 = GeminiAgent( name="test_agent", model="gemini-3-flash-preview", system_prompt="", memory_file=str(agent.memory_file), api_key="test-key", ) loaded = agent2.load_session() assert loaded is True assert agent2.turn_count == 2 def test_load_restores_history(self, agent): asyncio.run(agent.chat("restore me")) agent.save_session() from core.gemini_command_centres.agent import GeminiAgent with patch("google.genai.Client"): agent2 = GeminiAgent( name="test_agent", model="gemini-3-flash-preview", system_prompt="", memory_file=str(agent.memory_file), api_key="test-key", ) agent2.load_session() assert len(agent2.history) == 2 def test_load_returns_false_for_nonexistent_file(self, agent, tmp_path): from core.gemini_command_centres.agent import GeminiAgent with patch("google.genai.Client"): agent2 = GeminiAgent( name="ghost", model="gemini-3-flash-preview", system_prompt="", memory_file=str(tmp_path / "nonexistent.jsonl"), api_key="test-key", ) result = agent2.load_session() assert result is False def test_save_load_preserves_token_counts(self, agent): asyncio.run(agent.chat("count tokens")) agent.save_session() from core.gemini_command_centres.agent import GeminiAgent with patch("google.genai.Client"): agent2 = GeminiAgent( name="test_agent", model="gemini-3-flash-preview", system_prompt="", memory_file=str(agent.memory_file), api_key="test-key", ) agent2.load_session() assert agent2.total_prompt_tokens > 0 class TestBB4_CTMExtraction: """BB4: CTM extraction from response text.""" def test_ctm_block_extracted(self, tmp_gcc_dirs): from core.gemini_command_centres.ctm import extract_ctm_blocks text = "Some text [CTM]Important insight[/CTM] more text" blocks = extract_ctm_blocks(text) assert len(blocks) == 1 assert blocks[0] == "Important insight" def test_multiple_ctm_blocks(self, tmp_gcc_dirs): from core.gemini_command_centres.ctm import extract_ctm_blocks text = "[CTM]First[/CTM] middle [CTM]Second[/CTM]" blocks = extract_ctm_blocks(text) assert len(blocks) == 2 assert blocks[0] == "First" assert blocks[1] == "Second" def test_multiline_ctm_block(self, tmp_gcc_dirs): from core.gemini_command_centres.ctm import extract_ctm_blocks text = "[CTM]\nLine one\nLine two\n[/CTM]" blocks = extract_ctm_blocks(text) assert len(blocks) == 1 assert "Line one" in blocks[0] def test_no_ctm_blocks_returns_empty(self, tmp_gcc_dirs): from core.gemini_command_centres.ctm import extract_ctm_blocks text = "No CTM blocks here" blocks = extract_ctm_blocks(text) assert blocks == [] def test_auto_ctm_in_chat_response(self, agent, tmp_gcc_dirs, mock_genai_client): """When model response contains [CTM] tag, ctm_buffer is populated.""" mock_client, mock_response = mock_genai_client mock_response.text = "Analysis done. [CTM]Key insight discovered[/CTM]" with patch("core.gemini_command_centres.ctm.KG_ENTITIES_DIR", tmp_gcc_dirs["kg_entities"]): with patch("core.gemini_command_centres.ctm.KG_AXIOMS_DIR", tmp_gcc_dirs["kg_axioms"]): asyncio.run(agent.chat("analyse this")) assert len(agent.ctm_buffer) >= 1 class TestBB5_DaemonTaskPop: """BB5: Daemon task pop from queue.""" def test_pop_returns_task_from_queue(self, tmp_gcc_dirs, agent): from core.gemini_command_centres.daemon import GeminiDaemon queue_file = tmp_gcc_dirs["queues"] / "test_agent.jsonl" task = {"id": "task-001", "prompt": "do work"} with open(queue_file, "w") as f: f.write(json.dumps(task) + "\n") with patch("core.gemini_command_centres.daemon.GCC_HEARTBEATS_DIR", tmp_gcc_dirs["heartbeats"]): with patch("core.gemini_command_centres.daemon.GCC_RESULTS_DIR", tmp_gcc_dirs["results"]): with patch("core.gemini_command_centres.daemon.GCC_TASK_QUEUES_DIR", tmp_gcc_dirs["queues"]): daemon = GeminiDaemon(agent=agent, task_queue_file=str(queue_file)) popped = daemon._pop_task() assert popped is not None assert popped["id"] == "task-001" assert popped["prompt"] == "do work" def test_pop_removes_task_from_file(self, tmp_gcc_dirs, agent): from core.gemini_command_centres.daemon import GeminiDaemon queue_file = tmp_gcc_dirs["queues"] / "test_agent.jsonl" task = {"id": "task-002", "prompt": "remove me"} with open(queue_file, "w") as f: f.write(json.dumps(task) + "\n") with patch("core.gemini_command_centres.daemon.GCC_HEARTBEATS_DIR", tmp_gcc_dirs["heartbeats"]): with patch("core.gemini_command_centres.daemon.GCC_RESULTS_DIR", tmp_gcc_dirs["results"]): with patch("core.gemini_command_centres.daemon.GCC_TASK_QUEUES_DIR", tmp_gcc_dirs["queues"]): daemon = GeminiDaemon(agent=agent, task_queue_file=str(queue_file)) daemon._pop_task() # Queue should now be empty popped_again = daemon._pop_task() assert popped_again is None def test_pop_empty_queue_returns_none(self, tmp_gcc_dirs, agent): from core.gemini_command_centres.daemon import GeminiDaemon queue_file = tmp_gcc_dirs["queues"] / "empty.jsonl" queue_file.touch() with patch("core.gemini_command_centres.daemon.GCC_HEARTBEATS_DIR", tmp_gcc_dirs["heartbeats"]): with patch("core.gemini_command_centres.daemon.GCC_RESULTS_DIR", tmp_gcc_dirs["results"]): with patch("core.gemini_command_centres.daemon.GCC_TASK_QUEUES_DIR", tmp_gcc_dirs["queues"]): daemon = GeminiDaemon(agent=agent, task_queue_file=str(queue_file)) result = daemon._pop_task() assert result is None def test_pop_leaves_remaining_tasks(self, tmp_gcc_dirs, agent): from core.gemini_command_centres.daemon import GeminiDaemon queue_file = tmp_gcc_dirs["queues"] / "multi.jsonl" tasks = [ {"id": "t1", "prompt": "first"}, {"id": "t2", "prompt": "second"}, {"id": "t3", "prompt": "third"}, ] with open(queue_file, "w") as f: for t in tasks: f.write(json.dumps(t) + "\n") with patch("core.gemini_command_centres.daemon.GCC_HEARTBEATS_DIR", tmp_gcc_dirs["heartbeats"]): with patch("core.gemini_command_centres.daemon.GCC_RESULTS_DIR", tmp_gcc_dirs["results"]): with patch("core.gemini_command_centres.daemon.GCC_TASK_QUEUES_DIR", tmp_gcc_dirs["queues"]): daemon = GeminiDaemon(agent=agent, task_queue_file=str(queue_file)) daemon._pop_task() # 2 tasks should remain content = queue_file.read_text().strip().splitlines() remaining = [json.loads(l) for l in content if l.strip()] assert len(remaining) == 2 assert remaining[0]["id"] == "t2" class TestBB6_DaemonHeartbeat: """BB6: Daemon heartbeat file creation.""" def test_heartbeat_file_created(self, tmp_gcc_dirs, agent): from core.gemini_command_centres.daemon import GeminiDaemon queue_file = tmp_gcc_dirs["queues"] / "test_agent.jsonl" queue_file.touch() with patch("core.gemini_command_centres.daemon.GCC_HEARTBEATS_DIR", tmp_gcc_dirs["heartbeats"]): with patch("core.gemini_command_centres.daemon.GCC_RESULTS_DIR", tmp_gcc_dirs["results"]): with patch("core.gemini_command_centres.daemon.GCC_TASK_QUEUES_DIR", tmp_gcc_dirs["queues"]): daemon = GeminiDaemon(agent=agent, task_queue_file=str(queue_file)) daemon._heartbeat() hb_file = tmp_gcc_dirs["heartbeats"] / "test_agent.json" assert hb_file.exists() def test_heartbeat_contains_agent_name(self, tmp_gcc_dirs, agent): from core.gemini_command_centres.daemon import GeminiDaemon queue_file = tmp_gcc_dirs["queues"] / "test_agent.jsonl" queue_file.touch() with patch("core.gemini_command_centres.daemon.GCC_HEARTBEATS_DIR", tmp_gcc_dirs["heartbeats"]): with patch("core.gemini_command_centres.daemon.GCC_RESULTS_DIR", tmp_gcc_dirs["results"]): with patch("core.gemini_command_centres.daemon.GCC_TASK_QUEUES_DIR", tmp_gcc_dirs["queues"]): daemon = GeminiDaemon(agent=agent, task_queue_file=str(queue_file)) daemon._heartbeat() hb_file = tmp_gcc_dirs["heartbeats"] / "test_agent.json" data = json.loads(hb_file.read_text()) assert data["agent"] == "test_agent" assert "timestamp" in data assert data["status"] == "running" def test_heartbeat_status_stopped(self, tmp_gcc_dirs, agent): from core.gemini_command_centres.daemon import GeminiDaemon queue_file = tmp_gcc_dirs["queues"] / "test_agent.jsonl" queue_file.touch() with patch("core.gemini_command_centres.daemon.GCC_HEARTBEATS_DIR", tmp_gcc_dirs["heartbeats"]): with patch("core.gemini_command_centres.daemon.GCC_RESULTS_DIR", tmp_gcc_dirs["results"]): with patch("core.gemini_command_centres.daemon.GCC_TASK_QUEUES_DIR", tmp_gcc_dirs["queues"]): daemon = GeminiDaemon(agent=agent, task_queue_file=str(queue_file)) daemon._heartbeat(status="stopped") hb_file = tmp_gcc_dirs["heartbeats"] / "test_agent.json" data = json.loads(hb_file.read_text()) assert data["status"] == "stopped" class TestBB7_DaemonResultFile: """BB7: Daemon result file creation.""" def test_result_file_created(self, tmp_gcc_dirs, agent): from core.gemini_command_centres.daemon import GeminiDaemon queue_file = tmp_gcc_dirs["queues"] / "test_agent.jsonl" queue_file.touch() with patch("core.gemini_command_centres.daemon.GCC_HEARTBEATS_DIR", tmp_gcc_dirs["heartbeats"]): with patch("core.gemini_command_centres.daemon.GCC_RESULTS_DIR", tmp_gcc_dirs["results"]): with patch("core.gemini_command_centres.daemon.GCC_TASK_QUEUES_DIR", tmp_gcc_dirs["queues"]): daemon = GeminiDaemon(agent=agent, task_queue_file=str(queue_file)) daemon._write_result( "task-999", "great result", {"id": "task-999", "prompt": "test"}, success=True, ) result_file = tmp_gcc_dirs["results"] / "task-999.json" assert result_file.exists() data = json.loads(result_file.read_text()) assert data["task_id"] == "task-999" assert data["result"] == "great result" assert data["success"] is True def test_result_file_failure(self, tmp_gcc_dirs, agent): from core.gemini_command_centres.daemon import GeminiDaemon queue_file = tmp_gcc_dirs["queues"] / "test_agent.jsonl" queue_file.touch() with patch("core.gemini_command_centres.daemon.GCC_HEARTBEATS_DIR", tmp_gcc_dirs["heartbeats"]): with patch("core.gemini_command_centres.daemon.GCC_RESULTS_DIR", tmp_gcc_dirs["results"]): with patch("core.gemini_command_centres.daemon.GCC_TASK_QUEUES_DIR", tmp_gcc_dirs["queues"]): daemon = GeminiDaemon(agent=agent, task_queue_file=str(queue_file)) daemon._write_result( "task-fail", "API Error", {"id": "task-fail", "prompt": "broken"}, success=False, error="API Error", ) result_file = tmp_gcc_dirs["results"] / "task-fail.json" data = json.loads(result_file.read_text()) assert data["success"] is False assert data["error"] == "API Error" class TestBB8_LauncherTmux: """BB8: Launcher creates tmux sessions (mock subprocess).""" def test_launch_all_calls_tmux(self): from core.gemini_command_centres.launcher import launch_all with patch("subprocess.run") as mock_run: with patch("time.sleep"): mock_run.return_value = MagicMock(returncode=0) launch_all(verbose=False) # Should have called tmux for each of 5 centres tmux_calls = [ call for call in mock_run.call_args_list if call.args[0][0] == "tmux" ] assert len(tmux_calls) >= 5 def test_launch_one_success(self): from core.gemini_command_centres.launcher import launch_one with patch("subprocess.run") as mock_run: with patch("time.sleep"): mock_run.return_value = MagicMock(returncode=0) result = launch_one("orchestrator", verbose=False) assert result is True def test_launch_one_unknown_name(self): from core.gemini_command_centres.launcher import launch_one result = launch_one("nonexistent_centre", verbose=False) assert result is False class TestBB9_WatchdogStaleDetection: """BB9: Watchdog detects stale heartbeat.""" def test_stale_file_detected(self, tmp_gcc_dirs): from core.gemini_command_centres.watchdog import GCCWatchdog hb_file = tmp_gcc_dirs["heartbeats"] / "orchestrator.json" old_ts = ( datetime.now(timezone.utc) - timedelta(seconds=300) ).isoformat() hb_file.write_text(json.dumps({"timestamp": old_ts, "status": "running"})) with patch("core.gemini_command_centres.watchdog.GCC_HEARTBEATS_DIR", tmp_gcc_dirs["heartbeats"]): watchdog = GCCWatchdog(stale_threshold=120) is_stale = watchdog._is_stale(hb_file) assert is_stale is True def test_fresh_file_not_stale(self, tmp_gcc_dirs): from core.gemini_command_centres.watchdog import GCCWatchdog hb_file = tmp_gcc_dirs["heartbeats"] / "orchestrator.json" fresh_ts = datetime.now(timezone.utc).isoformat() hb_file.write_text(json.dumps({"timestamp": fresh_ts, "status": "running"})) with patch("core.gemini_command_centres.watchdog.GCC_HEARTBEATS_DIR", tmp_gcc_dirs["heartbeats"]): watchdog = GCCWatchdog(stale_threshold=120) is_stale = watchdog._is_stale(hb_file) assert is_stale is False def test_missing_file_is_stale(self, tmp_gcc_dirs): from core.gemini_command_centres.watchdog import GCCWatchdog missing_file = tmp_gcc_dirs["heartbeats"] / "no_such_file.json" watchdog = GCCWatchdog(stale_threshold=120) is_stale = watchdog._is_stale(missing_file) assert is_stale is True class TestBB10_WatchdogRespawn: """BB10: Watchdog respawns crashed centre.""" def test_respawn_called_for_stale_heartbeat(self, tmp_gcc_dirs): from core.gemini_command_centres.watchdog import GCCWatchdog, CENTRES_CONFIG watchdog = GCCWatchdog() with patch.object(watchdog, "_respawn") as mock_respawn: with patch.object(watchdog, "_is_stale", return_value=True): with patch("core.gemini_command_centres.watchdog.GCC_HEARTBEATS_DIR", tmp_gcc_dirs["heartbeats"]): # Create dummy heartbeat files so exists() returns True for c in CENTRES_CONFIG: hb = tmp_gcc_dirs["heartbeats"] / f"{c['name']}.json" hb.write_text("{}") # Manually trigger one iteration of the monitoring logic for centre in CENTRES_CONFIG[:1]: hb_file = tmp_gcc_dirs["heartbeats"] / f"{centre['name']}.json" if watchdog._is_stale(hb_file): watchdog._maybe_respawn(centre) mock_respawn.assert_called_once() def test_respawn_skipped_in_cooldown(self, tmp_gcc_dirs): from core.gemini_command_centres.watchdog import GCCWatchdog, CENTRES_CONFIG watchdog = GCCWatchdog() centre = CENTRES_CONFIG[0] # Set last respawn to just now (within cooldown period) watchdog._last_respawn[centre["name"]] = time.monotonic() with patch.object(watchdog, "_respawn") as mock_respawn: watchdog._maybe_respawn(centre) mock_respawn.assert_not_called() def test_respawn_not_called_when_max_exceeded(self, tmp_gcc_dirs): from core.gemini_command_centres.watchdog import GCCWatchdog, MAX_RESPAWN_ATTEMPTS, CENTRES_CONFIG watchdog = GCCWatchdog() centre = CENTRES_CONFIG[0] watchdog._respawn_counts[centre["name"]] = MAX_RESPAWN_ATTEMPTS watchdog._last_respawn[centre["name"]] = None # No cooldown with patch.object(watchdog, "_respawn") as mock_respawn: watchdog._maybe_respawn(centre) mock_respawn.assert_not_called() # ============================================================================= # WB: White-Box Tests (implementation internals) # ============================================================================= class TestWB1_TokenCounting: """WB1: Token counting from response usage_metadata.""" def test_token_counts_updated_after_chat(self, agent, mock_genai_client): asyncio.run(agent.chat("count these tokens")) assert agent.total_prompt_tokens == 100 assert agent.total_completion_tokens == 50 def test_token_counts_accumulate(self, agent, mock_genai_client): asyncio.run(agent.chat("first call")) asyncio.run(agent.chat("second call")) # Each call adds 100 prompt + 50 completion assert agent.total_prompt_tokens == 200 assert agent.total_completion_tokens == 100 def test_context_usage_returns_dict(self, agent): usage = agent.get_context_usage() assert "agent" in usage assert "total_prompt_tokens" in usage assert "utilisation_pct" in usage assert 0.0 <= usage["utilisation_pct"] <= 100.0 class TestWB2_HistoryCompaction: """WB2: History compaction when approaching context limit.""" def test_compaction_reduces_history(self, agent): from core.gemini_command_centres.agent import COMPACTION_KEEP_TURNS # Manually build a large history for i in range(100): agent.history.append({"role": "user", "parts": [{"text": f"msg {i}"}]}) agent.history.append({"role": "model", "parts": [{"text": f"resp {i}"}]}) agent._compact_history() # After compaction: 2 synthetic + COMPACTION_KEEP_TURNS*2 real expected_max = (COMPACTION_KEEP_TURNS * 2) + 2 assert len(agent.history) <= expected_max def test_compaction_inserts_summary_marker(self, agent): for i in range(60): agent.history.append({"role": "user", "parts": [{"text": f"msg {i}"}]}) agent.history.append({"role": "model", "parts": [{"text": f"resp {i}"}]}) agent._compact_history() # First entry should be a compaction note first_msg = agent.history[0]["parts"][0]["text"] assert "CONTEXT COMPACTED" in first_msg def test_compaction_not_triggered_for_small_history(self, agent): agent.history = [ {"role": "user", "parts": [{"text": "hello"}]}, {"role": "model", "parts": [{"text": "hi"}]}, ] original_len = len(agent.history) agent._compact_history() assert len(agent.history) == original_len class TestWB3_ErrorRetryLogic: """WB3: Error retry logic in daemon.""" def test_retry_increments_on_failure(self, tmp_gcc_dirs, agent, mock_genai_client): from core.gemini_command_centres.daemon import GeminiDaemon queue_file = tmp_gcc_dirs["queues"] / "test_agent.jsonl" queue_file.touch() call_count = 0 async def failing_chat(msg): nonlocal call_count call_count += 1 if call_count < 3: raise ConnectionError("Simulated API failure") return "success after retries" with patch("core.gemini_command_centres.daemon.GCC_HEARTBEATS_DIR", tmp_gcc_dirs["heartbeats"]): with patch("core.gemini_command_centres.daemon.GCC_RESULTS_DIR", tmp_gcc_dirs["results"]): with patch("core.gemini_command_centres.daemon.GCC_TASK_QUEUES_DIR", tmp_gcc_dirs["queues"]): daemon = GeminiDaemon(agent=agent, task_queue_file=str(queue_file)) with patch.object(agent, "chat", side_effect=failing_chat): task = {"id": "retry-task", "prompt": "do retry test"} asyncio.run(daemon._process_task(task)) # Was attempted 3 times before success assert call_count == 3 def test_handle_error_writes_failed_result(self, tmp_gcc_dirs, agent): from core.gemini_command_centres.daemon import GeminiDaemon queue_file = tmp_gcc_dirs["queues"] / "test_agent.jsonl" queue_file.touch() with patch("core.gemini_command_centres.daemon.GCC_HEARTBEATS_DIR", tmp_gcc_dirs["heartbeats"]): with patch("core.gemini_command_centres.daemon.GCC_RESULTS_DIR", tmp_gcc_dirs["results"]): with patch("core.gemini_command_centres.daemon.GCC_TASK_QUEUES_DIR", tmp_gcc_dirs["queues"]): daemon = GeminiDaemon(agent=agent, task_queue_file=str(queue_file)) task = {"id": "fail-task", "prompt": "this will fail"} daemon._handle_error(task, ValueError("critical failure")) result_file = tmp_gcc_dirs["results"] / "fail-task.json" assert result_file.exists() data = json.loads(result_file.read_text()) assert data["success"] is False assert daemon.tasks_failed == 1 class TestWB4_TaskQueueFileLocking: """WB4: Task queue file locking — concurrent pop safety.""" def test_concurrent_pops_unique_tasks(self, tmp_gcc_dirs, agent): """Two sequential pops from a 2-task queue return distinct tasks.""" from core.gemini_command_centres.daemon import GeminiDaemon queue_file = tmp_gcc_dirs["queues"] / "concurrent.jsonl" tasks = [ {"id": "c-1", "prompt": "task one"}, {"id": "c-2", "prompt": "task two"}, ] with open(queue_file, "w") as f: for t in tasks: f.write(json.dumps(t) + "\n") with patch("core.gemini_command_centres.daemon.GCC_HEARTBEATS_DIR", tmp_gcc_dirs["heartbeats"]): with patch("core.gemini_command_centres.daemon.GCC_RESULTS_DIR", tmp_gcc_dirs["results"]): with patch("core.gemini_command_centres.daemon.GCC_TASK_QUEUES_DIR", tmp_gcc_dirs["queues"]): daemon = GeminiDaemon(agent=agent, task_queue_file=str(queue_file)) first = daemon._pop_task() second = daemon._pop_task() assert first["id"] != second["id"] assert {first["id"], second["id"]} == {"c-1", "c-2"} class TestWB5_CTMJsonlFormat: """WB5: CTM JSONL format validation.""" def test_ctm_to_kg_writes_valid_jsonl(self, tmp_gcc_dirs): from core.gemini_command_centres.ctm import ctm_to_kg # Use a unique agent name to guarantee a fresh JSONL file each run unique_name = f"wb5_unique_{id(self)}" with patch("core.gemini_command_centres.ctm.KG_ENTITIES_DIR", tmp_gcc_dirs["kg_entities"]): with patch("core.gemini_command_centres.ctm.KG_AXIOMS_DIR", tmp_gcc_dirs["kg_axioms"]): result = ctm_to_kg(unique_name, "Test content", category="entity") assert result["status"] == "ok" written_file = Path(result["file"]) assert written_file.exists() # Validate JSONL format lines = written_file.read_text().strip().splitlines() assert len(lines) == 1 entry = json.loads(lines[0]) assert "id" in entry assert entry["source"] == f"gcc-{unique_name}" assert entry["content"] == "Test content" assert entry["type"] == "entity" assert "timestamp" in entry def test_ctm_to_kg_appends_multiple_entries(self, tmp_gcc_dirs): from core.gemini_command_centres.ctm import ctm_to_kg import uuid as _uuid # Unique per-test-run agent name to avoid cross-run file accumulation unique_name = f"agent_append_{_uuid.uuid4().hex[:8]}" with patch("core.gemini_command_centres.ctm.KG_ENTITIES_DIR", tmp_gcc_dirs["kg_entities"]): with patch("core.gemini_command_centres.ctm.KG_AXIOMS_DIR", tmp_gcc_dirs["kg_axioms"]): ctm_to_kg(unique_name, "Entry one", category="entity") result = ctm_to_kg(unique_name, "Entry two", category="entity") written_file = Path(result["file"]) lines = written_file.read_text().strip().splitlines() assert len(lines) == 2 def test_ctm_to_memory_appends_to_file(self, tmp_gcc_dirs): from core.gemini_command_centres.ctm import ctm_to_memory memory_file = tmp_gcc_dirs["memory"] / "MEMORY.md" with patch("core.gemini_command_centres.ctm.MEMORY_FILE", memory_file): result = ctm_to_memory("Important memory entry", agent_name="scout") assert result["status"] == "ok" assert memory_file.exists() content = memory_file.read_text() assert "Important memory entry" in content assert "GCC-SCOUT" in content def test_ctm_id_is_unique(self, tmp_gcc_dirs): from core.gemini_command_centres.ctm import ctm_to_kg with patch("core.gemini_command_centres.ctm.KG_ENTITIES_DIR", tmp_gcc_dirs["kg_entities"]): with patch("core.gemini_command_centres.ctm.KG_AXIOMS_DIR", tmp_gcc_dirs["kg_axioms"]): r1 = ctm_to_kg("agent", "content 1") r2 = ctm_to_kg("agent", "content 2") assert r1["id"] != r2["id"] # ============================================================================= # Additional Integration Tests # ============================================================================= class TestIntegration_DispatchAndStatus: """Integration: dispatch + status via launcher.""" def test_dispatch_writes_queue_file(self, tmp_gcc_dirs): from core.gemini_command_centres.launcher import dispatch with patch("core.gemini_command_centres.launcher.GCC_TASK_QUEUES_DIR", tmp_gcc_dirs["queues"]): task_id = dispatch("test task", target="orchestrator", verbose=False) queue_file = tmp_gcc_dirs["queues"] / "orchestrator.jsonl" assert queue_file.exists() entry = json.loads(queue_file.read_text().strip()) assert entry["id"] == task_id assert entry["prompt"] == "test task" def test_status_returns_no_heartbeat_for_fresh_install(self, tmp_gcc_dirs): from core.gemini_command_centres.launcher import status with patch("core.gemini_command_centres.launcher.GCC_HEARTBEATS_DIR", tmp_gcc_dirs["heartbeats"]): result = status(verbose=False) for name, info in result.items(): # No heartbeat files in tmp_path → all should be no_heartbeat or similar assert info.get("healthy") is False def test_centres_config_has_five_entries(self): from core.gemini_command_centres.launcher import CENTRES_CONFIG assert len(CENTRES_CONFIG) == 5 def test_centres_config_model_ids_present(self): from core.gemini_command_centres.launcher import CENTRES_CONFIG for c in CENTRES_CONFIG: assert "model" in c assert c["model"].startswith("gemini-") assert "name" in c assert "system_prompt" in c # ============================================================================= # Test Runner Summary # ============================================================================= if __name__ == "__main__": import subprocess import sys result = subprocess.run( [sys.executable, "-m", "pytest", __file__, "-v", "--tb=short"], cwd="/mnt/e/genesis-system", ) sys.exit(result.returncode)