Fix durable research failing before it produces a report

agent-svc/agent/experimental/durable_plan.py

@@ -26,6 +26,7 @@ async def plan_queries(
     model: str,
     selected_sources: tuple[CapturedSource, ...] = (),
     as_of: str | None = None,
+    max_output_tokens: int = 2048,
 ) -> tuple[str, ...]:
     reply = await complete(
         ReviewRequest(
@@ -44,7 +45,7 @@ async def plan_queries(
                 }
             ).encode(),
             model,
-            2048,
+            max_output_tokens,
             QueryPlan.model_json_schema(),
         )
     )

agent-svc/agent/experimental/live_runner.py

@@ -33,7 +33,11 @@ async def run_live(record: Any, app: Any):
         raise ValueError("admission unavailable")
     evaluated = datetime.fromisoformat(admitted.payload["evaluation_time"])
     model = admitted.payload["model"]
-    planned_policy = admitted.payload.get("policy_version") == "durable-live/2"
+    bounded_evidence_policy = admitted.payload.get("policy_version") == "durable-live/3"
+    planned_policy = admitted.payload.get("policy_version") in {
+        "durable-live/2",
+        "durable-live/3",
+    }
     selected_sources = tuple(
         CapturedSource(**source)
         for source in admitted.payload.get("selected_sources", [])
@@ -198,6 +202,7 @@ async def fetch():
         token_parameter, token_budget = next(
             iter(admitted.payload["token_params"].items())
         )
+        completion_budget = min(token_budget, 8192) if bounded_evidence_policy else 2048
         transport = ReviewTransport(
             client,
             base_url=app.state.llm_base_url,
@@ -243,6 +248,7 @@ async def fetch():
                     model=model,
                     selected_sources=selected_sources,
                     as_of=admitted.payload.get("as_of"),
+                    max_output_tokens=completion_budget,
                 )
                 if planned_policy and len(selected_sources) < 8
                 else None
@@ -265,4 +271,6 @@ async def fetch():
                 queries=queries,
                 source_limit=8 if queries is not None else 3,
                 allow_partial=planned_policy,
+                max_output_tokens=completion_budget,
+                evidence_budget_chars=32000 if bounded_evidence_policy else None,
             )

agent-svc/agent/experimental/query_construction.py

@@ -6,7 +6,7 @@
 from collections.abc import Callable
 from dataclasses import dataclass
 from datetime import UTC, datetime
-from typing import Annotated, Literal
+from typing import Annotated, Literal, NotRequired, TypedDict
 from uuid import NAMESPACE_URL, uuid4, uuid5
 
 from pydantic import Field
