"""Only the public wrapper records bounded program preflight telemetry.""" from copy import deepcopy import json import pytest from backend.browser import tools from backend.config import settings from test_reply_preflight_d44 import Page, Session, find, node, speaker, target, thread, tree @pytest.fixture(autouse=True) def enable_reply_relation_preflight(monkeypatch): monkeypatch.setattr(settings, 'reply_relation_preflight_enabled', True) @pytest.fixture def telemetry(tmp_path, monkeypatch): task_id = 'preflight_telemetry_d44' tools.set_prospecting_context(task_id) tools.set_prospecting_run_context(task_id, 'f'*32, goal_version=3) tools._reply_target_proofs[task_id] = {'private-source': {'run_id': 'old', 'profile_id': 'c'*24}} monkeypatch.setattr(tools, 'task_doc_dir', lambda _: tmp_path) def forbidden(*_args, **_kwargs): raise AssertionError('preflight telemetry is not a source, operation, knowledge fact or proof') for name in ('_operation_evidence', '_save_source', '_apply_reply_target_proof'): monkeypatch.setattr(tools, name, forbidden) proof = deepcopy(tools._reply_target_proofs) yield task_id, tmp_path assert proof == tools._reply_target_proofs tools.clear_task_context(task_id) async def public(task_id, page): action = tools.build_prospecting_tools(task_id).registry.registry.actions['inspect_reply_relations'] result = await action.function(browser_session=Session(page), max_replies=5) return json.loads(result.extracted_content) def receipt(task_id): return tools._task_artifacts[task_id]['preflight_observations'][-1] def payload(): return {'read_only': True, 'scope': 'current_viewport_only', 'note_id': 'a'*24, 'status': 'checked', 'checked_count': 1, 'eligible_count': 1, 'discussion_state': 'visible_comments', 'visible_comment_count': 3, 'has_unexpanded_replies': False, 'candidates': [{'classification': 'unique_dom_deep_reply', 'eligible': True, 'observation_stable': True, 'parent_is_reply': True, 'parent_relation_observed': True, 'native_navigation_confirmed': False, 'source_selector': '#PRIVATE-LOCATOR', 'quote': 'PRIVATE-QUOTE', 'rawhref': 'https://private.invalid/?xsec_token=PRIVATE-TOKEN', 'user_id': 'c'*24}]} def record(task_id, result): return tools._record_reply_preflight_observation(task_id, result, deepcopy(tools._prospecting_run_context[task_id])) async def test_public_wrapper_counts_match_real_helper_and_only_telemetry_is_added(telemetry): task_id, directory = telemetry page = Page(tree(thread(speaker('parent', 'c'), speaker('child', 'd', target())))) before = deepcopy(tools._task_artifacts[task_id]) result = await public(task_id, page) row = receipt(task_id) assert result['telemetry_recorded'] is True assert row['run_id'] == 'f'*32 and row['goal_version'] == 3 and row['note_id'] == 'a'*24 assert row['checked_count'] == result['checked_count'] == len(result['candidates']) == 2 assert row['eligible_count'] == result['eligible_count'] == 1 assert row['classification_counts'] == {'unique_dom_deep_reply': 1, 'needs_native_identity': 0, 'ambiguous': 0, 'unknown': 1} assert row['discussion_state'] == 'visible_comments' and row['visible_comment_count'] == 3 assert row['has_unexpanded_replies'] is False assert {key: value for key, value in tools._task_artifacts[task_id].items() if key != 'preflight_observations'} == before evidence = json.loads((directory/'evidence.json').read_text()) assert evidence['preflight_observations'] == [row] assert all(evidence[key] == [] for key in ('sources', 'source_observations', 'operation_evidence', 'learning_candidates', 'learning_outcomes')) serialized = json.dumps(evidence) assert 'PRIVATE' not in serialized and 'https://' not in serialized and 'selector' not in serialized assert 'c'*24 not in serialized and 'd'*24 not in serialized async def test_no_visible_reply_preserves_comments_not_in_view_instead_of_claiming_read_complete(telemetry): task_id, _ = telemetry value = tree(thread()) find(value, 'thread-root')['rect'] = {'left': 0, 'right': 300, 'top': 800, 'bottom': 835, 'width': 300, 'height': 35} value['children'].append(node('comments', 'comments-container')) result = await public(task_id, Page(value)) row = receipt(task_id) assert result['status'] == row['status'] == 'no_visible_reply' assert row['checked_count'] == row['eligible_count'] == 0 assert row['discussion_state'] == 'comments_not_in_view' and row['visible_comment_count'] == 0 assert row['has_unexpanded_replies'] is False async def test_failed_dom_listing_records_unknown_metadata_not_invented_zero(telemetry, monkeypatch): task_id, directory = telemetry async def failed(*_args, **_kwargs): raise ValueError('PRIVATE-RESPONSE https://private.invalid') monkeypatch.setattr(tools, '_inspect_source_candidates', failed) result = await public(task_id, Page(tree())) row = receipt(task_id) assert result['telemetry_recorded'] is True and row['status'] == 'unknown' assert row['discussion_state'] == 'unknown' assert row['visible_comment_count'] is None and row['has_unexpanded_replies'] is None assert row['checked_count'] == row['eligible_count'] == 0 assert 'PRIVATE' not in (directory/'evidence.json').read_text() async def test_visible_root_with_folded_replies_keeps_unexpanded_scope_distinct_from_zero_reply_candidates(telemetry): task_id, _ = telemetry value = tree(thread()) find(value, 'thread-replies')['children'].append(node('expand', '', '展开2条回复')) result = await public(task_id, Page(value)) row = receipt(task_id) assert result['status'] == row['status'] == 'no_visible_reply' assert row['checked_count'] == 0 and row['discussion_state'] == 'visible_comments' assert row['visible_comment_count'] == 1 and row['has_unexpanded_replies'] is True @pytest.mark.parametrize('context', [{}, {'run_id': 'f'*32}, {'run_id': 'PRIVATE-RUN', 'goal_version': 3}, {'run_id': 'f'*32, 'goal_version': True}, {'run_id': 'f'*32, 'goal_version': 0}]) async def test_wrapper_does_not_attach_new_receipts_to_missing_or_invalid_run_goal(telemetry, monkeypatch, context): task_id, directory = telemetry tools._prospecting_run_context[task_id] = context async def checked(*_args, **_kwargs): return payload() monkeypatch.setattr(tools, '_inspect_reply_relations', checked) result = await public(task_id, Page(tree())) assert result['telemetry_recorded'] is False assert 'preflight_observations' not in tools._task_artifacts[task_id] assert list(directory.iterdir()) == [] async def test_run_or_goal_changed_during_read_does_not_record_stale_result(telemetry, monkeypatch): task_id, directory = telemetry async def checked(*_args, **_kwargs): tools._prospecting_run_context[task_id] = {'run_id': 'e'*32, 'goal_version': 4} return payload() monkeypatch.setattr(tools, '_inspect_reply_relations', checked) result = await public(task_id, Page(tree())) assert result['telemetry_recorded'] is False assert 'preflight_observations' not in tools._task_artifacts[task_id] assert list(directory.iterdir()) == [] @pytest.mark.parametrize('mutate', [ lambda row: row.update(checked_count=2), lambda row: row.update(eligible_count=0), lambda row: row.update(checked_count=True), lambda row: row.update(candidates=row['candidates']*6, checked_count=6, eligible_count=6), lambda row: row.update(note_id='https://private.invalid'), lambda row: row.update(status=['PRIVATE']), lambda row: row.update(discussion_state={'PRIVATE': 'QUOTE'}), lambda row: row.update(visible_comment_count=-1), lambda row: row.update(visible_comment_count=1_000_001), lambda row: row.update(visible_comment_count=True), lambda row: row.update(has_unexpanded_replies='PRIVATE'), lambda row: row['candidates'][0].update(classification=['PRIVATE']), lambda row: row['candidates'][0].update(eligible=False), lambda row: row['candidates'][0].update(observation_stable=False), lambda row: row['candidates'][0].update(native_navigation_confirmed=True), ]) def test_invalid_or_inconsistent_program_counts_never_persist(telemetry, mutate): task_id, directory = telemetry result = payload() mutate(result) assert record(task_id, result) is None assert 'preflight_observations' not in tools._task_artifacts[task_id] assert list(directory.iterdir()) == [] def test_receipt_projects_only_safe_program_fields_and_never_backfills_history(telemetry): task_id, directory = telemetry result = payload() result.update(rawhref='https://private.invalid/?key=PRIVATE', quote='PRIVATE', classification_counts={'PRIVATE': 99}) row = record(task_id, result) assert tools._valid_preflight_telemetry(row) assert len(tools._task_artifacts[task_id]['preflight_observations']) == 1 assert 'PRIVATE' not in json.dumps(row) and 'c'*24 not in json.dumps(row) assert 'PRIVATE' not in (directory/'evidence.json').read_text() def test_telemetry_history_is_bounded_and_exposes_truncation_without_a_business_quota(telemetry, monkeypatch): task_id, _ = telemetry writes = [] monkeypatch.setattr(tools, '_persist_evidence', lambda _: writes.append(len(tools._task_artifacts[task_id]['preflight_observations']))) first = record(task_id, payload()) for _ in range(tools._PREFLIGHT_TELEMETRY_LIMIT): assert record(task_id, payload()) rows = tools._task_artifacts[task_id]['preflight_observations'] assert len(rows) == tools._PREFLIGHT_TELEMETRY_LIMIT and first['id'] not in {row['id'] for row in rows} assert rows[-1]['history_truncated'] is True and max(writes) == tools._PREFLIGHT_TELEMETRY_LIMIT assert all(tools._valid_preflight_telemetry(row) for row in rows) def test_invalid_existing_telemetry_is_not_copied_into_a_new_receipt_or_rewritten(telemetry): task_id, directory = telemetry old = [{'quote': 'PRIVATE', 'rawhref': 'https://private.invalid'}] tools._task_artifacts[task_id]['preflight_observations'] = deepcopy(old) assert record(task_id, payload()) is None assert tools._task_artifacts[task_id]['preflight_observations'] == old assert list(directory.iterdir()) == []