Something went wrong. Try again.
Personal fork of prime agent. Mirror of: https://mnemosyne.sigilzero.dev/sigilzero/prime-agent
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137#!/usr/bin/env python3"""Prove stale tool text cannot satisfy a current kernel probe and cell quoting works."""import astimport asynciofrom types import SimpleNamespaceimport unittestimport jsonfrom pathlib import Pathimport signalfrom unittest.mock import patch, MagicMockfrom review_kernel_env_parity import current_probe, CODE, MARKER, owned_kernel, finish_daemon, KernelClient, signal_owned, reap_leftovers
def requests(content=None): messages = [{'role': 'tool', 'tool_call_id': 'kernel_old', 'content': MARKER + '{"old":true}'}] if content is not None: messages.append({'role': 'tool', 'tool_call_id': 'kernel_new', 'content': content}) return [{}, {'messages': messages}]
class ProbeTests(unittest.TestCase): def test_only_current_result_counts(self): self.assertEqual(current_probe(requests(MARKER + '{"new":true}'), 'new'), {'new': True})
def test_old_probe_cannot_hide_failed_cell(self): with self.assertRaisesRegex(AssertionError, 'current kernel probe'): current_probe(requests('Traceback: failed current cell'), 'new')
def test_old_probe_cannot_hide_missing_result(self): with self.assertRaisesRegex(AssertionError, 'current tool result'): current_probe(requests(), 'new')
def test_duplicate_probe_is_failure(self): with self.assertRaisesRegex(AssertionError, 'current kernel probe'): current_probe(requests(MARKER + '{}\n' + MARKER + '{}'), 'new')
def test_cell_passes_the_intended_bash_command(self): commands = [] async def bash(command): commands.append(command) return SimpleNamespace(output='ABSENT', exit_code=0) compiled = compile(CODE, '<fixture-cell>', 'exec', flags=ast.PyCF_ALLOW_TOP_LEVEL_AWAIT) asyncio.run(eval(compiled, {'bash': bash, 'print': lambda *_args: None})) self.assertEqual(commands, ['printf \'%s\' "${PA_PREFIX_SENTINEL:-ABSENT}"'])
def test_foreign_fixture_kernel_is_not_owned(self): env = b'PRIME_AGENT_KERNEL_OWNER_PID=123\0PA_KERNEL_ENV_FIXTURE_ID=/other-fixture\0' with patch('review_kernel_env_parity.process_identity', return_value=('proc:456', 'S')), \ patch('pathlib.Path.read_bytes', return_value=env): self.assertIsNone(owned_kernel(789, 123, '/fixture'))
def test_wrong_kernel_owner_is_not_owned(self): env = b'PRIME_AGENT_KERNEL_OWNER_PID=999\0PA_KERNEL_ENV_FIXTURE_ID=/fixture\0' with patch('review_kernel_env_parity.process_identity', return_value=('proc:456', 'S')), \ patch('pathlib.Path.read_bytes', return_value=env): self.assertIsNone(owned_kernel(789, 123, '/fixture'))
def test_recycled_kernel_pid_is_not_owned(self): with patch('review_kernel_env_parity.process_identity', return_value=('proc:new', 'S')): self.assertIsNone(owned_kernel(789, 123, '/fixture', 'proc:old'))
def test_matching_kernel_owner_fixture_and_start_identity(self): env = b'PRIME_AGENT_KERNEL_OWNER_PID=123\0PA_KERNEL_ENV_FIXTURE_ID=/fixture\0' with patch('review_kernel_env_parity.process_identity', return_value=('proc:456', 'S')), \ patch('pathlib.Path.read_bytes', side_effect=[env, b'/fixture/python3\0-m\0rlm.kernel\0']): self.assertEqual(owned_kernel(789, 123, '/fixture', 'proc:456'), 'proc:456')
def test_shutdown_ack_then_control_close_then_process_exit(self): events = [] client, process = MagicMock(), MagicMock(returncode=0) client.call.side_effect = lambda *a, **k: events.append('ack') client.close.side_effect = lambda: events.append('client_closed') process.wait.side_effect = lambda **k: events.append('process_exit') finish_daemon(client, process) self.assertEqual(events, ['ack', 'client_closed', 'process_exit'])
def test_refused_shutdown_is_not_cleanup_success(self): client, process = MagicMock(), MagicMock(returncode=0) client.call.side_effect = AssertionError('refused') with self.assertRaisesRegex(AssertionError, 'refused'): finish_daemon(client, process) process.wait.assert_not_called()
def test_cold_kernel_response_after_35_seconds_uses_request_budget(self): client = KernelClient.__new__(KernelClient) client.counter, client.wire = 0, [] client.sock = MagicMock() client.sock.gettimeout.return_value = 35 client.reader = MagicMock() client.reader.readline.return_value = json.dumps({'type': 'response', 'id': '1', 'success': True, 'data': {'done': True}}).encode() with patch('review_kernel_env_parity.time.monotonic', side_effect=[0, 36, 36]): self.assertEqual(client.call('prompt_and_wait', timeout_seconds=90), {'done': True}) self.assertEqual([call.args[0] for call in client.sock.settimeout.call_args_list], [90, 54, 35])
def test_recycled_pid_never_receives_fallback_signal(self): with patch('os.pidfd_open', return_value=42, create=True), patch('os.close'), \ patch('review_kernel_env_parity.process_identity', return_value=('proc:new', 'S')), \ patch('signal.pidfd_send_signal', create=True) as sent: self.assertFalse(signal_owned(789, 'proc:old', lambda: True, signal.SIGTERM)) sent.assert_not_called()
def test_failed_ownership_check_never_receives_fallback_signal(self): with patch('os.pidfd_open', return_value=42, create=True), patch('os.close'), \ patch('review_kernel_env_parity.process_identity', return_value=('proc:456', 'S')), \ patch('signal.pidfd_send_signal', create=True) as sent: self.assertFalse(signal_owned(789, 'proc:456', lambda: False, signal.SIGTERM)) sent.assert_not_called()
def test_fallback_signals_only_owned_kernel_and_preserves_red_status(self): observation = {'cleanup_error': 'graceful shutdown refused', 'cleanup_verified': False} with patch('pathlib.Path.rglob', return_value=[]), \ patch('pathlib.Path.iterdir', return_value=[Path('/proc/789')]), \ patch('review_kernel_env_parity.owned_kernel', return_value='proc:456'), \ patch('review_kernel_env_parity.signal_owned', return_value=True) as sent, \ patch('review_kernel_env_parity.live_identities', return_value=set()): reap_leftovers(Path('/fixture'), Path('/fixture/binary'), Path('/fixture/d.sock'), None, {789: 'proc:456'}, 123, '/fixture', observation) sent.assert_called_once() self.assertEqual(sent.call_args.args[:2], (789, 'proc:456')) self.assertTrue(observation['fallback_processes_exited']) self.assertEqual(observation['cleanup_error'], 'graceful shutdown refused') self.assertFalse(observation['cleanup_verified'])
def test_foreign_kernel_never_becomes_fallback_target(self): with patch('pathlib.Path.rglob', return_value=[]), \ patch('pathlib.Path.iterdir', return_value=[Path('/proc/789')]), \ patch('review_kernel_env_parity.owned_kernel', return_value=None), \ patch('review_kernel_env_parity.signal_owned') as sent, \ patch('review_kernel_env_parity.live_identities', return_value=set()): reap_leftovers(Path('/fixture'), Path('/fixture/binary'), Path('/fixture/d.sock'), None, {789: 'proc:456'}, 123, '/fixture', {}) sent.assert_not_called()
if __name__ == '__main__': unittest.main()