Isolate cross-database repair validation

src/lai_automation/policy.py

@@ -5,6 +5,7 @@ from pathlib import Path, PurePosixPath
 
 from .config import Settings
 from .git import ChangeSummary
+from .models import FailureReport
 
 _ALWAYS_DENIED = (
     ".env",
@@ -66,3 +67,16 @@ def evaluate_changes(
             errors.append(f"Direct-main denied path changed: {filename}")
 
     return PolicyResult(not errors, tuple(errors))
+
+
+def evaluate_report_scope(report: FailureReport, summary: ChangeSummary) -> PolicyResult:
+    expected = f"{report.user_feedback or ''}\n{report.expected_behavior or ''}".lower()
+    requires_cross_database_orchestration = "cockpit_config" in expected and "fk_" in expected
+    has_application_change = any(filename.startswith("app/") for filename in summary.files)
+    errors = []
+    if requires_cross_database_orchestration and not has_application_change:
+        errors.append(
+            "Cross-database report changed no application code; inspect routing and implement staged orchestration"
+        )
+
+    return PolicyResult(not errors, tuple(errors))

src/lai_automation/worker.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import logging
+import re
 import signal
 import threading
 import time
@@ -8,13 +9,15 @@ from collections.abc import Iterator
 from contextlib import contextmanager
 from dataclasses import asdict
 
+import pymysql
+
 from .agent import AiderAgent
 from .config import Settings
 from .database import FailureRepository
 from .git import ChangeSummary, GitWorkspace
 from .gitea import GiteaClient
 from .models import FailureReport
-from .policy import evaluate_changes
+from .policy import evaluate_changes, evaluate_report_scope
 from .prompt import build_retry_task, build_task
 
 LOG = logging.getLogger(__name__)
@@ -77,10 +80,13 @@ class AutomationWorker:
                     ",".join(summary.files) or "none",
                 )
                 policy = evaluate_changes(summary, self.settings, workspace.path)
+                scope_policy = evaluate_report_scope(report, summary)
+                if policy.allowed and not scope_policy.allowed:
+                    policy = scope_policy
                 if not policy.allowed:
                     retryable = not summary.files or all(
                         filename.startswith("tests/") for filename in summary.files
-                    )
+                    ) or not scope_policy.allowed
                     if retryable and attempt < self.settings.max_agent_attempts:
                         LOG.warning(
                             "Policy rejected retryable changes for report %s on attempt %s: %s",
@@ -224,6 +230,7 @@ class AutomationWorker:
         attempt: int | None = None,
     ) -> list[dict]:
         results = []
+        self._prepare_test_schema()
         test_environment = {
             "AGENT_TEST_DB_ENABLED": "true",
             "APP_ENV": "testing",
@@ -279,6 +286,47 @@ class AutomationWorker:
                 raise RuntimeError(f"Test command failed: {command}\n{output}")
         return results
 
+    def _prepare_test_schema(self) -> None:
+        database = self.settings.test_customer_database
+        if not re.fullmatch(r"[A-Za-z0-9_]+", database):
+            raise ValueError("TEST_CUSTOMER_DB contains invalid characters")
+        connection = pymysql.connect(
+            host=self.settings.test_database_host,
+            port=self.settings.test_database_port,
+            user=self.settings.test_database_user,
+            password=self.settings.test_database_password,
+            database=database,
+            autocommit=True,
+            connect_timeout=10,
+        )
+        try:
+            with connection.cursor() as cursor:
+                cursor.execute(
+                    """
+                    CREATE TABLE IF NOT EXISTS FK_Nummerkreis (
+                        Kennung INT PRIMARY KEY,
+                        Bezeichnung VARCHAR(100) NOT NULL,
+                        Bezeichnung_Kurz VARCHAR(20) NOT NULL
+                    )
+                    """
+                )
+                cursor.executemany(
+                    """
+                    INSERT INTO FK_Nummerkreis (Kennung, Bezeichnung, Bezeichnung_Kurz)
+                    VALUES (%s, %s, %s)
+                    ON DUPLICATE KEY UPDATE
+                        Bezeichnung = VALUES(Bezeichnung),
+                        Bezeichnung_Kurz = VALUES(Bezeichnung_Kurz)
+                    """,
+                    (
+                        (3, "Rechnung", "RG"),
+                        (4, "Lieferschein", "LS"),
+                        (8, "Gutschrift", "GS"),
+                    ),
+                )
+        finally:
+            connection.close()
+
     @contextmanager
     def _progress(
         self,

tests/test_policy.py

@@ -3,7 +3,8 @@ from dataclasses import replace
 from test_config import settings
 
 from lai_automation.git import ChangeSummary
-from lai_automation.policy import evaluate_changes
+from lai_automation.models import FailureReport
+from lai_automation.policy import evaluate_changes, evaluate_report_scope
 
 
 def test_policy_accepts_small_application_change() -> None:
@@ -52,3 +53,33 @@ def test_policy_rejects_symlinks(tmp_path) -> None:
     (tmp_path / "link.php").symlink_to("/etc/passwd")
     result = evaluate_changes(ChangeSummary(("link.php",), 1, 0), settings(), tmp_path)
     assert not result.allowed
+
+
+def test_cross_database_report_requires_application_orchestration() -> None:
+    report = FailureReport(
+        id=56,
+        chat_message_id=1,
+        source="user_feedback",
+        category="incorrect_answer",
+        user_feedback="Use cockpit_config.time_registration",
+        feedback_type="wrong_result",
+        expected_behavior="Join cockpit_config manager_id and FK_AuftragPos",
+        prompt="Hours in April",
+        response="0 hours",
+        sql_query="SELECT 0",
+        error_message=None,
+        context={},
+    )
+
+    rejected = evaluate_report_scope(
+        report,
+        ChangeSummary(("config/erp_semantics.php", "tests/Unit/FixTest.php"), 20, 2),
+    )
+    accepted = evaluate_report_scope(
+        report,
+        ChangeSummary(("app/Http/Controllers/ChatController.php", "tests/Unit/FixTest.php"), 20, 2),
+    )
+
+    assert not rejected.allowed
+    assert "staged orchestration" in rejected.errors[0]
+    assert accepted.allowed

tests/test_worker.py

@@ -1,6 +1,6 @@
 import threading
 from pathlib import Path
-from unittest.mock import Mock, patch
+from unittest.mock import MagicMock, Mock, patch
 
 from test_config import settings
 
@@ -44,6 +44,7 @@ def test_stop_signal_interrupts_worker_wait() -> None:
 
 def test_database_credentials_are_only_passed_to_test_process() -> None:
     worker = AutomationWorker(settings())
+    worker._prepare_test_schema = Mock()
     workspace = Mock()
     workspace.run.return_value = Mock(returncode=0, stdout="ok", stderr="")
 
@@ -61,8 +62,25 @@ def test_database_credentials_are_only_passed_to_test_process() -> None:
     assert environment["COCKPIT_CONFIG_DATABASE"] == "cockpit_agent_test"
 
 
+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) -> None:
     worker = AutomationWorker(settings())
+    worker._prepare_test_schema = Mock()
     workspace = Mock()
     workspace.run.return_value = Mock(returncode=0, stdout="sensitive output", stderr="")
 
@@ -76,6 +94,7 @@ def test_validation_commands_are_logged_without_output(caplog) -> None:
 
 def test_failed_validation_preserves_start_and_end_of_long_output() -> None:
     worker = AutomationWorker(settings())
+    worker._prepare_test_schema = Mock()
     workspace = Mock()
     workspace.run.return_value = Mock(
         returncode=1,