diff --git a/tests/test_zellij_metadata.py b/tests/test_zellij_metadata.py index 7dbe971..6356f17 100644 --- a/tests/test_zellij_metadata.py +++ b/tests/test_zellij_metadata.py @@ -481,5 +481,213 @@ class TestEndToEndZellijMetadata(unittest.TestCase): self.assertEqual(data["session_id"], "non-zellij-agent") +# =========================================================================== +# Pending spawn registry for Codex correlation +# =========================================================================== + +class TestPendingSpawnRegistry(unittest.TestCase): + """Verify pending spawn files enable Codex session correlation.""" + + def test_write_pending_spawn_creates_file(self): + """_write_pending_spawn creates a JSON file with correct data.""" + with tempfile.TemporaryDirectory() as tmpdir: + pending_dir = Path(tmpdir) / "pending" + with patch.object(spawn_mod, "PENDING_SPAWNS_DIR", pending_dir): + spawn_mod._write_pending_spawn( + "test-spawn-id", + Path("/home/user/projects/myproject"), + "codex", + ) + + pending_file = pending_dir / "test-spawn-id.json" + self.assertTrue(pending_file.exists()) + data = json.loads(pending_file.read_text()) + self.assertEqual(data["spawn_id"], "test-spawn-id") + self.assertEqual(data["project_path"], "/home/user/projects/myproject") + self.assertEqual(data["agent_type"], "codex") + self.assertIn("timestamp", data) + + def test_cleanup_removes_stale_pending_spawns(self): + """_cleanup_stale_pending_spawns removes old files.""" + import time + with tempfile.TemporaryDirectory() as tmpdir: + pending_dir = Path(tmpdir) + pending_dir.mkdir(exist_ok=True) + + # Create an old file (simulate 120 seconds ago) + old_file = pending_dir / "old-spawn.json" + old_file.write_text('{"spawn_id": "old"}') + old_mtime = time.time() - 120 + os.utime(old_file, (old_mtime, old_mtime)) + + # Create a fresh file + fresh_file = pending_dir / "fresh-spawn.json" + fresh_file.write_text('{"spawn_id": "fresh"}') + + with patch.object(spawn_mod, "PENDING_SPAWNS_DIR", pending_dir), \ + patch.object(spawn_mod, "PENDING_SPAWN_TTL", 60): + spawn_mod._cleanup_stale_pending_spawns() + + self.assertFalse(old_file.exists(), "Old file should be deleted") + self.assertTrue(fresh_file.exists(), "Fresh file should remain") + + +class TestParseSessionTimestamp(unittest.TestCase): + """Verify _parse_session_timestamp handles various formats.""" + + def test_parses_iso_with_z_suffix(self): + """Parse ISO timestamp with Z suffix.""" + from amc_server.mixins.discovery import _parse_session_timestamp + result = _parse_session_timestamp("2026-02-27T10:00:00Z") + self.assertIsNotNone(result) + self.assertIsInstance(result, float) + + def test_parses_iso_with_offset(self): + """Parse ISO timestamp with timezone offset.""" + from amc_server.mixins.discovery import _parse_session_timestamp + result = _parse_session_timestamp("2026-02-27T10:00:00+00:00") + self.assertIsNotNone(result) + + def test_returns_none_for_empty_string(self): + """Empty string returns None.""" + from amc_server.mixins.discovery import _parse_session_timestamp + self.assertIsNone(_parse_session_timestamp("")) + + def test_returns_none_for_invalid_format(self): + """Invalid format returns None.""" + from amc_server.mixins.discovery import _parse_session_timestamp + self.assertIsNone(_parse_session_timestamp("not-a-timestamp")) + + +class TestMatchPendingSpawn(unittest.TestCase): + """Verify _match_pending_spawn correlates Codex sessions to spawns.""" + + def _make_iso_timestamp(self, offset_seconds=0): + """Create ISO timestamp string offset from now.""" + import time + from datetime import datetime, timezone + ts = time.time() + offset_seconds + return datetime.fromtimestamp(ts, tz=timezone.utc).isoformat() + + def test_matches_by_cwd_and_returns_spawn_id(self): + """Pending spawn matched by CWD returns spawn_id and deletes file.""" + import time + from amc_server.mixins.discovery import _match_pending_spawn + + with tempfile.TemporaryDirectory() as tmpdir: + pending_dir = Path(tmpdir) + + # Create pending spawn 5 seconds ago + spawn_ts = time.time() - 5 + pending_file = pending_dir / "match-me.json" + pending_file.write_text(json.dumps({ + "spawn_id": "match-me", + "project_path": "/home/user/projects/amc", + "agent_type": "codex", + "timestamp": spawn_ts, + })) + + with patch("amc_server.mixins.discovery.PENDING_SPAWNS_DIR", pending_dir): + # Session started now (ISO string, after spawn) + session_ts = self._make_iso_timestamp(0) + result = _match_pending_spawn("/home/user/projects/amc", session_ts) + + self.assertEqual(result, "match-me") + self.assertFalse(pending_file.exists(), "Pending file should be deleted") + + def test_no_match_when_cwd_differs(self): + """Pending spawn not matched when CWD differs.""" + import time + from amc_server.mixins.discovery import _match_pending_spawn + + with tempfile.TemporaryDirectory() as tmpdir: + pending_dir = Path(tmpdir) + + pending_file = pending_dir / "no-match.json" + pending_file.write_text(json.dumps({ + "spawn_id": "no-match", + "project_path": "/home/user/projects/other", + "agent_type": "codex", + "timestamp": time.time() - 5, + })) + + with patch("amc_server.mixins.discovery.PENDING_SPAWNS_DIR", pending_dir): + session_ts = self._make_iso_timestamp(0) + result = _match_pending_spawn("/home/user/projects/amc", session_ts) + + self.assertIsNone(result) + self.assertTrue(pending_file.exists(), "Pending file should remain") + + def test_no_match_when_session_older_than_spawn(self): + """Pending spawn not matched if session started before spawn.""" + import time + from amc_server.mixins.discovery import _match_pending_spawn + + with tempfile.TemporaryDirectory() as tmpdir: + pending_dir = Path(tmpdir) + + # Spawn happens NOW + spawn_ts = time.time() + pending_file = pending_dir / "future-spawn.json" + pending_file.write_text(json.dumps({ + "spawn_id": "future-spawn", + "project_path": "/home/user/projects/amc", + "agent_type": "codex", + "timestamp": spawn_ts, + })) + + with patch("amc_server.mixins.discovery.PENDING_SPAWNS_DIR", pending_dir): + # Session started 10 seconds BEFORE spawn (ISO string) + session_ts = self._make_iso_timestamp(-10) + result = _match_pending_spawn("/home/user/projects/amc", session_ts) + + self.assertIsNone(result) + + def test_no_match_for_claude_agent_type(self): + """Pending spawn for claude not matched to codex discovery.""" + import time + from amc_server.mixins.discovery import _match_pending_spawn + + with tempfile.TemporaryDirectory() as tmpdir: + pending_dir = Path(tmpdir) + + pending_file = pending_dir / "claude-spawn.json" + pending_file.write_text(json.dumps({ + "spawn_id": "claude-spawn", + "project_path": "/home/user/projects/amc", + "agent_type": "claude", # Not codex + "timestamp": time.time() - 5, + })) + + with patch("amc_server.mixins.discovery.PENDING_SPAWNS_DIR", pending_dir): + session_ts = self._make_iso_timestamp(0) + result = _match_pending_spawn("/home/user/projects/amc", session_ts) + + self.assertIsNone(result) + + def test_no_match_when_session_timestamp_unparseable(self): + """Session with invalid timestamp format doesn't match.""" + import time + from amc_server.mixins.discovery import _match_pending_spawn + + with tempfile.TemporaryDirectory() as tmpdir: + pending_dir = Path(tmpdir) + + pending_file = pending_dir / "valid-spawn.json" + pending_file.write_text(json.dumps({ + "spawn_id": "valid-spawn", + "project_path": "/home/user/projects/amc", + "agent_type": "codex", + "timestamp": time.time() - 5, + })) + + with patch("amc_server.mixins.discovery.PENDING_SPAWNS_DIR", pending_dir): + # Invalid timestamp format should not match + result = _match_pending_spawn("/home/user/projects/amc", "invalid-timestamp") + + self.assertIsNone(result) + self.assertTrue(pending_file.exists(), "Pending file should remain") + + if __name__ == "__main__": unittest.main()