210 lines
9.9 KiB
Python
210 lines
9.9 KiB
Python
"""End-to-end invariants from the September 11 live audit."""
|
|
import json
|
|
import multiprocessing
|
|
import os
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
import state_store
|
|
import trigger
|
|
import workflow
|
|
import worker
|
|
import completion_watcher
|
|
from types import SimpleNamespace
|
|
|
|
|
|
@pytest.fixture
|
|
def board(tmp_path, monkeypatch):
|
|
for name in ('intent', 'research', 'inbox', 'processing', 'builder', 'builder-processing'):
|
|
(tmp_path / name).mkdir()
|
|
for name, value in {
|
|
'BOARD_STATE_FILE': str(tmp_path / 'state.json'),
|
|
'BOARD_INTENT_DIR': str(tmp_path / 'intent'),
|
|
'BOARD_RESEARCH_INTENT_DIR': str(tmp_path / 'research'),
|
|
'BOARD_INBOX_DIR': str(tmp_path / 'inbox'),
|
|
'BOARD_PROCESSING_DIR': str(tmp_path / 'processing'),
|
|
'AGENT_BUILDER_QUEUE_DIR': str(tmp_path / 'builder'),
|
|
'AGENT_BUILDER_PROCESSING_DIR': str(tmp_path / 'builder-processing'),
|
|
}.items():
|
|
monkeypatch.setenv(name, value)
|
|
(tmp_path / 'intent' / 'goal.md').write_text('# Goal\n')
|
|
trigger.save_state({'cards': {}})
|
|
return tmp_path
|
|
|
|
|
|
@pytest.mark.parametrize('phase', ['deployed', 'bereit_fuer_deployment', 'recherche_fertig'])
|
|
def test_manual_moves_cannot_claim_verified_outcomes(board, phase):
|
|
with pytest.raises(PermissionError):
|
|
trigger.update_phase('goal', phase)
|
|
assert not workflow.verified_deployment({'phase': 'deployed'})
|
|
assert trigger.deployment_evidence({'phase': 'deployed'})['status'] == 'unknown'
|
|
|
|
|
|
def test_failed_inspector_can_archive_and_reactivate(board):
|
|
trigger.save_state({'cards': {'goal': {'phase': 'inspector_fehlgeschlagen', 'last_run_status': 'failed'}}})
|
|
assert 'archive' in [a['id'] for a in trigger.available_actions('goal', 'intent', trigger.load_state()['cards']['goal'])]
|
|
trigger.update_phase('goal', 'archiviert')
|
|
trigger.update_phase('goal', 'geplant')
|
|
assert trigger.load_state()['cards']['goal']['phase'] == 'geplant'
|
|
|
|
|
|
def test_research_reset_returns_without_partial_failure(board):
|
|
(board / 'research' / 'study.md').write_text('# Study\n')
|
|
trigger.save_state({'cards': {'study': {'source': 'research', 'phase': 'recherche_fehlgeschlagen'}}})
|
|
assert trigger.update_phase('study', 'recherche_angefordert') is None
|
|
assert trigger.load_state()['cards']['study']['phase'] == 'recherche_angefordert'
|
|
|
|
|
|
def test_downstream_builder_is_visible_to_all_trigger_guards(board):
|
|
(board / 'builder-processing' / 'job.json').write_text(json.dumps({'board_slug': 'goal'}))
|
|
assert trigger.builder_trigger_blocked('goal')
|
|
with pytest.raises(PermissionError):
|
|
trigger.update_phase('goal', 'archiviert')
|
|
|
|
|
|
def test_routing_to_builder_queues_real_job(board):
|
|
trigger.save_state({'cards': {'goal': {'phase': 'inspector_fehlgeschlagen', 'routing_pending': True,
|
|
'routing_recommendation': 'in_umsetzung'}}})
|
|
assert trigger.apply_routing_recommendation('goal') == 'in_umsetzung'
|
|
assert len(list((board / 'inbox').glob('*.json'))) == 1
|
|
assert trigger.load_state()['cards']['goal']['last_run_status'] == 'queued'
|
|
|
|
|
|
def _increment(path, count):
|
|
os.environ['BOARD_STATE_FILE'] = path
|
|
for _ in range(count):
|
|
with state_store.transaction():
|
|
state = state_store.load()
|
|
state['count'] = state.get('count', 0) + 1
|
|
state_store.save(state)
|
|
|
|
|
|
def test_concurrent_processes_do_not_lose_updates(board):
|
|
context = multiprocessing.get_context('fork')
|
|
processes = [context.Process(target=_increment, args=(str(board / 'state.json'), 20)) for _ in range(3)]
|
|
for process in processes:
|
|
process.start()
|
|
for process in processes:
|
|
process.join(10)
|
|
assert process.exitcode == 0
|
|
assert state_store.load()['count'] == 60
|
|
|
|
|
|
def test_corrupt_state_fails_closed(board):
|
|
(board / 'state.json').write_text('{broken')
|
|
with pytest.raises(ValueError):
|
|
trigger.create_idea('Do not erase state')
|
|
assert (board / 'state.json').read_text() == '{broken'
|
|
|
|
|
|
def test_claimed_job_is_not_republished_after_outbox_crash(board):
|
|
target = board / 'inbox' / 'durable-job.json'
|
|
state_store.atomic_json(target, {'job_id': 'durable-job'})
|
|
assert state_store.claim_job(target, board / 'processing' / target.name)
|
|
state_store.atomic_json(board / 'state.json', {'cards': {}, '_pending_writes': {str(target): {'job_id': 'durable-job'}}})
|
|
with state_store.transaction():
|
|
pass
|
|
assert not target.exists()
|
|
assert (board / 'processing' / target.name).exists()
|
|
|
|
|
|
def test_changed_commit_invalidates_old_activation_receipt(board):
|
|
trigger.save_state({'cards': {'goal': {'commit_hash': 'a' * 40, 'deployment_status': 'deployed',
|
|
'deployed_commit_hash': 'a' * 40, 'activated_system': '/nix/store/old'}}})
|
|
worker.update_state('goal', 'done', 'review-new', commit_hash='b' * 40)
|
|
assert not workflow.verified_deployment(trigger.load_state()['cards']['goal'])
|
|
|
|
|
|
@pytest.mark.parametrize('verdict', [None, {'schema_version': 1, 'commit_hash': 'badbeef', 'decision': 'approve'},
|
|
{'schema_version': True, 'commit_hash': 'abc1234', 'decision': 'approve'}])
|
|
def test_invalid_review_never_replaces_previous_report(board, monkeypatch, verdict):
|
|
from test_board_agent import RunReviewTests
|
|
for key, value in RunReviewTests()._environment(str(board)).items():
|
|
monkeypatch.setenv(key, value)
|
|
report_dir = board / 'docs' / 'review'
|
|
report_dir.mkdir(parents=True)
|
|
report = report_dir / 'goal.md'
|
|
report.write_text('Previous verified report')
|
|
response = '===REVIEW-REPORT===\nNot ready for deployment. Decision: back to planning.\n===END-REVIEW-REPORT==='
|
|
if verdict is not None:
|
|
response += '\n===REVIEW-RESULT===\n' + json.dumps(verdict) + '\n===END-REVIEW-RESULT==='
|
|
output = json.dumps({'type': 'item.completed', 'item': {'type': 'agent_message', 'text': response}})
|
|
with patch.object(worker.subprocess, 'run', return_value=SimpleNamespace(returncode=0, stdout=output, stderr='')):
|
|
with pytest.raises(RuntimeError):
|
|
worker.run_review('goal', 'abc1234', 'codex')
|
|
assert report.read_text() == 'Previous verified report'
|
|
|
|
|
|
def test_deployment_retry_reuses_review_without_model(board, monkeypatch):
|
|
monkeypatch.setenv('AGENT_BUILDER_DEPLOY_QUEUE_DIR', str(board / 'deploy'))
|
|
monkeypatch.setenv('BOARD_REPO_DIR', str(board))
|
|
monkeypatch.setenv('BOARD_BUILDER_RESULTS_DIR', str(board / 'results'))
|
|
trigger.save_state({'cards': {'goal': {'source': 'intent', 'phase': 'fixer_fehlgeschlagen',
|
|
'failure_stage': 'deployment', 'review_decision': 'bereit_fuer_deployment', 'commit_hash': 'abc1234'}}})
|
|
job_id = trigger.retry_deployment('goal')
|
|
with patch.object(worker, 'run_review', side_effect=AssertionError('No new model run')):
|
|
worker.process_job(board / 'inbox' / (job_id + '.json'), board / 'failed', board / 'done')
|
|
request = json.loads(next((board / 'deploy').glob('*.json')).read_text())
|
|
assert request['commit_hash'] == 'abc1234'
|
|
assert trigger.load_state()['cards']['goal']['deployment_status'] == 'queued'
|
|
assert not workflow.verified_deployment(trigger.load_state()['cards']['goal'])
|
|
|
|
|
|
def test_archived_parent_does_not_restart_children(board):
|
|
import orchestration
|
|
trigger.save_state({'cards': {'parent': {'phase': 'archiviert', 'orchestration_kind': 'operator', 'depends_on': ['goal']},
|
|
'goal': {'phase': 'geplant', 'parent_id': 'parent', 'operator_generated': True}}})
|
|
with patch.object(trigger, 'write_job', side_effect=AssertionError('Archived family must remain inactive')):
|
|
orchestration.progress_all(trigger.write_job, trigger.COLUMN_CONFIG_DEFAULTS)
|
|
assert trigger.load_state()['cards']['parent']['phase'] == 'archiviert'
|
|
|
|
|
|
def test_outbox_publishes_only_after_state_commit(board):
|
|
target = board / 'inbox' / 'unique.json'
|
|
with state_store.transaction():
|
|
state = state_store.load()
|
|
state['cards']['goal'] = {'last_run_status': 'queued'}
|
|
state_store.save(state)
|
|
state_store.write_payload(target, {'job_id': 'unique'})
|
|
assert not target.exists()
|
|
assert target.exists()
|
|
assert state_store.load()['cards']['goal']['last_run_status'] == 'queued'
|
|
|
|
|
|
def test_transaction_rollback_does_not_publish_job(board):
|
|
target = board / 'inbox' / 'rollback.json'
|
|
with pytest.raises(RuntimeError):
|
|
with state_store.transaction():
|
|
state_store.write_payload(target, {'job_id': 'rollback'})
|
|
raise RuntimeError('aborted')
|
|
assert not target.exists()
|
|
|
|
|
|
def test_completion_is_idempotent_and_stale_pipeline_is_ignored(board):
|
|
record = board / 'record.json'
|
|
record.write_text(json.dumps({'job_id':'board-goal-1', 'board_slug':'goal', 'status':'no_changes', 'pipeline_run_id':'old'}))
|
|
trigger.save_state({'cards': {'goal': {'phase':'geplant', 'pipeline_run_id':'new'}}})
|
|
completion_watcher.process_record(record)
|
|
completion_watcher.process_record(record)
|
|
assert trigger.load_state()['cards']['goal']['phase'] == 'geplant'
|
|
assert len(trigger.load_state()['processed_completions']) == 1
|
|
|
|
|
|
def test_blocked_children_override_finished_operator_job():
|
|
child = {'id':'child', 'phase':'in_umsetzung', 'last_run_status':'no_changes'}
|
|
display = workflow.presentation({'phase':'wartet_auf_kinder', 'last_run_status':'done'}, [child])
|
|
assert display['group'] == 'attention'
|
|
assert display['label'] == 'Blocked by children'
|
|
assert display['blocking_children'] == ['child']
|
|
|
|
|
|
def test_negative_review_and_not_applicable_are_not_success():
|
|
assert workflow.presentation({'phase':'planung_mit_feedback', 'last_run_status':'done'})['group'] == 'attention'
|
|
assert workflow.presentation({'phase':'klaerung_reparatur', 'last_run_status':'not_applicable'})['group'] == 'attention'
|
|
|
|
|
|
def test_idea_refinement_preserves_identity(board):
|
|
card_id = trigger.create_idea('A useful goal')
|
|
assert trigger.plan_card(card_id) == card_id
|
|
assert trigger.load_state()['cards'][card_id]['source'] == 'draft'
|