@@ -159,6 +159,57 @@ def _edges(content: ConstructedContent, revision_id: str) -> list[dict[str, obje
     return edges
 
 
+class ConstructionLine(TypedDict):
+    line: int
+    text: str
+
+
+class ConstructionSource(TypedDict):
+    source_index: int
+    url: str | None
+    lines: list[ConstructionLine]
+    omitted_lines: NotRequired[int]
+
+
+def construction_sources(
+    sources: tuple[CapturedSource, ...], objective: str, budget: int | None = None
+) -> list[ConstructionSource]:
+    """Select whole original lines; retain full snapshots separately for citation reads."""
+    from ..research.evidence import select_passages, validate_evidence_budget
+
+    if budget is not None:
+        validate_evidence_budget(budget)
+    remaining = budget
+    result: list[ConstructionSource] = []
+    for index, source in enumerate(sources, 1):
+        lines = source.text.splitlines(keepends=True)
+        rows: list[ConstructionLine]
+        if remaining is None:
+            rows = [{"line": n, "text": line} for n, line in enumerate(lines, 1)]
+        else:
+            allowance = remaining // (len(sources) - index + 1)
+            spans = select_passages(source.text, objective, allowance)["spans"]
+            rows = []
+            offset = 0
+            for n, line in enumerate(lines, 1):
+                end = offset + len(line)
+                if any(
+                    offset >= span["start"] and end <= span["end"] for span in spans
+                ):
+                    rows.append({"line": n, "text": line})
+                offset = end
+            remaining -= sum(len(row["text"]) for row in rows)
+        entry: ConstructionSource = {
+            "source_index": index,
+            "url": source.url,
+            "lines": rows,
+        }
+        if budget is not None:
+            entry["omitted_lines"] = len(lines) - len(rows)
+        result.append(entry)
+    return result
+
+
 async def construct_research(
     objective: str,
     sources: tuple[CapturedSource, ...],
@@ -171,6 +222,8 @@ async def construct_research(
     revision_id: str | None = None,
     as_of: str | None = None,
     acquisition_failures: tuple[str, ...] = (),
+    max_output_tokens: int = 2048,
+    evidence_budget_chars: int | None = None,
 ) -> ConstructedResearch:
     """One model call, no retries; returned knowledge is NOT verified or publishable."""
     if (
@@ -214,23 +267,16 @@ async def construct_research(
                 reference, source.text.encode(), "utf8-exact/1", "text/plain"
             )
         )
+    if type(max_output_tokens) is not int or not 1 <= max_output_tokens <= 8192:
+        raise ValueError("invalid construction completion budget")
+    presented_sources = construction_sources(sources, objective, evidence_budget_chars)
     payload = json.dumps(
         {
             "objective": objective,
             "as_of": as_of,
             "acquisition_failures": acquisition_failures,
             "as_of_semantics": "Evidence evaluation target only; newly captured text is not an archived capture. Do not invent dates or historical validity.",
-            "sources": [
-                {
-                    "source_index": index,
-                    "url": s.url,
-                    "lines": [
-                        {"line": n, "text": line}
-                        for n, line in enumerate(s.text.splitlines(keepends=True), 1)
-                    ],
-                }
-                for index, s in enumerate(sources, 1)
-            ],
+            "sources": presented_sources,
             "schema": ConstructedContent.model_json_schema(),
         }
     ).encode()
@@ -240,10 +286,15 @@ async def construct_research(
         reply = await asyncio.ensure_future(
             complete(
                 ReviewRequest(
-                    CONSTRUCTION_PROMPT,
+                    CONSTRUCTION_PROMPT
+                    + (
+                        "\nSources are selected excerpts with original line numbers and possible gaps. Select only contiguous lines actually supplied. Never infer absence or exhaustive coverage from omitted lines."
+                        if evidence_budget_chars is not None
+                        else ""
+                    ),
                     payload,
                     model,
-                    2048,
+                    max_output_tokens,
                     ConstructedContent.model_json_schema(),
                 )
             )
@@ -255,6 +306,15 @@ async def construct_research(
         reply.content, schema_version="research-construction/4"
     )
     content = ConstructedContent.model_validate_json(document.data)
+    if evidence_budget_chars is not None:
+        for item in content.evidence:
+            if not 1 <= item.source_index <= len(presented_sources):
+                raise ValueError("model evidence was not supplied")
+            shown = {
+                row["line"] for row in presented_sources[item.source_index - 1]["lines"]
+            }
+            if not all(n in shown for n in range(item.start_line, item.end_line + 1)):
+                raise ValueError("model evidence was not supplied")
     answer = _selected((content.answer_claim_index,), len(content.claims), "claim")[0]
     now = clock()
     offset = now.utcoffset()

agent-svc/agent/experimental/query_runner.py

@@ -31,6 +31,8 @@ async def run_query(
     as_of: str | None = None,
     queries: tuple[str, ...] | None = None,
     allow_partial: bool = False,
+    max_output_tokens: int = 2048,
+    evidence_budget_chars: int | None = None,
 ) -> JourneyResult:
     """Run a bounded query plan, retained acquisitions, construction and checks.
 
@@ -150,4 +152,6 @@ async def run_query(
             artifact_set_id=artifact_set_id,
             as_of=as_of,
             acquisition_failures=tuple(failures),
+            max_output_tokens=max_output_tokens,
+            evidence_budget_chars=evidence_budget_chars,
         )

agent-svc/agent/experimental/real_journey.py

@@ -190,6 +190,8 @@ async def research_from_sources(
     artifact_set_id: str | None = None,
     as_of: str | None = None,
     acquisition_failures: tuple[str, ...] = (),
+    max_output_tokens: int = 2048,
+    evidence_budget_chars: int | None = None,
 ) -> JourneyResult:
     """Construct, check and audit a bounded root; no retained publication yet."""
     execution_clock = clock or (lambda: datetime.now(UTC))
@@ -204,12 +206,15 @@ async def research_from_sources(
         revision_id=revision_id,
         as_of=as_of,
         acquisition_failures=acquisition_failures,
+        max_output_tokens=max_output_tokens,
+        evidence_budget_chars=evidence_budget_chars,
     )
     adapter = ModelReviewAdapter(
         provider="configured-litellm",
         model=model,
         complete=complete,
         timeout_seconds=90,
+        max_output_tokens=max_output_tokens,
     )
     mechanical_reviewer = ToolReviewer(
         kind="tool",

agent-svc/agent/routes/experimental_research.py

@@ -1127,7 +1127,7 @@ async def admit_research_run(
                     "objective": payload.objective,
                     "research_id": str(uuid4()),
                     "execution_mode": "fixture" if fixture else "live",
-                    "policy_version": "durable-live/2",
+                    "policy_version": "durable-live/3",
                     "model": model,
                     "token_params": _token_budget_params(),
                     "selected_sources": [asdict(source) for source in selected_sources],

docs/adr/0070-execute-live-durable-research.md

@@ -23,7 +23,11 @@ Startup scans reclaimable admissions and tracks worker shutdown. Terminal failur
 
 ## Bounded discovery, time and callbacks
 
-New admissions pin policy `durable-live/2`: one recorded model planning call produces at most three distinct queries, followed by round-robin source selection across their results. At most eight sources, 256000 source bytes, 64 model calls and an absolute 180-second execution deadline are permitted. Selected retained sources consume the same budgets; eight retained sources need no new search. A known failed capture can leave a checked report with insufficient coverage, never an unqualified complete result. Ambiguous dispatch still stops immediately. Older admissions keep their one-search/three-fresh-source policy so restart does not alter retained operation identities. These bounds are not equivalent to the existing iterative deep-research loop.
+New admissions pin policy `durable-live/3`: one recorded model planning call produces at most three distinct queries, followed by round-robin source selection across their results. At most eight sources, 256000 source bytes, 64 model calls and an absolute 180-second execution deadline are permitted. Selected retained sources consume the same budgets; eight retained sources need no new search. A known failed capture can leave a checked report with insufficient coverage, never an unqualified complete result. Ambiguous dispatch still stops immediately. Older admissions keep their one-search/three-fresh-source policy so restart does not alter retained operation identities. These bounds are not equivalent to the existing iterative deep-research loop.
+
+Policy version 3 also uses the admitted completion allowance, capped at 8192 tokens, for planning, construction and review. A production smoke test showed that the former 2048-token construction ceiling could be consumed entirely by model reasoning, leaving no answer. Raising that ceiling alone did not resolve the observed latency. Construction therefore selects relevant whole lines within a shared 32000-character allowance, divided across sources. Full captured source text remains retained. Original line numbers are preserved, and a model citation that crosses omitted lines is rejected. The prompt explicitly warns that excerpts cannot establish exhaustive coverage. These are request limits, not performance targets; measured completion times remain observations.
+
+Admissions under `durable-live/2` keep their original full-source construction and 2048-token request ceiling. Their recorded request identities and receipts must remain replayable. Version 3 changes only newly admitted work and does not authorize retries of ambiguous external calls.
 
 `as_of` means a timezone-aware, nonfuture evidence evaluation target. It does not request historical search or fabricate archived captures. Original URLs and retrieval dates remain null when unknown. Freshness checks compare actual retained evidence times with that target; unsupported past validity fails as `historical_validity_unestablished`, even if a model returns a passing opinion. Evidence captured before the target remains eligible for the other checks.
 

tests/service/test_live_research_recovery.py

@@ -36,7 +36,7 @@ def ledger():
         value.redis.close()
 
 
-def admitted(ledger, policy=None):
+def admitted(ledger, policy=None, max_tokens=2048):
     value = ledger.admit(
         "test-owner",
         str(uuid4()),
@@ -46,7 +46,7 @@ def admitted(ledger, policy=None):
             "research_id": str(uuid4()),
             "execution_mode": "live",
             "model": "configured-test-model",
-            "token_params": {"max_tokens": 2048},
+            "token_params": {"max_tokens": max_tokens},
             "evaluation_time": datetime.now(UTC).isoformat(),
             "policy_version": policy,
         },
@@ -249,15 +249,15 @@ async def blocked(*_args):
 
 
 @pytest.mark.asyncio
-@pytest.mark.parametrize("policy", [None, "durable-live/2"])
+@pytest.mark.parametrize("policy", [None, "durable-live/2", "durable-live/3"])
 async def test_real_construction_and_review_replay_without_new_model_calls(
     ledger, monkeypatch, policy
 ):
     from agent.experimental import real_journey
 
     from tests.unit.test_query_construction import TEXT, batch_decisions, proposal
 
-    record = admitted(ledger, policy)
+    record = admitted(ledger, policy, max_tokens=8192)
     model_calls = []
     search_calls = []
 
@@ -283,7 +283,15 @@ def __init__(self, *_args, **_kwargs):
             pass
 
         async def __call__(self, request):
+            assert request.max_output_tokens == (
+                8192 if policy == "durable-live/3" else 2048
+            )
             value = json.loads(request.payload)
+            if "schema" in value:
+                assert all(
+                    ("omitted_lines" in source) == (policy == "durable-live/3")
+                    for source in value["sources"]
+                )
             model_calls.append(value)
             reply = (
                 {