Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
78 changes: 59 additions & 19 deletions agent-svc/agent/experimental/publication_gate.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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}
Expand All @@ -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()
Expand All @@ -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)
Expand Down
36 changes: 30 additions & 6 deletions agent-svc/agent/experimental/real_journey.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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,
Expand Down
8 changes: 8 additions & 0 deletions agent-svc/agent/routes/experimental_research.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down
19 changes: 19 additions & 0 deletions docs/guides/durable-research.md
Original file line number Diff line number Diff line change
Expand Up @@ -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.
57 changes: 55 additions & 2 deletions tests/integration/test_live_research_http_adapters.py
Original file line number Diff line number Diff line change
Expand Up @@ -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] = []
Expand Down Expand Up @@ -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": [],
}
Expand All @@ -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.",
Expand Down Expand Up @@ -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),
)
28 changes: 28 additions & 0 deletions tests/service/test_live_research_recovery.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Loading
Loading