Preserve integration concerns and diagnostics for large reviews

.github/workflows/llm-review.yml

@@ -225,11 +225,11 @@ jobs:
         run: uv run python -m lean_pool.review
 
       - name: Upload review evidence
-        if: success() && steps.ci.outputs.conclusion == 'success'
+        if: always() && steps.ci.outputs.conclusion == 'success'
         uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02
         with:
           name: review-evidence
-          path: ${{ runner.temp }}/review-evidence.json
+          path: ${{ runner.temp }}/review-evidence*
           if-no-files-found: ignore
           retention-days: 30
 

python/lean_pool/review.py

@@ -1087,6 +1087,7 @@ class ReviewResult:
     calls: int = 1
     coverage: str | None = None
     requests: tuple[ReviewResult, ...] = ()
+    source_reviews: tuple[ReviewResult, ...] = ()
 
 
 def _review_messages(
@@ -1207,7 +1208,9 @@ def _review_in_portions(
         manifest, [r.payload for r in results], shared
     )
     final = _integrate_portions(diff, evidence, obligations, budget, prepare, send)
-    return replace(final, portions=len(results), coverage=manifest)
+    return replace(
+        final, portions=len(results), coverage=manifest, source_reviews=tuple(results)
+    )
 
 
 def _integrate_portions(
@@ -1233,13 +1236,11 @@ def _integrate_portions(
             )
         final = send(messages)
         queries = final.payload.get("source_requests", [])
-        if (
-            not isinstance(queries, list)
-            or len(queries) > 20
-            or any(not isinstance(query, str) or not query.strip() for query in queries)
+        if not isinstance(queries, list) or any(
+            not isinstance(query, str) or not query.strip() for query in queries
         ):
             raise ValueError(
-                "Integration source requests must be at most 20 nonempty strings"
+                f"Integration source requests must be nonempty strings: {queries!r}"
             )
         if not queries:
             break
@@ -1250,19 +1251,38 @@ def _integrate_portions(
             )
         )
         bundle["obligations"] = obligations
-        updated = prepare(
-            json.dumps(bundle, ensure_ascii=False),
-            review_portions.integration_instructions(),
-        )
-        available = budget - _message_tokens(updated)
-        allowance = max(0, (available - 512) // len(queries))
-        bundle["source_followups"].extend(
-            review_portions.source_excerpts(diff, query, allowance) for query in queries
-        )
+        if not _fit_source_followups(diff, queries, bundle, budget, prepare):
+            break
     payload = review_portions.enforce_resolutions(final.payload, obligations)
     return replace(final, payload=payload)
 
 
+def _fit_source_followups(
+    diff: str, queries: list[str], bundle: dict, budget: int, prepare: Callable
+) -> bool:
+    """Fit the serialized excerpts and their metadata within the next call."""
+    rules = review_portions.integration_instructions()
+    available = budget - _message_tokens(
+        prepare(json.dumps(bundle, ensure_ascii=False), rules)
+    )
+    allowance = max(
+        0, int((available - 512) * CHARS_PER_TOKEN_ESTIMATE) // len(queries)
+    )
+    for _ in range(20):
+        excerpts = [
+            review_portions.source_excerpts(diff, query, allowance) for query in queries
+        ]
+        candidate = bundle | {"source_followups": bundle["source_followups"] + excerpts}
+        messages = prepare(json.dumps(candidate, ensure_ascii=False), rules)
+        if _message_tokens(messages) <= budget:
+            bundle["source_followups"] = candidate["source_followups"]
+            return True
+        if allowance == 0:
+            break
+        allowance //= 2
+    return False
+
+
 class _ReviewSession:
     """Account for every call across concurrent portions and context retries."""
 
@@ -1279,6 +1299,9 @@ def send(self, messages: list[dict[str, str]]) -> ReviewResult:
         result = _send_review(self.model, messages, self.effort)
         with self.lock:
             self.completed.append(result)
+            if destination := os.environ.get("REVIEW_EVIDENCE_PATH"):
+                with Path(destination).with_suffix(".calls.jsonl").open("a") as stream:
+                    stream.write(json.dumps(asdict(result), default=vars) + "\n")
         return result
 
     def accounted(self, result: ReviewResult) -> ReviewResult:

python/lean_pool/review_portions.py

@@ -216,10 +216,21 @@ def enforce_resolutions(payload: dict, obligations: dict[str, str]) -> dict:
     if resolved - obligations.keys():
         raise ValueError("Integration cites an unknown review obligation")
     missing = obligations.keys() - resolved
+    final_concerns = {
+        identifier: concern
+        for identifier, concern in report_obligations(
+            "integration:final", payload
+        ).items()
+        if not identifier.endswith(":verdict")
+    }
+    obligations = obligations | final_concerns
+    missing |= final_concerns.keys()
     if missing:
         payload = dict(payload)
         findings = list(payload.get("findings") or [])
         for identifier in sorted(missing):
+            if identifier.startswith("integration:final:finding:"):
+                continue  # The concrete finding already appears in this final report.
             findings.append(
                 {
                     "file": "",

python/tests/test_review_portions.py

@@ -87,13 +87,19 @@ def send(model, messages, effort):
                 "findings": [],
                 "open_questions": [],
                 "evidence_summary": "Reviewed unique_2_499 and local definitions.",
+                "source_number": int(
+                    re.search(r"source portion (\d+)/", user).group(1)
+                ),
             }
         )
 
     monkeypatch.setattr(review, "_send_review", send)
     answer = review.request_review(review.DEFAULT_MODEL, "rules", diff, "review")
     assert answer.payload["verdict"] == "pass"
     assert answer.portions > 1
+    assert [item.payload["source_number"] for item in answer.source_reviews] == list(
+        range(1, answer.portions + 1)
+    )
     assert answer.calls == answer.portions + 1 == len(calls)
     assert answer.usage.prompt_tokens == len(calls) * 10
     assert answer.usage.completion_tokens == len(calls) * 3
@@ -496,3 +502,111 @@ def send(messages):
     if not resolve:
         assert "Potentially vacuous definition" in str(answer.payload["findings"])
         assert "MissingDefinition" in str(answer.payload["findings"])
+
+
+@pytest.mark.parametrize("field", ["findings", "open_questions", "source_requests"])
+def test_final_integration_concerns_prevent_approval(field):
+    """A final pass cannot override unresolved concerns in that same report."""
+    payload = {"verdict": "pass", field: ["Unresolved semantic question"]}
+    answer = review_portions.enforce_resolutions(payload, {})
+    assert answer["verdict"] == "discuss"
+    assert "Unresolved semantic question" in str(answer["findings"])
+
+
+def test_source_excerpt_allowance_uses_characters(monkeypatch):
+    """An excerpt can use the remaining token budget at the configured ratio."""
+    calls = []
+    allowances = []
+
+    def prepare(evidence, rules):
+        return [{"role": "user", "content": evidence}]
+
+    def send(messages):
+        calls.append(messages)
+        return result(
+            {"verdict": "pass", "source_requests": ["Known"] if len(calls) == 1 else []}
+        )
+
+    def excerpt(diff, query, allowance):
+        allowances.append(allowance)
+        return {"source": "def Known := 0"}
+
+    monkeypatch.setattr(review_portions, "source_excerpts", excerpt)
+    review._integrate_portions("def Known := 0", "{}", {}, 10_000, prepare, send)
+    assert 15_000 < allowances[0] < 20_000
+
+
+def test_many_source_requests_are_not_discarded(monkeypatch):
+    """The input budget, rather than an arbitrary request count, bounds retrieval."""
+    requested = [f"Definition{index}" for index in range(25)]
+    retrieved = []
+    calls = []
+
+    def prepare(evidence, rules):
+        return [{"role": "user", "content": evidence}]
+
+    def send(messages):
+        calls.append(messages)
+        return result(
+            {"verdict": "pass", "source_requests": requested if len(calls) == 1 else []}
+        )
+
+    def excerpt(diff, query, allowance):
+        retrieved.append(query)
+        return {"query": query, "status": "No matching source"}
+
+    monkeypatch.setattr(review_portions, "source_excerpts", excerpt)
+    answer = review._integrate_portions("", "{}", {}, 10_000, prepare, send)
+    assert retrieved == requested
+    assert answer.payload["verdict"] == "discuss"
+
+
+def test_completed_calls_are_saved_before_later_failure(monkeypatch, tmp_path):
+    """Successful source work survives an integration or transport failure."""
+    destination = tmp_path / "review-evidence.json"
+    monkeypatch.setenv("REVIEW_EVIDENCE_PATH", str(destination))
+    monkeypatch.setattr(
+        review, "_send_review", lambda *args: result({"verdict": "pass"})
+    )
+    session = review._ReviewSession(review.DEFAULT_MODEL, "xhigh")
+    session.send([])
+    saved = json.loads(destination.with_suffix(".calls.jsonl").read_text())
+    assert saved["payload"]["verdict"] == "pass"
+
+
+def test_followup_budget_includes_serialized_metadata_and_escaping(monkeypatch):
+    """Escaping and many per-query envelopes cannot overflow the next call."""
+    calls = []
+
+    def prepare(evidence, rules):
+        return [{"role": "user", "content": evidence}]
+
+    def send(messages):
+        assert review._message_tokens(messages) <= 10_000
+        calls.append(messages)
+        return result(
+            {
+                "verdict": "pass",
+                "source_requests": [f"D{i}" for i in range(20)]
+                if len(calls) == 1
+                else [],
+            }
+        )
+
+    def excerpt(diff, query, allowance):
+        return {"query": query, "source": "\\" * allowance, "status": "metadata " * 50}
+
+    monkeypatch.setattr(review_portions, "source_excerpts", excerpt)
+    answer = review._integrate_portions("", "{}", {}, 10_000, prepare, send)
+    assert len(calls) == 2
+    assert answer.payload["verdict"] == "discuss"
+
+
+def test_final_findings_are_reported_once():
+    """A retained finding blocks approval without a duplicate synthetic entry."""
+    finding = {"comment": "Unresolved semantic concern"}
+    answer = review_portions.enforce_resolutions(
+        {"verdict": "pass", "findings": [finding]}, {}
+    )
+    assert answer["verdict"] == "discuss"
+    assert answer["findings"] == [finding]