import threading from pathlib import Path from unittest.mock import MagicMock, Mock, patch import pytest from test_config import settings from lai_automation.agent import AgentProviderError, AgentTimeoutError from lai_automation.git import ChangeSummary from lai_automation.models import FailureReport from lai_automation.worker import AutomationWorker, ValidationSetupError def test_dry_run_generates_task_without_agent_or_git() -> None: report = FailureReport( id=11, chat_message_id=1, source="automatic", category="sql_execution", user_feedback=None, feedback_type=None, expected_behavior=None, prompt="List locations", response=None, sql_query=None, error_message="Unknown column", context={}, ) worker = AutomationWorker(settings()) worker.reports = Mock() worker.reports.claim_next.return_value = report assert worker.run_once() worker.reports.release.assert_called_once() worker.agent = Mock() worker.agent.run.assert_not_called() def test_stop_signal_interrupts_worker_wait() -> None: worker = AutomationWorker(settings()) worker._stop(15, None) assert worker.stop_event.is_set() def test_pull_request_body_contains_redacted_original_report() -> None: report = FailureReport( id=7, chat_message_id=1, source="user_feedback", category="incorrect_answer", user_feedback="Use a table\ntoken=super-secret-token", feedback_type="wrong_result", expected_behavior="Show every supplied row", prompt="Count delivery notes by month", response="January: 38, February: 95", sql_query="SELECT month, COUNT(*) FROM deliveries GROUP BY month", error_message=None, context={}, ) summary = ChangeSummary(("app/Services/AIService.php",), 5, 1) body = AutomationWorker._pull_request_body( report, summary, [{"command": "php artisan test", "returncode": 0}], ) assert "## Original failure report" in body assert "**Source**\n> user_feedback" in body assert "**Category**\n> incorrect_answer" in body assert "**Feedback type**\n> wrong_result" in body assert "**User prompt**\n> Count delivery notes by month" in body assert "**Observed response**\n> January: 38, February: 95" in body assert "**User feedback**\n> Use a table\n> token=[REDACTED]" in body assert "**Expected behavior**\n> Show every supplied row" in body assert "**Error message**\n> _Not provided._" in body assert "**Generated SQL**\n> SELECT month, COUNT(*)" in body assert "super-secret-token" not in body assert "## Scope\n- `app/Services/AIService.php`" in body assert "## Validation\n- PASS `php artisan test`" in body def test_database_credentials_are_only_passed_to_test_process(tmp_path) -> None: worker = AutomationWorker(settings()) worker._prepare_test_schema = Mock() workspace = Mock() workspace.path = tmp_path workspace.run.return_value = Mock(returncode=0, stdout="ok", stderr="") results = worker._run_tests(workspace) assert results[0]["returncode"] == 0 environment = workspace.run.call_args.kwargs["extra_environment"] assert environment["AGENT_TEST_DB_ENABLED"] == "true" assert environment["APP_ENV"] == "testing" assert environment["AI_MISTRAL_API_KEY"] == "automation-test-key" assert environment["TEST_DB_USER"] == "test_runner" assert environment["TEST_DB_PASSWORD"] == "test-secret" assert environment["DB_DATABASE"] == "lai_agent_test" assert environment["ERP_DATABASE"] == "customer_agent_test" assert environment["COCKPIT_CONFIG_DATABASE"] == "cockpit_agent_test" def test_workspace_setup_creates_safe_env_and_prepares_assets(tmp_path) -> None: worker = AutomationWorker(settings( workspace_root=tmp_path, setup_commands=("npm run build",), )) worker._prepare_test_schema = Mock() workspace = Mock() workspace.path = tmp_path / "failure-1" workspace.path.mkdir() workspace.run.return_value = Mock(returncode=0, stdout="built", stderr="") results = worker._prepare_workspace_for_agent(workspace, 1) env_contents = (workspace.path / ".env").read_text() assert "APP_ENV=testing" in env_contents assert "DB_PASSWORD" not in env_contents assert results[0]["stage"] == "workspace setup" assert workspace.run.call_args.kwargs["extra_environment"]["DB_PASSWORD"] == "test-secret" def test_processing_connection_failure_is_not_sent_to_agent() -> None: report = FailureReport( id=15, chat_message_id=1, source="automatic", category="processing", user_feedback=None, feedback_type=None, expected_behavior=None, prompt="List customers", response=None, sql_query=None, error_message="Access denied for user 'runtime'@'host'", context={}, ) worker = AutomationWorker(settings(mode="pr", dry_run=False, gitea_token="token")) worker.reports = Mock() worker.reports.claim_next.return_value = report worker.agent = Mock() assert worker.run_once() worker.agent.run.assert_not_called() worker.reports.fail.assert_called_once() assert worker.reports.fail.call_args.args[2]["classification"] == "operational" def test_model_provider_failure_is_classified_as_operational() -> None: report = FailureReport( id=16, chat_message_id=1, source="user_feedback", category="incorrect_answer", user_feedback="Wrong result", feedback_type="wrong_result", expected_behavior="Use the correct rule", prompt="List records", response="0 rows", sql_query="SELECT 0", error_message=None, context={}, ) worker = AutomationWorker(settings(mode="pr", dry_run=False, gitea_token="token")) worker.reports = Mock() worker.reports.claim_next.return_value = report worker.agent = Mock() worker.agent.run.side_effect = AgentProviderError("Mistral DNS failed") worker._prepare_workspace_for_agent = Mock(return_value=[]) workspace = Mock() workspace.path = Path("/tmp/nonexistent-workspace") workspace.branch = "ai/failure-16" with patch("lai_automation.worker.GitWorkspace", return_value=workspace): assert worker.run_once() worker.reports.fail.assert_called_once() metadata = worker.reports.fail.call_args.args[2] assert metadata["classification"] == "operational" assert metadata["automation_action"] == "no_source_change" workspace.commit.assert_not_called() def test_test_schema_creates_minimal_erp_lookup(monkeypatch) -> None: worker = AutomationWorker(settings()) cursor = Mock() connection = MagicMock() connection.cursor.return_value.__enter__.return_value = cursor connect = Mock(return_value=connection) monkeypatch.setattr("lai_automation.worker.pymysql.connect", connect) worker._prepare_test_schema() assert connect.call_args.kwargs["database"] == "customer_agent_test" assert "CREATE TABLE IF NOT EXISTS FK_Nummerkreis" in cursor.execute.call_args.args[0] assert cursor.executemany.call_args.args[1][0] == (3, "Rechnung", "RG") connection.close.assert_called_once() def test_validation_commands_are_logged_without_output(caplog, tmp_path) -> None: worker = AutomationWorker(settings()) worker._prepare_test_schema = Mock() workspace = Mock() workspace.path = tmp_path workspace.run.return_value = Mock(returncode=0, stdout="sensitive output", stderr="") with caplog.at_level("INFO"): worker._run_tests(workspace, report_id=7, attempt=2) assert "Report 7 attempt 2 validation command: php artisan test starting" in caplog.text assert "returncode=0" in caplog.text assert "sensitive output" not in caplog.text def test_failed_validation_preserves_start_and_end_of_long_output(tmp_path) -> None: worker = AutomationWorker(settings()) worker._prepare_test_schema = Mock() workspace = Mock() workspace.path = tmp_path workspace.run.return_value = Mock( returncode=1, stdout="DECISIVE ASSERTION\n" + "x" * 13_000, stderr="FINAL ERROR", ) try: worker._run_tests(workspace) except RuntimeError as exception: failure = str(exception) else: raise AssertionError("Expected failed validation") assert "DECISIVE ASSERTION" in failure assert "...[middle omitted]..." in failure assert "FINAL ERROR" in failure def test_validation_signature_ignores_counts_timings_and_line_numbers() -> None: first = """ FAILED Tests\\Unit\\RouteTest > prefixes are removed Expectation failed for method determineRoute when invoked 1 time. Method was expected to be called 1 time, actually called 0 times. Tests: 1 failed, 50 passed (186 assertions) Duration: 0.49s at tests/Unit/RouteTest.php:229 """ second = """ FAILED Tests\\Unit\\RouteTest > prefixes are removed Expectation failed for method determineRoute when invoked 1 time. Method was expected to be called 1 time, actually called 0 times. Tests: 1 failed, 50 passed (191 assertions) Duration: 0.50s at tests/Unit/RouteTest.php:269 """ assert AutomationWorker._validation_signature(first) == AutomationWorker._validation_signature( second ) def test_validation_setup_failure_is_typed_and_not_a_test_failure(tmp_path) -> None: worker = AutomationWorker(settings(setup_commands=("npm ci",))) worker._prepare_test_schema = Mock() workspace = Mock() workspace.path = tmp_path workspace.run.return_value = Mock( returncode=1, stdout="network unavailable", stderr="", ) with pytest.raises(ValidationSetupError, match="Validation setup command failed"): worker._run_tests(workspace) def test_long_stage_emits_heartbeat(caplog) -> None: worker = AutomationWorker(settings()) with caplog.at_level("INFO"), worker._progress(7, "agent attempt 1", interval=0.001): threading.Event().wait(0.01) assert "Report 7 agent attempt 1 still running" in caplog.text assert "Report 7 agent attempt 1 finished" in caplog.text def test_long_stage_refreshes_database_claim_lease() -> None: worker = AutomationWorker(settings(mode="pr", dry_run=False, gitea_token="token")) worker.reports = Mock() with worker._progress(7, "agent attempt 1", interval=0.001): threading.Event().wait(0.01) assert worker.reports.heartbeat.call_count >= 2 def test_timed_out_agent_partial_changes_are_validated_and_published() -> None: report = FailureReport( id=4, chat_message_id=1, source="user_feedback", category="incorrect_answer", user_feedback="Wrong result", feedback_type="wrong_result", expected_behavior="Use System_created", prompt="Delivery notes by month", response="0 rows", sql_query="SELECT 0", error_message=None, context={"database_type": "erp"}, ) worker = AutomationWorker(settings(mode="pr", dry_run=False, gitea_token="token")) worker.reports = Mock() worker.reports.claim_next.return_value = report worker.agent = Mock() worker.agent.run.side_effect = AgentTimeoutError("timed out after applying edits") worker.gitea = Mock() worker._prepare_workspace_for_agent = Mock(return_value=[]) worker._run_tests = Mock(return_value=[]) workspace = Mock() workspace.path = Path("/tmp/nonexistent-workspace") workspace.branch = "ai/failure-4" workspace.summary.return_value = ChangeSummary( ("config/erp_semantics.php", "tests/Unit/SQLGeneratorBusinessContextTest.php"), 20, 2, ) workspace.added_text.return_value = ( "Use DATE_FORMAT(System_created, '%Y-%m') and AuftragsKennung = 2; " "COALESCE(obsolet, 0) = 0 and COALESCE(bStatus_storniert, 0) = 0" ) workspace.run_args.return_value = Mock( returncode=0, stdout="No syntax errors", stderr="", ) workspace.commit.return_value = "commit-id" with patch("lai_automation.worker.GitWorkspace", return_value=workspace): assert worker.run_once() worker._run_tests.assert_called_once() workspace.push_branch.assert_called_once() worker.reports.complete.assert_called_once() def test_provider_failure_partial_changes_are_validated_and_published() -> None: report = FailureReport( id=4, chat_message_id=1, source="user_feedback", category="incorrect_answer", user_feedback="Wrong result", feedback_type="wrong_result", expected_behavior="Use System_created", prompt="Delivery notes by month", response="0 rows", sql_query="SELECT 0", error_message=None, context={"database_type": "erp"}, ) worker = AutomationWorker(settings(mode="pr", dry_run=False, gitea_token="token")) worker.reports = Mock() worker.reports.claim_next.return_value = report worker.agent = Mock() worker.agent.run.side_effect = AgentProviderError("provider failed after edits") worker.gitea = Mock() worker._prepare_workspace_for_agent = Mock(return_value=[]) worker._run_tests = Mock(return_value=[]) workspace = Mock() workspace.path = Path("/tmp/nonexistent-workspace") workspace.branch = "ai/failure-4" workspace.summary.return_value = ChangeSummary( ("config/erp_semantics.php", "tests/Unit/SQLGeneratorBusinessContextTest.php"), 20, 2, ) workspace.added_text.return_value = ( "Use DATE_FORMAT(System_created, '%Y-%m') and AuftragsKennung = 2; " "COALESCE(obsolet, 0) = 0 and COALESCE(bStatus_storniert, 0) = 0" ) workspace.run_args.return_value = Mock( returncode=0, stdout="No syntax errors", stderr="", ) workspace.commit.return_value = "commit-id" with patch("lai_automation.worker.GitWorkspace", return_value=workspace): assert worker.run_once() worker._run_tests.assert_called_once() workspace.push_branch.assert_called_once() worker.reports.complete.assert_called_once() def test_php_syntax_failure_discards_attempt_before_full_validation() -> None: report = FailureReport( id=56, chat_message_id=1, source="user_feedback", category="incorrect_answer", user_feedback="Wrong hours", feedback_type="wrong_result", expected_behavior="Use time registration", prompt="Hours in April", response="0 hours", sql_query="SELECT 0", error_message=None, context={}, ) config = settings(mode="pr", dry_run=False, gitea_token="token") worker = AutomationWorker(config) worker.reports = Mock() worker.reports.claim_next.return_value = report worker.agent = Mock() worker.agent.run.side_effect = ["broken PHP", "fixed PHP"] worker.gitea = Mock() worker._run_tests = Mock(return_value=[]) worker._prepare_workspace_for_agent = Mock(return_value=[]) workspace = Mock() workspace.path = Path("/tmp/nonexistent-workspace") workspace.branch = "ai/failure-56" workspace.summary.return_value = ChangeSummary( ("config/erp_semantics.php", "tests/Unit/FixTest.php"), 20, 2 ) workspace.run_args.side_effect = [ Mock(returncode=255, stdout="Parse error", stderr="unexpected token"), Mock(returncode=0, stdout="No syntax errors", stderr=""), Mock(returncode=0, stdout="No syntax errors", stderr=""), ] workspace.diff.return_value = "--- app/Service.php\n+++ app/Service.php\n+intended fix" workspace.checkpoint.return_value = "valid-checkpoint" workspace.commit.return_value = "commit-id" with patch("lai_automation.worker.GitWorkspace", return_value=workspace): assert worker.run_once() assert worker.agent.run.call_count == 2 assert "unexpected token" in worker.agent.run.call_args_list[1].args[1] assert "clean repository" in worker.agent.run.call_args_list[1].args[1] assert "intended fix" in worker.agent.run.call_args_list[1].args[1] assert "repair-only attempt" in worker.agent.run.call_args_list[1].args[1] workspace.restore_checkpoint.assert_called_once_with(None) worker._run_tests.assert_called_once() def test_malformed_retry_restores_last_syntax_valid_attempt() -> None: report = FailureReport( id=56, chat_message_id=1, source="user_feedback", category="incorrect_answer", user_feedback="Wrong hours", feedback_type="wrong_result", expected_behavior="Use time registration", prompt="Hours in April", response="0 hours", sql_query="SELECT 0", error_message=None, context={}, ) config = settings(mode="pr", dry_run=False, gitea_token="token", max_agent_attempts=3) worker = AutomationWorker(config) worker.reports = Mock() worker.reports.claim_next.return_value = report worker.agent = Mock() worker.agent.run.side_effect = ["valid but failing", "broken retry", "corrected retry"] worker.gitea = Mock() worker._run_tests = Mock(side_effect=[RuntimeError("Expected hour rule"), []]) worker._prepare_workspace_for_agent = Mock(return_value=[]) workspace = Mock() workspace.path = Path("/tmp/nonexistent-workspace") workspace.branch = "ai/failure-56" workspace.summary.return_value = ChangeSummary( ("config/erp_semantics.php", "tests/Unit/FixTest.php"), 20, 2 ) workspace.run_args.side_effect = [ Mock(returncode=0, stdout="No syntax errors", stderr=""), Mock(returncode=0, stdout="No syntax errors", stderr=""), Mock(returncode=255, stdout="Parse error", stderr="unexpected token"), Mock(returncode=0, stdout="No syntax errors", stderr=""), Mock(returncode=0, stdout="No syntax errors", stderr=""), ] workspace.diff.return_value = "--- app/Service.php\n+++ app/Service.php\n+application repair" workspace.checkpoint.side_effect = ["checkpoint-1", "checkpoint-3"] workspace.commit.return_value = "commit-id" with patch("lai_automation.worker.GitWorkspace", return_value=workspace): assert worker.run_once() assert worker.agent.run.call_count == 3 workspace.restore_checkpoint.assert_called_once_with("checkpoint-1") assert "last syntax-valid changes were restored" in worker.agent.run.call_args_list[2].args[1] assert "application repair" in worker.agent.run.call_args_list[2].args[1] assert worker._run_tests.call_count == 2 def test_failed_validation_gets_one_correction_before_pr() -> None: report = FailureReport( id=13, chat_message_id=1, source="automatic", category="sql_execution", user_feedback=None, feedback_type=None, expected_behavior=None, prompt="List products", response=None, sql_query="SELECT missing_column FROM invoice_row", error_message="Unknown column", context={}, ) config = settings(mode="pr", dry_run=False, gitea_token="token") worker = AutomationWorker(config) worker.reports = Mock() worker.reports.claim_next.return_value = report worker.agent = Mock() worker.agent.run.side_effect = ["first change", "corrected change"] worker.gitea = Mock() worker._run_tests = Mock(side_effect=[RuntimeError("Missing import"), []]) worker._prepare_workspace_for_agent = Mock(return_value=[]) workspace = Mock() workspace.path = Path("/tmp/nonexistent-workspace") workspace.branch = "ai/failure-13" workspace.summary.return_value = ChangeSummary( ("app/Services/Fix.php", "tests/Feature/FixTest.php"), 10, 1 ) workspace.run_args.return_value = Mock(returncode=0, stdout="No syntax errors", stderr="") workspace.commit.return_value = "commit-id" with patch("lai_automation.worker.GitWorkspace", return_value=workspace): assert worker.run_once() assert worker.agent.run.call_count == 2 assert "Missing import" in worker.agent.run.call_args_list[1].args[1] assert worker._run_tests.call_count == 2 workspace.push_branch.assert_called_once() worker.reports.complete.assert_called_once() def test_failed_validation_retry_keeps_full_routed_context() -> None: report = FailureReport( id=4, chat_message_id=1, source="user_feedback", category="incorrect_answer", user_feedback="Wrong result", feedback_type="wrong_result", expected_behavior="AuftragsKennung 2 verwenden", prompt="Delivery notes by month", response=None, sql_query="SELECT Datum_erfassung", error_message=None, context={"database_type": "erp"}, ) worker = AutomationWorker(settings(mode="pr", dry_run=False, gitea_token="token")) worker.reports = Mock() worker.reports.claim_next.return_value = report worker.agent = Mock() worker.agent.run.side_effect = ["valid production with bad test", "fixed assertion"] worker.gitea = Mock() worker._prepare_workspace_for_agent = Mock(return_value=[]) worker._run_tests = Mock(side_effect=[RuntimeError("case-sensitive assertion failed"), []]) workspace = Mock() workspace.path = Path("/tmp/nonexistent-workspace") workspace.branch = "ai/failure-4" workspace.summary.return_value = ChangeSummary( ("config/erp_semantics.php", "tests/Unit/SQLGeneratorBusinessContextTest.php"), 20, 1, ) added = ( "AuftragsKennung = 2\n" "SELECT * FROM FK_Auftrag WHERE AuftragsKennung = 2" ) workspace.added_text.return_value = added workspace.run_args.return_value = Mock(returncode=0, stdout="No syntax errors", stderr="") workspace.checkpoint.return_value = "valid-production" workspace.commit.return_value = "commit-id" with patch("lai_automation.worker.GitWorkspace", return_value=workspace): assert worker.run_once() assert worker.agent.run.call_args_list[1].args[2] == ( "config/erp_semantics.php", "tests/Unit/SQLGeneratorBusinessContextTest.php", ) retry_task = worker.agent.run.call_args_list[1].args[1] assert "inspect every routed production and test file" in retry_task assert "Do not merely rewrite assertions" in retry_task assert "case-sensitive" in worker.agent.run.call_args_list[1].args[1] attempts = worker.reports.complete.call_args.args[1]["agent_attempts"] assert attempts[0]["validation_status"] == "failed" assert "case-sensitive assertion failed" in attempts[0]["validation_error_tail"] assert attempts[1]["validation_status"] == "passed" def test_no_change_response_gets_one_correction() -> None: report = FailureReport( id=14, chat_message_id=1, source="automatic", category="sql_execution", user_feedback=None, feedback_type=None, expected_behavior=None, prompt="List products", response=None, sql_query=None, error_message="Unknown column", context={}, ) config = settings(mode="pr", dry_run=False, gitea_token="token") worker = AutomationWorker(config) worker.reports = Mock() worker.reports.claim_next.return_value = report worker.agent = Mock() worker.agent.run.side_effect = ["no edit", "applied edit"] worker.gitea = Mock() worker._run_tests = Mock(return_value=[]) worker._prepare_workspace_for_agent = Mock(return_value=[]) workspace = Mock() workspace.path = Path("/tmp/nonexistent-workspace") workspace.branch = "ai/failure-14" workspace.summary.side_effect = [ ChangeSummary((), 0, 0), ChangeSummary(("app/Services/Fix.php", "tests/Feature/FixTest.php"), 10, 1), ChangeSummary(("app/Services/Fix.php", "tests/Feature/FixTest.php"), 10, 1), ] workspace.run_args.return_value = Mock(returncode=0, stdout="No syntax errors", stderr="") workspace.commit.return_value = "commit-id" with patch("lai_automation.worker.GitWorkspace", return_value=workspace): assert worker.run_once() assert worker.agent.run.call_count == 2 assert "Agent produced no file changes" in worker.agent.run.call_args_list[1].args[1] worker._run_tests.assert_called_once() workspace.push_branch.assert_called_once() def test_test_only_change_gets_correction_before_validation() -> None: report = FailureReport( id=49, chat_message_id=1, source="user_feedback", category="incorrect_answer", user_feedback="Token was not restored", feedback_type="wrong_result", expected_behavior="Return the real value", prompt="Average price", response="__LAI_UNKNOWN__", sql_query="SELECT AVG(price)", error_message=None, context={}, ) config = settings(mode="pr", dry_run=False, gitea_token="token") worker = AutomationWorker(config) worker.reports = Mock() worker.reports.claim_next.return_value = report worker.agent = Mock() worker.agent.run.side_effect = ["tests only", "production fix"] worker.gitea = Mock() worker._run_tests = Mock(return_value=[]) worker._prepare_workspace_for_agent = Mock(return_value=[]) workspace = Mock() workspace.path = Path("/tmp/nonexistent-workspace") workspace.branch = "ai/failure-49" workspace.summary.side_effect = [ ChangeSummary(("tests/Feature/AIPrivacyBoundaryTest.php",), 8, 0), ChangeSummary( ("app/Services/AIService.php", "tests/Feature/AIPrivacyBoundaryTest.php"), 15, 1, ), ChangeSummary( ("app/Services/AIService.php", "tests/Feature/AIPrivacyBoundaryTest.php"), 15, 1, ), ] workspace.run_args.return_value = Mock(returncode=0, stdout="No syntax errors", stderr="") workspace.commit.return_value = "commit-id" with patch("lai_automation.worker.GitWorkspace", return_value=workspace): assert worker.run_once() assert worker.agent.run.call_count == 2 assert "changed only tests" in worker.agent.run.call_args_list[1].args[1] assert "exact repository-relative paths" in worker.agent.run.call_args_list[1].args[1] workspace.checkpoint.assert_called_once() worker._run_tests.assert_called_once() def test_semantically_invalid_change_is_not_kept_as_retry_checkpoint() -> None: report = FailureReport( id=4, chat_message_id=1, source="user_feedback", category="incorrect_answer", user_feedback="Wrong result", feedback_type="wrong_result", expected_behavior="System_created verwenden und obsolet ausfiltern", prompt="Delivery notes by month", response="0 rows", sql_query="SELECT Datum_erfassung", error_message=None, context={"database_type": "erp"}, ) worker = AutomationWorker(settings(mode="pr", dry_run=False, gitea_token="token")) worker.reports = Mock() worker.reports.claim_next.return_value = report worker.agent = Mock() worker.agent.run.side_effect = ["missing filter", "correct repair"] worker.gitea = Mock() worker._prepare_workspace_for_agent = Mock(return_value=[]) worker._run_tests = Mock(return_value=[]) workspace = Mock() workspace.path = Path("/tmp/nonexistent-workspace") workspace.branch = "ai/failure-4" workspace.summary.return_value = ChangeSummary( ("config/erp_semantics.php", "tests/Unit/SQLGeneratorBusinessContextTest.php"), 12, 1, ) workspace.added_text.side_effect = [ "System_created", "System_created AND COALESCE(obsolet, 0) = 0", "System_created AND COALESCE(obsolet, 0) = 0", ] workspace.run_args.return_value = Mock(returncode=0, stdout="No syntax errors", stderr="") workspace.checkpoint.return_value = "policy-valid-checkpoint" workspace.commit.return_value = "commit-id" with patch("lai_automation.worker.GitWorkspace", return_value=workspace): assert worker.run_once() workspace.restore_checkpoint.assert_called_once_with(None) workspace.checkpoint.assert_called_once() worker._run_tests.assert_called_once() def test_policy_constraints_accumulate_across_retries() -> None: report = FailureReport( id=4, chat_message_id=1, source="user_feedback", category="incorrect_answer", user_feedback="Wrong result", feedback_type="wrong_result", expected_behavior="AuftragsKennung 2 und System_created verwenden; obsolet ausfiltern", prompt="Delivery notes by month", response=None, sql_query="SELECT * WHERE Bezeichnung_Kurz = 'LS'", error_message=None, context={"database_type": "erp"}, ) worker = AutomationWorker(settings( mode="pr", dry_run=False, gitea_token="token", max_agent_attempts=3 )) worker.reports = Mock() worker.reports.claim_next.return_value = report worker.agent = Mock() worker.agent.run.side_effect = ["incomplete", "test only", "complete"] worker.gitea = Mock() worker._prepare_workspace_for_agent = Mock(return_value=[]) worker._run_tests = Mock(return_value=[]) workspace = Mock() workspace.path = Path("/tmp/nonexistent-workspace") workspace.branch = "ai/failure-4" workspace.summary.side_effect = [ ChangeSummary(("config/erp_semantics.php",), 5, 0), ChangeSummary(("tests/Unit/SQLGeneratorBusinessContextTest.php",), 5, 0), ChangeSummary( ("config/erp_semantics.php", "tests/Unit/SQLGeneratorBusinessContextTest.php"), 20, 1, ), ChangeSummary( ("config/erp_semantics.php", "tests/Unit/SQLGeneratorBusinessContextTest.php"), 20, 1, ), ] complete = ( "AuftragsKennung = 2 System_created COALESCE(obsolet, 0) = 0\n" "SELECT * FROM FK_Auftrag WHERE AuftragsKennung = 2 " "AND System_created IS NOT NULL AND COALESCE(obsolet, 0) = 0" ) workspace.added_text.side_effect = ["AuftragsKennung = 2", "test", complete, complete] workspace.run_args.return_value = Mock(returncode=0, stdout="No syntax errors", stderr="") workspace.checkpoint.return_value = "valid" workspace.commit.return_value = "commit-id" with patch("lai_automation.worker.GitWorkspace", return_value=workspace): assert worker.run_once() third_task = worker.agent.run.call_args_list[2].args[1] assert "All policy constraints from earlier attempts remain mandatory" in third_task assert "one reference SELECT" in third_task assert "changed only tests" in third_task def test_already_fixed_removal_report_skips_agent_and_dependency_setup(tmp_path) -> None: report = FailureReport( id=5, chat_message_id=1, source="user_feedback", category="incorrect_answer", user_feedback="Wrong delivery note count", feedback_type="wrong_result", expected_behavior=( 'Es muss " a.bStatus_gebucht = 1 " entfernt werden. ' "Einfach nur das aus der SQL entfernen" ), prompt="Gib mir eine Anzahl von Lieferscheinen pro Monat in 2024", response="0 rows", sql_query=( "SELECT COUNT(*) FROM FK_Auftrag a WHERE a.AuftragsKennung = 2 " "AND a.bStatus_gebucht = 1" ), error_message=None, context={"database_type": "erp"}, ) source = tmp_path / "config/erp_semantics.php" source.parent.mkdir(parents=True) source.write_text(""" [[ 'question' => 'Gib mir eine Anzahl von Lieferscheinen pro Monat in 2024', 'sql' => <<<'SQL' SELECT COUNT(*) FROM FK_Auftrag a WHERE a.AuftragsKennung = 2; SQL, ]]]; """) worker = AutomationWorker(settings(mode="pr", dry_run=False, gitea_token="token")) worker.reports = Mock() worker.reports.claim_next.return_value = report worker.agent = Mock() worker.gitea = Mock() worker._prepare_workspace_for_agent = Mock() workspace = Mock() workspace.path = tmp_path workspace.branch = "ai/failure-5" with patch("lai_automation.worker.GitWorkspace", return_value=workspace): assert worker.run_once() workspace.prepare.assert_called_once() worker._prepare_workspace_for_agent.assert_not_called() worker.agent.run.assert_not_called() workspace.commit.assert_not_called() workspace.push_branch.assert_not_called() worker.gitea.create_pull_request.assert_not_called() worker.reports.complete.assert_called_once() metadata = worker.reports.complete.call_args.args[1] assert metadata["classification"] == "already_fixed" assert metadata["automation_action"] == "no_source_change" assert metadata["forbidden_fragments"] == ["a.bStatus_gebucht = 1"] def test_identical_policy_failure_stops_after_two_attempts() -> None: report = FailureReport( id=18, chat_message_id=1, source="automatic", category="sql_execution", user_feedback=None, feedback_type=None, expected_behavior=None, prompt="List generic records", response=None, sql_query="SELECT missing FROM records", error_message="Unknown column", context={}, ) worker = AutomationWorker(settings( mode="pr", dry_run=False, gitea_token="token", max_agent_attempts=5, )) worker.reports = Mock() worker.reports.claim_next.return_value = report worker.agent = Mock() worker.agent.run.side_effect = ["tests only", "still tests only", "must not run"] worker._prepare_workspace_for_agent = Mock(return_value=[]) workspace = Mock() workspace.path = Path("/tmp/nonexistent-workspace") workspace.branch = "ai/failure-18" workspace.summary.side_effect = [ ChangeSummary(("tests/Unit/FixTest.php",), 2, 0), ChangeSummary(("tests/Unit/FixTest.php",), 3, 0), ] workspace.added_text.return_value = "" workspace.diff.return_value = "same rejected candidate" workspace.run_args.return_value = Mock( returncode=0, stdout="No syntax errors", stderr="", ) with patch("lai_automation.worker.GitWorkspace", return_value=workspace): assert worker.run_once() assert worker.agent.run.call_count == 2 workspace.commit.assert_not_called() worker.reports.fail.assert_called_once() assert "Repeated identical candidate rejection" in worker.reports.fail.call_args.args[1] def test_same_policy_error_with_changed_candidate_can_progress() -> None: report = FailureReport( id=18, chat_message_id=1, source="automatic", category="sql_execution", user_feedback=None, feedback_type=None, expected_behavior=None, prompt="List generic records", response=None, sql_query="SELECT missing FROM records", error_message="Unknown column", context={}, ) worker = AutomationWorker(settings( mode="pr", dry_run=False, gitea_token="token", max_agent_attempts=3, )) worker.reports = Mock() worker.reports.claim_next.return_value = report worker.agent = Mock() worker.agent.run.side_effect = ["first tests only", "improved tests only", "complete fix"] worker.gitea = Mock() worker._prepare_workspace_for_agent = Mock(return_value=[]) worker._run_tests = Mock(return_value=[]) workspace = Mock() workspace.path = Path("/tmp/nonexistent-workspace") workspace.branch = "ai/failure-18" workspace.summary.side_effect = [ ChangeSummary(("tests/Unit/FixTest.php",), 2, 0), ChangeSummary(("tests/Unit/FixTest.php",), 3, 0), ChangeSummary(("app/Services/Fix.php", "tests/Unit/FixTest.php"), 8, 1), ChangeSummary(("app/Services/Fix.php", "tests/Unit/FixTest.php"), 8, 1), ] workspace.added_text.return_value = "" workspace.diff.side_effect = ["first rejected candidate", "second rejected candidate"] workspace.run_args.return_value = Mock( returncode=0, stdout="No syntax errors", stderr="", ) workspace.checkpoint.return_value = "valid" workspace.commit.return_value = "commit-id" with patch("lai_automation.worker.GitWorkspace", return_value=workspace): assert worker.run_once() assert worker.agent.run.call_count == 3 worker._run_tests.assert_called_once() workspace.push_branch.assert_called_once() worker.reports.complete.assert_called_once()