diff --git a/agent-svc/agent/experimental/publication_gate.py b/agent-svc/agent/experimental/publication_gate.py index 5597bd33..f0504d92 100644 --- a/agent-svc/agent/experimental/publication_gate.py +++ b/agent-svc/agent/experimental/publication_gate.py @@ -3,7 +3,7 @@ import asyncio from dataclasses import dataclass -from .checked_knowledge import admit_checked_history +from .checked_knowledge import CheckedKnowledge, admit_checked_history from .context_sources import ContextSourceResolver from .knowledge_checks import HistoricalValidityError, Reviewer from .knowledge_context import moment @@ -18,6 +18,62 @@ class PublicationCandidate: fixture_only: bool +class UnverifiedResearchError(ValueError): + """Safe explanation when a required answer cannot pass existing checks.""" + + def __init__(self, *, freshness: bool) -> None: + self.reason = ( + "source_freshness_unverified" if freshness else "answer_support_unverified" + ) + self.message = ( + "Source freshness could not be verified, so the answer was not published. " + "No searches or model calls were repeated." + if freshness + else "The answer did not pass evidence verification, so it was not published. " + "No searches or model calls were repeated." + ) + super().__init__(self.message) + + +def claim_publication_failures(knowledge: CheckedKnowledge) -> dict[str, str]: + """Use the same eligibility decision for rendering and final publication.""" + inputs = {i.input_id: i for i in knowledge.verification_inputs} + states = {a.claim_id: a.state for a in knowledge.assessment_links} + contested = { + c for conflict in knowledge.context.conflicts for c in conflict.claim_ids + } + failures = {} + for claim in knowledge.context.claims: + checks = [ + r + for r in knowledge.verifications + if inputs[r.input_id].subject_id == claim.claim_id + ] + assessments = [ + a + for a in knowledge.assessments + if inputs[a.input_id].subject_id == claim.claim_id + ] + if ( + states.get(claim.claim_id) != "supported" + or claim.claim_id in contested + or not assessments + or any(a.outcome != "supported" for a in assessments) + or {inputs[r.input_id].check_type for r in checks} + != {"semantic_support", "freshness"} + or any(r.verdict != "pass" for r in checks) + ): + failures[claim.claim_id] = ( + "source_freshness_unverified" + if any( + inputs[r.input_id].check_type == "freshness" and r.verdict != "pass" + for r in checks + ) + else "answer_support_unverified" + ) + return failures + + def _check_eligibility(admitted: AdmittedManifest) -> None: knowledge = admitted.knowledge inputs = {i.input_id: i for i in knowledge.verification_inputs} @@ -33,10 +89,7 @@ def _check_eligibility(admitted: AdmittedManifest) -> None: raise ValueError( "publication requires passing structural and conflict/coverage checks" ) - states = {a.claim_id: a.state for a in knowledge.assessment_links} - contested = { - c for conflict in knowledge.context.conflicts for c in conflict.claim_ids - } + failures = claim_publication_failures(knowledge) for artifact in admitted.manifest.artifacts: for statement in artifact.statements: expected_evidence: set[str] = set() @@ -46,20 +99,7 @@ def _check_eligibility(admitted: AdmittedManifest) -> None: for r in knowledge.verifications if inputs[r.input_id].subject_id == claim_id ] - assessments = [ - a - for a in knowledge.assessments - if inputs[a.input_id].subject_id == claim_id - ] - if ( - states.get(claim_id) != "supported" - or claim_id in contested - or not assessments - or any(a.outcome != "supported" for a in assessments) - or {inputs[r.input_id].check_type for r in checks} - != {"semantic_support", "freshness"} - or any(r.verdict != "pass" for r in checks) - ): + if claim_id in failures: if ( knowledge.context.as_of is not None and moment(knowledge.context.as_of) diff --git a/agent-svc/agent/experimental/real_journey.py b/agent-svc/agent/experimental/real_journey.py index dcd829fe..1ae4132e 100644 --- a/agent-svc/agent/experimental/real_journey.py +++ b/agent-svc/agent/experimental/real_journey.py @@ -8,10 +8,16 @@ from .checked_knowledge import CheckedKnowledge from .consolidated_journey import ConsolidatedJourney, JourneyResult, RenderedReport from .knowledge import text_digest -from .knowledge_checks import KnowledgeCheckInput, ModelReviewer, ToolReviewer -from .knowledge_context import KnowledgeContext +from .knowledge_checks import ( + HistoricalValidityError, + KnowledgeCheckInput, + ModelReviewer, + ToolReviewer, +) +from .knowledge_context import KnowledgeContext, moment from .knowledge_execution import ExecutionDecision from .model_review import Complete, ModelReviewAdapter +from .publication_gate import UnverifiedResearchError, claim_publication_failures from .query_construction import CapturedSource, construct_research from .render_execution import RenderInspection from .render_manifest import ManifestArtifact, Renderer @@ -110,11 +116,29 @@ async def render_research( ) -> tuple[RenderedReport, ...]: """Deterministic presentation of assessed claims; full output still needs audit.""" context = knowledge.context + failures = claim_publication_failures(knowledge) + required = {q.report_claim_id for q in context.questions} + blocked = required & failures.keys() + if blocked: + freshness = any(failures[c] == "source_freshness_unverified" for c in blocked) + if ( + freshness + and context.as_of is not None + and moment(context.as_of) < moment(context.created_at) + ): + raise HistoricalValidityError( + "historical validity is not established by retained evidence" + ) + raise UnverifiedResearchError(freshness=freshness) reports = [] for layer in ("summary", "analysis", "dossier"): text = f"# {layer.title()}\n\nModel-reviewed experimental research.\n\n{context.objective}\n\nCoverage: {knowledge.coverage}.\n\n" statements = [] + if failures: + text += f"{len(failures)} additional conclusions could not be verified and are omitted.\n\n" for claim in context.claims: + if claim.claim_id in failures: + continue start = len(text) text += claim.text evidence = _evidence(context, claim.claim_id) @@ -256,8 +280,8 @@ async def render(knowledge): auditor = ToolReviewer( kind="tool", identity="deterministic-research-render-audit", - version="1", - configuration_digest=text_digest("assessed-claim-pyramid/1"), + version="2", + configuration_digest=text_digest("assessed-claim-pyramid/2"), ) async def audit(inspection: RenderInspection) -> ExecutionDecision: @@ -289,8 +313,8 @@ async def audit(inspection: RenderInspection) -> ExecutionDecision: verify=verify, renderer=Renderer( identity="assessed-claim-pyramid", - version="1", - configuration_digest=text_digest("assessed-claim-pyramid/1"), + version="2", + configuration_digest=text_digest("assessed-claim-pyramid/2"), ), render=render, auditor=auditor, diff --git a/agent-svc/agent/routes/experimental_research.py b/agent-svc/agent/routes/experimental_research.py index 4421e72a..9941abe2 100644 --- a/agent-svc/agent/routes/experimental_research.py +++ b/agent-svc/agent/routes/experimental_research.py @@ -39,6 +39,7 @@ ) from ..experimental.knowledge_checks import HistoricalValidityError from ..experimental.live_runner import run_live +from ..experimental.publication_gate import UnverifiedResearchError from ..experimental.query_construction import CapturedSource from ..experimental.source_store import StorageConflictError from ..llm import _token_budget_params @@ -807,6 +808,13 @@ async def _execute_run(record: _RunRecord) -> None: else "fixture_run_failed", "retryable": False, } + if isinstance(exc, UnverifiedResearchError) and not pending_dispatch: + record.error = { + "code": "evidence_not_verified", + "reason": exc.reason, + "message": exc.message, + "retryable": False, + } failure = _build_event(record, "error", "failed", error=record.error) if ( record.durable_ledger is not None diff --git a/docs/guides/durable-research.md b/docs/guides/durable-research.md index bc5c4aaf..762441ff 100644 --- a/docs/guides/durable-research.md +++ b/docs/guides/durable-research.md @@ -24,3 +24,22 @@ External call intents and exact completed responses are retained before dependen The optional webhook receives `research.completed`, `research.failed` or `research.cancelled` with `id` and `webhookId` equal to the run UUID, plus `data.run_id`, `data.research_id`, `data.state` and the result receipt or error. The outbox revalidates destination safety for every attempt, signs the exact body using the configured webhook secret, and never follows redirects. Deduplicate by `webhookId`: an acknowledgement lost after your receiver accepted a request can cause a repeat. Three bounded attempts lead to `delivered`, `rejected` or `unconfirmed`. The status response exposes only notification state and attempt count, not the destination or payload. Outbox recovery lasts within the execution ledger retention advertised by capabilities; it does not promise delivery after that retention expires. The verification model evaluates support; passing checks are not proof of factual truth or a human review. Read the sources and coverage qualification before relying on a report. API and integration tests use controlled local HTTP twins with real isolated retained stores and do not contact public search or paid providers. + + +## When evidence cannot be verified + +A model can find supporting text without establishing that it is current. A recent +capture alone does not establish when a page was published or took effect. The +publication checks keep that distinction. + +Reports omit additional conclusions that do not pass verification and say how +many were omitted. If a conclusion needed to answer the question cannot pass, +the run ends as `failed` with code `evidence_not_verified`, a readable `message`, +and reason `source_freshness_unverified` or `answer_support_unverified`. It does +not publish an unverified answer. A request for a past date still uses +`historical_validity_unestablished` when the available evidence cannot establish +that historical answer. + +These are final outcomes for that admission. Reading status or recovering the +run preserves its completed search and model responses; it does not repeat those +calls. A clearer error is not evidence that the original question was answered. diff --git a/tests/integration/test_live_research_http_adapters.py b/tests/integration/test_live_research_http_adapters.py index 3ec60113..41dff489 100644 --- a/tests/integration/test_live_research_http_adapters.py +++ b/tests/integration/test_live_research_http_adapters.py @@ -48,6 +48,7 @@ class LocalProviderTwin: """Three local HTTP endpoints with request logs and deterministic responses.""" def __init__(self) -> None: + self.current_claim = False self.search_requests: list[dict] = [] self.capture_requests: list[dict] = [] self.model_requests: list[dict] = [] @@ -202,7 +203,9 @@ def do_POST(self) -> None: "qualifiers": [ "One source captured by the local acceptance twin" ], - "temporal_scope": "historical", + "temporal_scope": "current" + if owner.current_claim + else "historical", "supported_by": [1], "contradicted_by": [], } @@ -217,7 +220,10 @@ def do_POST(self) -> None: "decisions": [ { "check_index": item["check_index"], - "outcome": "supported" + "outcome": "indeterminate" + if owner.current_claim + and item["check_type"] == "freshness" + else "supported" if item["check_type"] == "assessment" else "pass", "reason": "Local deterministic acceptance response; not an independent quality judgment.", @@ -1004,3 +1010,50 @@ async def test_live_retained_source_drives_independent_followup_and_owned_sessio if session_id is not None: await store.adelete(session_id) store.redis.close() + + +@pytest.mark.asyncio +async def test_current_undated_sources_explain_failure_and_survive_recovery( + durable_apps, provider_twin +): + apps, _, _, _, make_app, cleanup_ids, _ = durable_apps + provider_twin.current_claim = True + async with httpx.AsyncClient( + transport=httpx.ASGITransport(app=apps[0]), base_url="http://acceptance" + ) as client: + response = await client.post( + "/experimental/research/v1/runs", + headers={"Idempotency-Key": "unverified-source-freshness"}, + json={"objective": next(iter(OBJECTIVES))}, + ) + assert response.status_code == 202 + admitted = response.json() + cleanup_ids.append(admitted["research_id"]) + status = await _terminal(client, admitted["status_url"]) + assert status["state"] == "failed" + assert status["result"] is None + assert status["error"]["code"] == "evidence_not_verified" + assert status["error"]["reason"] == "source_freshness_unverified" + assert "Source freshness could not be verified" in status["error"]["message"] + assert status["error"]["retryable"] is False + counts = ( + len(provider_twin.model_requests), + len(provider_twin.search_requests), + len(provider_twin.capture_requests), + ) + assert counts[0] == 3 # planning, construction, review + experimental._RUNS.clear() + recovered = make_app() + apps.append(recovered) + async with httpx.AsyncClient( + transport=httpx.ASGITransport(app=recovered), base_url="http://acceptance" + ) as fresh: + reread = await fresh.get(admitted["status_url"]) + assert reread.json()["error"] == status["error"] + events = await fresh.get(admitted["events_url"]) + assert "evidence_not_verified" in events.text + assert counts == ( + len(provider_twin.model_requests), + len(provider_twin.search_requests), + len(provider_twin.capture_requests), + ) diff --git a/tests/service/test_live_research_recovery.py b/tests/service/test_live_research_recovery.py index 9059d815..782a0ba5 100644 --- a/tests/service/test_live_research_recovery.py +++ b/tests/service/test_live_research_recovery.py @@ -384,3 +384,31 @@ def unexpected(*_args, **_kwargs): claim(record) with pytest.raises((ValueError, StorageConflictError)): await live_runner.run_live(record, record.app) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("freshness", [True, False]) +async def test_evidence_failure_keeps_safe_terminal_details( + ledger, monkeypatch, freshness +): + from agent.experimental.publication_gate import UnverifiedResearchError + + record = admitted(ledger) + record.artifact_authority = EmptyAuthority() + failure = UnverifiedResearchError(freshness=freshness) + + async def fail(*_args): + raise failure + + monkeypatch.setattr(routes, "run_live", fail) + await routes._execute_run(record) + restored = routes._restore_durable_record(ledger.get(record.run_id)) + assert restored.state == "failed" + assert restored.error == { + "code": "evidence_not_verified", + "reason": failure.reason, + "message": failure.message, + "retryable": False, + } + assert restored.events[-1].error == restored.error + assert record.artifact_authority.commits == 0 diff --git a/tests/unit/test_research_publication_eligibility.py b/tests/unit/test_research_publication_eligibility.py new file mode 100644 index 00000000..41ca5159 --- /dev/null +++ b/tests/unit/test_research_publication_eligibility.py @@ -0,0 +1,103 @@ +"""Live-shaped review outcomes must reach an honest publication decision.""" + +import json + +import pytest +from agent.experimental.model_review import ModelReply +from agent.experimental.publication_gate import UnverifiedResearchError +from agent.experimental.query_construction import CapturedSource +from agent.experimental.real_journey import research_from_sources + +from tests.unit.test_query_construction import ( + QUESTION, + TEXT, + batch_decisions, + journey_clock, + proposal, +) + + +async def reviewed_run( + *, required_current, freshness="indeterminate", reject_support=False +): + calls = [] + + async def complete(request): + payload = json.loads(request.payload) + calls.append(payload) + if "schema" in payload: + content = proposal() + if required_current: + content["claims"][0]["temporal_scope"] = "current" + else: + content["claims"].append( + { + **content["claims"][0], + "text": "Unverified current conclusion.", + "temporal_scope": "current", + } + ) + else: + content = batch_decisions(payload) + for check, decision in zip( + payload["checks"], content["decisions"], strict=True + ): + if ( + check["check_type"] == "freshness" + and check["freshness"]["sources"][0]["basis"] == "unknown" + ): + decision["outcome"] = freshness + if reject_support and check["check_type"] == "semantic_support": + decision["outcome"] = "fail" + return ModelReply(json.dumps(content).encode(), "test-model", 1, 1) + + try: + return await research_from_sources( + QUESTION, + ( + CapturedSource( + "https://example.test/pilot", TEXT, "2026-09-06T00:00:00Z" + ), + ), + complete=complete, + scope_id="test-scope", + clock=journey_clock, + ) + finally: + assert len(calls) == 2, "Publication must not repeat model operations" + + +@pytest.mark.asyncio +async def test_current_reporting_claim_explains_unverified_freshness(): + with pytest.raises( + UnverifiedResearchError, match="Source freshness could not be verified" + ) as error: + await reviewed_run(required_current=True) + assert error.value.reason == "source_freshness_unverified" + + +@pytest.mark.asyncio +async def test_optional_unverified_claim_is_omitted_without_losing_verified_answer(): + result = await reviewed_run(required_current=False) + assert len(result.reports) == 3 + assert len(result.candidate.admitted.knowledge.context.claims) == 2 + for report in result.reports: + assert "Unverified current conclusion." not in report.body.decode() + assert "could not be verified" in report.body.decode() + assert {c for s in report.artifact.statements for c in s.claim_ids} == { + "claim-1" + } + + +@pytest.mark.asyncio +async def test_model_cannot_pass_freshness_without_an_allowed_basis(): + with pytest.raises(ValueError): + await reviewed_run(required_current=True, freshness="pass") + + +@pytest.mark.asyncio +async def test_required_answer_support_failure_has_a_safe_explanation(): + with pytest.raises(UnverifiedResearchError) as error: + await reviewed_run(required_current=False, reject_support=True) + assert error.value.reason == "answer_support_unverified" + assert "did not pass evidence verification" in error.value.message