merge(search-with-proof): served-root control-signal feedback to the agent [skip ci]

Non-blocking advisory hint when the control shows the app didn't serve ${served_root},
so the agent self-corrects its served-directory wiring. Host-side; verdicts unchanged.

Co-Authored-By: Claude Opus 4.8 <[email protected]>

benchmarks/search/pilot_run.py

@@ -328,7 +328,13 @@ class Pilot:
                 if result.outcome != "demonstrated":
                     self.failure("verify", "no_consistent_differential" if result.outcome == "inconclusive" else result.outcome)
                 end = result.outcome
-                return dict(status="ok", outcome=result.outcome, evidence_refs=["verify:owned"], cost_usd=0)
+                return dict(
+                    status="ok",
+                    outcome=result.outcome,
+                    evidence_refs=["verify:owned"],
+                    cost_usd=0,
+                    control_observed=result.sides[0].control_observed if result.sides else None,
+                )
             if task.step == "explore":
                 if self.args.dry_run:
                     self.worker.executor = InProcessExecutor(self.sides[0 if self.args.side == "vulnerable" else 1].checkout)

src/openultrasast/search/coordinator.py

@@ -274,6 +274,17 @@ class Coordinator:
                         )
                         return self._end("goal_met")
                     self.progress["failures"].append({"task_id": result["task_id"], "reason": "not_demonstrated"})
+                    if result.get("control_observed") is False and not any(h["id"] == "served-root" for h in self.state["hints"]):
+                        self.state["hints"].append(
+                            dict(
+                                id="served-root",
+                                step="verify",
+                                task_id=result["task_id"],
+                                text="The application did not serve the oracle root: its served-directory configuration "
+                                "(argument, environment variable, or config file) must be set to the ${served_root} "
+                                "placeholder itself, not a real path or an explore-time test directory.",
+                            )
+                        )
                 elif intent is not None:
                     result = self._call("explore", {"intent_id": intent["id"], "description": intent["description"]})
                     intent["status"] = "concluded"

tests/test_search_coordinator.py

@@ -32,6 +32,48 @@ def test_reason_explore_verify_and_resume(tmp_path):
     assert len(calls) == 3
 
 
+@pytest.mark.parametrize("control_observed", [False, True, None])
+@pytest.mark.parametrize("outcome", ["inconclusive", "demonstrated"])
+def test_served_root_hint_is_advisory_and_deduplicated(tmp_path, control_observed, outcome):
+    import yaml
+
+    verifications = []
+    reason_hints = []
+
+    def execute(task):
+        if task.step == "reason":
+            reason_hints.append(yaml.safe_load(task.snapshot)["hints"])
+            descriptions = ["inspect startup configuration", "construct corrected demo"]
+            intents = [{"description": descriptions[len(verifications)]}] if len(verifications) < 2 else []
+            return {"status": "ok", "cost_usd": 0, "intents": intents, "complete": not intents}
+        if task.step == "explore":
+            return {"status": "ok", "cost_usd": 0, "fact": {"text": "candidate", "evidence_refs": ["capture:1"], "demo": "demo:1"}}
+        verifications.append(task.id)
+        return {"status": "ok", "cost_usd": 0, "outcome": outcome, "evidence_refs": ["verify:1"], "control_observed": control_observed}
+
+    b = board(tmp_path)
+    state = Coordinator(b, execute).run()
+    hints = [hint for hint in state["hints"] if hint["id"] == "served-root"]
+    if outcome == "inconclusive":
+        assert len(verifications) == 2
+        assert state["end_reason"] == "exhausted"
+        assert state["checkpoint"]["failures"] == [{"task_id": task_id, "reason": "not_demonstrated"} for task_id in verifications]
+    else:
+        assert len(verifications) == 1
+        assert state["end_reason"] == "goal_met"
+        assert not state["checkpoint"]["failures"]
+    if control_observed is False and outcome == "inconclusive":
+        assert len(hints) == 1
+        assert hints[0]["step"] == "verify" and hints[0]["task_id"] == verifications[0]
+        assert "${served_root}" in hints[0]["text"]
+        assert "argument, environment variable, or config file" in hints[0]["text"]
+        assert reason_hints == [[], hints, hints]
+    else:
+        assert not hints
+        assert all(not snapshot_hints for snapshot_hints in reason_hints)
+    assert Board.resume(b.store, "search-1", b.head, coordinator="host").state == state
+
+
 def test_reason_claim_cannot_prove_goal(tmp_path):
     result = Coordinator(board(tmp_path), lambda task: {"status": "ok", "cost_usd": 0, "complete": True, "facts": []}).run()
     assert result["end_reason"] == "exhausted"

tests/test_search_pilot_run.py

@@ -350,6 +350,30 @@ def test_fixed_effect_reporting_is_independent_of_differential(tmp_path, monkeyp
     assert pilot.record["fixed_effect_observed"] is expected
 
 
+@pytest.mark.parametrize("control_observed", [False, True, None])
+@pytest.mark.parametrize("with_sides", [False, True])
+def test_verify_dispatch_returns_affected_control_signal(tmp_path, monkeypatch, control_observed, with_sides):
+    from dataclasses import asdict
+
+    from openultrasast.search.budget import SearchBudget
+    from openultrasast.search.coordinator import SearchTask
+    from openultrasast.search.verify import SideRecord, VerificationRecord
+
+    pilot = make_pilot(tmp_path)
+    affected = SideRecord("observed", (), 0, "completed", control_observed=control_observed)
+    comparator = SideRecord("observed", (), 0, "completed", control_observed=control_observed is not True)
+    result = VerificationRecord("inconclusive", (affected, comparator) if with_sides else (), 0, "no differential")
+    monkeypatch.setattr(pilot, "verification", lambda limits: result)
+    task = SearchTask("task-1", "search-1", "verify", "", {}, asdict(SearchBudget()))
+    assert pilot.dispatch(task) == {
+        "status": "ok",
+        "outcome": "inconclusive",
+        "evidence_refs": ["verify:owned"],
+        "cost_usd": 0,
+        "control_observed": control_observed if with_sides else None,
+    }
+
+
 def test_vm_preflight_refuses_before_client_or_store(tmp_path, monkeypatch):
     from benchmarks.search import pilot_run