mirror of
https://github.com/morluto/rea.git
synced 2026-10-19 04:19:54 -05:00
Ysopwnableandfactory-droid[bot]
ee87111009
Add optional owned Codex tool/usage capture and exact-name discovery experiments while preserving default ephemeral trials and all REA public APIs. Retain scoped managed and actual Android results, mixed outcomes, source identities and resource unknowns. Six semantic Python tests, full checked pre-push, byte-identical own-file rebase verification, manual review and all 39 exact-head CI checks pass. Historical main fixture failures and their independent repair remain documented; measured source/runtime pins are unchanged. Co-authored-by: factory-droid[bot] <138933559+factory-droid[bot]@users.noreply.github.com>
176 lines
9.7 KiB
Python
176 lines
9.7 KiB
Python
"""Distinct capture/cleanup failures; simulated CLI is not real-model evidence."""
|
|
|
|
import importlib.util
|
|
import io
|
|
import json
|
|
import os
|
|
import runpy
|
|
import sys
|
|
from pathlib import Path
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import Mock, patch
|
|
|
|
from codex_context import capture_tool_context, selected_event
|
|
|
|
|
|
class ManagedCoverageTests(unittest.TestCase):
|
|
def test_verifies_extra_counts_and_preserves_unknown_fields(self):
|
|
review = runpy.run_path(str(Path(__file__).with_name('review-model.py')))
|
|
project = review['scope_with_verified_counts']
|
|
expected = {'state': 'partial', 'issues': ['unreturned rows remain unknown']}
|
|
original = {**expected, 'module_ref_count': 0, 'pinvoke_import_count': 2}
|
|
scope, checks = project(original, expected, {'module_ref_count': 0, 'pinvoke_import_count': 2})
|
|
self.assertEqual(scope, expected)
|
|
self.assertTrue(all(checks.values()))
|
|
self.assertEqual(original['pinvoke_import_count'], 2)
|
|
for count in [3, '2', True]:
|
|
scope, checks = project({**expected, 'pinvoke_import_count': count}, expected, {'pinvoke_import_count': 2})
|
|
self.assertFalse(checks['pinvoke_import_count'])
|
|
_, checks = project({**expected, 'module_ref_count': False}, {**expected, 'module_ref_count': 0}, {'module_ref_count': 0})
|
|
self.assertFalse(checks['module_ref_count'])
|
|
scope, _ = project({**expected, 'unsupported_claim': True}, expected, {})
|
|
self.assertNotEqual(scope, expected)
|
|
scope, _ = project({'state': 'partial', 'module_ref_count': 0}, expected, {'module_ref_count': 0})
|
|
self.assertNotEqual(scope, expected)
|
|
ambiguous = review['named_scope']({'native_boundaries': expected, 'native_boundary_declarations': {'state': 'complete'}}, ['native_boundaries', 'native_boundary_declarations'])
|
|
scope, _ = project(ambiguous, expected, {'module_ref_count': 0})
|
|
self.assertNotEqual(scope, expected)
|
|
|
|
|
|
class ContextCaptureTests(unittest.TestCase):
|
|
def test_filters_ambient_account_and_nested_usage_state(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
home = root / 'home'
|
|
sessions = home / 'sessions'
|
|
sessions.mkdir(parents=True)
|
|
output = root / 'output'
|
|
output.mkdir()
|
|
events = [
|
|
{'type': 'world_state', 'payload': {'state': {'env': 'ambient-marker'}}},
|
|
{'type': 'session_meta', 'payload': {'creator_account_id': 'account-marker'}},
|
|
{'type': 'response_item', 'payload': {
|
|
'type': 'custom_tool_call_output', 'call_id': 'original-call',
|
|
'output': 'caller-selected evidence and paths',
|
|
'internal_chat_message_metadata_passthrough': 'private-marker'}},
|
|
{'type': 'token_usage_record', 'payload': {
|
|
'response_id': 'original-response', 'usage': {
|
|
'input_tokens': 12, 'cached_input_tokens': 4,
|
|
'environment': 'nested-ambient-marker'},
|
|
'transport_auth': 'auth-marker'}},
|
|
]
|
|
(sessions / 'rollout.jsonl').write_text(''.join(json.dumps(x) + '\n' for x in events))
|
|
(home / 'auth.json').write_text('copied-credential-marker')
|
|
receipt = capture_tool_context(home, output)
|
|
text = (output / 'codex-tool-context.jsonl').read_text()
|
|
self.assertEqual(receipt['status'], 'captured')
|
|
self.assertEqual(receipt['retained_events'], 2)
|
|
for marker in ['ambient-marker', 'account-marker', 'private-marker',
|
|
'auth-marker', 'copied-credential-marker']:
|
|
self.assertNotIn(marker, text)
|
|
self.assertIn('caller-selected evidence and paths', text)
|
|
self.assertIn('original-response', text)
|
|
self.assertEqual((output / 'codex-tool-context.jsonl').stat().st_mode & 0o777, 0o600)
|
|
|
|
def test_missing_rollouts_are_unknown_and_malformed_lines_remain_partial(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
output = root / 'missing'
|
|
output.mkdir()
|
|
self.assertEqual(capture_tool_context(root / 'home', output)['status'], 'unavailable')
|
|
sessions = root / 'home' / 'sessions'
|
|
sessions.mkdir(parents=True)
|
|
(sessions / 'rollout.jsonl').write_text('not-json\n' + json.dumps({
|
|
'type': 'event_msg', 'payload': {'type': 'token_count', 'info': None}}) + '\n')
|
|
partial = root / 'partial'
|
|
partial.mkdir()
|
|
receipt = capture_tool_context(root / 'home', partial)
|
|
self.assertEqual(receipt['status'], 'partial')
|
|
self.assertEqual(receipt['errors'][0]['line'], 1)
|
|
self.assertEqual(receipt['retained_events'], 1)
|
|
with self.assertRaises(ValueError):
|
|
selected_event({'type': 'token_usage_record', 'payload': {'usage': {'input_tokens': True}}})
|
|
with self.assertRaises(ValueError):
|
|
selected_event({'type': {'environment': 'unexpected'}, 'payload': {}})
|
|
with self.assertRaises(ValueError):
|
|
selected_event({'type': 'response_item', 'timestamp': {'environment': 'unexpected'}, 'payload': {}})
|
|
unknown = selected_event({'type': 'event_msg', 'payload': {'type': 'token_count', 'info': {'model_context_window': None}}})
|
|
self.assertIsNone(unknown['payload']['info']['model_context_window'])
|
|
|
|
def test_capture_failure_still_removes_trial_credentials_and_profile(self):
|
|
spec = importlib.util.spec_from_file_location('agent_trial', Path(__file__).with_name('agent-trial.py'))
|
|
agent = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(agent)
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
runtime = root / 'runtime'
|
|
(runtime / 'scripts').mkdir(parents=True)
|
|
(runtime / 'scripts' / 'rea.mjs').write_text('')
|
|
(runtime / 'package-lock.json').write_text('{}')
|
|
auth_home = root / 'source-auth'
|
|
auth_home.mkdir()
|
|
(auth_home / 'auth.json').write_text('synthetic-credential')
|
|
question = root / 'question.txt'
|
|
question.write_text('Observe the declared fixture.')
|
|
output = root / 'trial'
|
|
child = Mock(stdin=io.StringIO())
|
|
child.wait.return_value = 0
|
|
arguments = ['agent-trial.py', '--runtime-root', str(runtime), '--revision', 'fixture',
|
|
'--prompt', str(question), '--output', str(output), '--capture-tool-context']
|
|
with patch.dict(os.environ, {'CODEX_HOME': str(auth_home)}, clear=True), \
|
|
patch.object(sys, 'argv', arguments), \
|
|
patch.object(agent.shutil, 'which', return_value='/fixture/node'), \
|
|
patch.object(agent.subprocess, 'check_output', return_value='v24.18.0\n'), \
|
|
patch.object(agent.subprocess, 'Popen', return_value=child), \
|
|
patch.object(agent, 'capture_tool_context', side_effect=OSError('fixture capture failure')), \
|
|
patch('builtins.print'):
|
|
with self.assertRaisesRegex(OSError, 'fixture capture failure'):
|
|
agent.main()
|
|
self.assertFalse(list(output.glob('profile-*')))
|
|
self.assertTrue(json.loads((output / 'cleanup.json').read_text())['profile_removed'])
|
|
self.assertEqual((auth_home / 'auth.json').read_text(), 'synthetic-credential')
|
|
|
|
def test_refuses_foreign_session_directory_and_symlink_records(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
home = root / 'home'
|
|
home.mkdir()
|
|
foreign = root / 'foreign'
|
|
foreign.mkdir()
|
|
(foreign / 'rollout.jsonl').write_text('private marker')
|
|
(home / 'sessions').symlink_to(foreign, target_is_directory=True)
|
|
output = root / 'directory-result'
|
|
output.mkdir()
|
|
receipt = capture_tool_context(home, output)
|
|
self.assertEqual(receipt['status'], 'partial')
|
|
self.assertEqual(receipt['retained_events'], 0)
|
|
(home / 'sessions').unlink()
|
|
(home / 'sessions').mkdir()
|
|
(home / 'sessions' / 'linked.jsonl').symlink_to(foreign / 'rollout.jsonl')
|
|
output = root / 'file-result'
|
|
output.mkdir()
|
|
receipt = capture_tool_context(home, output)
|
|
self.assertEqual(receipt['status'], 'partial')
|
|
self.assertEqual(receipt['errors'][0]['reason'], 'not_an_owned_regular_file')
|
|
self.assertEqual((foreign / 'rollout.jsonl').read_text(), 'private marker')
|
|
|
|
def test_guidance_requires_selected_tools_before_acquiring_a_profile(self):
|
|
spec = importlib.util.spec_from_file_location('agent_trial', Path(__file__).with_name('agent-trial.py'))
|
|
agent = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(agent)
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
output = root / 'trial'
|
|
arguments = ['agent-trial.py', '--runtime-root', str(root), '--revision', 'fixture',
|
|
'--prompt', str(root / 'question'), '--output', str(output), '--guide-enabled-tools']
|
|
with patch.object(sys, 'argv', arguments), patch.object(sys, 'stderr', io.StringIO()):
|
|
with self.assertRaises(SystemExit) as raised:
|
|
agent.main()
|
|
self.assertEqual(raised.exception.code, 2)
|
|
self.assertFalse(output.exists())
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|