Compare commits

...
6 changed files with 542 additions and 0 deletions
+128
View File
@@ -0,0 +1,128 @@
# Search Gateway Resilience
## Positioning
SR-02 is a generic Search Gateway capability improvement. It is not tied to education, exams, geography, policy, product search, or any single content vertical.
The goal is to convert Search Gateway from single-shot retrieval into an evidence-first retrieval pipeline:
```text
query normalization
→ source planning
→ official/direct adapters
→ provider fallback
→ extraction
→ evidence cache
→ confidence scoring
→ failure diagnosis
→ human evidence request when needed
```
## Generic capabilities
### 1. Multi-source fallback
Search Gateway should not fail just because one provider is blocked, truncated, rate-limited, or dynamically rendered.
Default fallback chain:
```text
direct_official_adapter
→ official_site_search
→ structured_api_or_feed
→ search_engine_provider
→ authoritative_mirror_or_media
→ cached_evidence
→ user_supplied_evidence_request
```
### 2. Source tiering
Every source is classified before being trusted:
```text
OFFICIAL_PRIMARY
OFFICIAL_SECONDARY
AUTHORITATIVE_MEDIA
SPECIALIZED_DATABASE
GENERAL_WEB
USER_SUPPLIED_EVIDENCE
```
The same tiering model works for public policy, software documentation, finance data, product specs, standards, academic sources, sports, travel, and user-supplied documents.
### 3. Failure diagnosis
Search failures must be structured, not vague.
Examples:
```text
SEARCH_PROVIDER_TIMEOUT
SEARCH_PROVIDER_CAPTCHA
SEARCH_RESULT_TRUNCATED
SOURCE_BLOCKED_BY_CDN
SOURCE_DYNAMIC_PAGE_UNREADABLE
OFFICIAL_SOURCE_FOUND_BUT_EXTRACTION_FAILED
SOURCE_CONFLICT_NEEDS_REVIEW
EVIDENCE_CACHE_STALE
HUMAN_EVIDENCE_REQUIRED
```
### 4. Evidence cache
A successful retrieval should leave structured evidence behind:
```yaml
evidence_id: ev-...
normalized_query: ...
source_tier: OFFICIAL_PRIMARY
publisher: ...
url: ...
extracted_at: ...
published_at: ...
answer_fields: {...}
confidence: high
```
Cache entries must carry staleness rules by topic type. Time-sensitive information should expire faster than static references.
### 5. Confidence scoring
Search Gateway should return both the answer and confidence metadata:
```json
{
"answer": {},
"confidence": 0.95,
"confidence_level": "high",
"sources": [],
"fallback_used": true,
"failure_codes": [],
"evidence_ids": []
}
```
### 6. Human evidence intake
When automated retrieval is blocked, Search Gateway should ask for a screenshot, URL, PDF, or pasted text and ingest it as `USER_SUPPLIED_EVIDENCE`, rather than silently failing.
## Non-goals
- No vertical-specific hardcoding.
- No CAPTCHA bypass.
- No access-control bypass.
- No scraping of protected/private content.
- No treating search snippets as authoritative without source classification.
## Acceptance
SR-02 is accepted when:
```text
AIOS_SEARCH_GATEWAY_MULTI_SOURCE_READY
AIOS_SEARCH_GATEWAY_FAILURE_DIAGNOSIS_READY
AIOS_SEARCH_GATEWAY_EVIDENCE_CACHE_READY
AIOS_SEARCH_GATEWAY_CONFIDENCE_SCORING_READY
AIOS_SEARCH_GATEWAY_GENERIC_RESILIENCE_READY
```
@@ -0,0 +1,9 @@
{
"type": "branch",
"branch": "feature/v1r4c4-sr02-search-gateway-resilience-20260710",
"base": "main",
"created_by": "chatgpt-gitops-bot",
"created_at": "2026-07-10T10:41:20Z",
"direct_main_push": false,
"owner_approval_required": true
}
+118
View File
@@ -0,0 +1,118 @@
version: 1
name: search-gateway-resilience
baseline: V1R4C3
created_at: "2026-07-10"
scope: generic_search_capability
status: proposed
purpose: >
Upgrade Search Gateway from single-shot search into a general-purpose,
multi-source, evidence-first retrieval system with fallback, diagnostics,
source confidence scoring, and reusable evidence cache.
non_goals:
- Do not hardcode education, exam, province, or any single domain logic.
- Do not depend on a single search engine.
- Do not treat scraped content as trusted without source classification.
- Do not bypass robots, authentication, CAPTCHA, paywalls, or access controls.
source_priority:
OFFICIAL_PRIMARY:
confidence_base: 0.95
examples:
- official_government_site
- official_regulator_site
- official_organization_site
- official_product_or_vendor_docs
OFFICIAL_SECONDARY:
confidence_base: 0.90
examples:
- official_mirror
- official_press_release_distribution
- official_public_archive
AUTHORITATIVE_MEDIA:
confidence_base: 0.80
examples:
- established_newsroom
- recognized_wire_service
- domain_authority_publication
SPECIALIZED_DATABASE:
confidence_base: 0.78
examples:
- standards_database
- academic_index
- legal_or_financial_database
- statistical_repository
GENERAL_WEB:
confidence_base: 0.55
examples:
- search_engine_result
- blog
- forum
- social_excerpt
USER_SUPPLIED_EVIDENCE:
confidence_base: 0.50
examples:
- screenshot
- document_upload
- manual_link
fallback_chain:
- direct_official_adapter
- official_site_search
- structured_api_or_feed
- search_engine_provider
- authoritative_mirror_or_media
- cached_evidence
- user_supplied_evidence_request
failure_codes:
- SEARCH_PROVIDER_TIMEOUT
- SEARCH_PROVIDER_RATE_LIMITED
- SEARCH_PROVIDER_CAPTCHA
- SEARCH_RESULT_TRUNCATED
- SEARCH_RESULT_LOW_RELEVANCE
- SOURCE_BLOCKED_BY_CDN
- SOURCE_404_OR_REMOVED
- SOURCE_DYNAMIC_PAGE_UNREADABLE
- SOURCE_EXTRACTION_FAILED
- OFFICIAL_SOURCE_FOUND_BUT_EXTRACTION_FAILED
- SOURCE_CONFLICT_NEEDS_REVIEW
- EVIDENCE_CACHE_MISS
- EVIDENCE_CACHE_STALE
- HUMAN_EVIDENCE_REQUIRED
result_contract:
required_fields:
- answer
- confidence
- sources
- fallback_used
- failure_codes
- verified_at
- evidence_ids
confidence_levels:
high: ">=0.85 with official or multi-source corroboration"
medium: ">=0.65 and <0.85 with credible but incomplete evidence"
low: "<0.65 or unresolved source/extraction limitations"
cache_policy:
evidence_cache_enabled: true
cache_key_fields:
- normalized_query
- topic_type
- jurisdiction_or_scope
- effective_date
stale_after_days_by_topic:
news: 7
policy_or_law: 30
product_or_price: 3
static_reference: 180
default: 14
acceptance_signals:
- AIOS_SEARCH_GATEWAY_MULTI_SOURCE_READY
- AIOS_SEARCH_GATEWAY_FAILURE_DIAGNOSIS_READY
- AIOS_SEARCH_GATEWAY_EVIDENCE_CACHE_READY
- AIOS_SEARCH_GATEWAY_CONFIDENCE_SCORING_READY
- AIOS_SEARCH_GATEWAY_GENERIC_RESILIENCE_READY
@@ -0,0 +1,147 @@
"""Generic Search Gateway resilience contract.
This module is intentionally domain-agnostic. It defines shared result,
failure, source, cache, and confidence primitives that concrete providers and
adapters can use without hardcoding a specific vertical.
"""
from __future__ import annotations
from dataclasses import dataclass, field
from datetime import datetime, timezone
from enum import Enum
from typing import Any
class SourceTier(str, Enum):
OFFICIAL_PRIMARY = "OFFICIAL_PRIMARY"
OFFICIAL_SECONDARY = "OFFICIAL_SECONDARY"
AUTHORITATIVE_MEDIA = "AUTHORITATIVE_MEDIA"
SPECIALIZED_DATABASE = "SPECIALIZED_DATABASE"
GENERAL_WEB = "GENERAL_WEB"
USER_SUPPLIED_EVIDENCE = "USER_SUPPLIED_EVIDENCE"
class FailureCode(str, Enum):
SEARCH_PROVIDER_TIMEOUT = "SEARCH_PROVIDER_TIMEOUT"
SEARCH_PROVIDER_RATE_LIMITED = "SEARCH_PROVIDER_RATE_LIMITED"
SEARCH_PROVIDER_CAPTCHA = "SEARCH_PROVIDER_CAPTCHA"
SEARCH_RESULT_TRUNCATED = "SEARCH_RESULT_TRUNCATED"
SEARCH_RESULT_LOW_RELEVANCE = "SEARCH_RESULT_LOW_RELEVANCE"
SOURCE_BLOCKED_BY_CDN = "SOURCE_BLOCKED_BY_CDN"
SOURCE_404_OR_REMOVED = "SOURCE_404_OR_REMOVED"
SOURCE_DYNAMIC_PAGE_UNREADABLE = "SOURCE_DYNAMIC_PAGE_UNREADABLE"
SOURCE_EXTRACTION_FAILED = "SOURCE_EXTRACTION_FAILED"
OFFICIAL_SOURCE_FOUND_BUT_EXTRACTION_FAILED = "OFFICIAL_SOURCE_FOUND_BUT_EXTRACTION_FAILED"
SOURCE_CONFLICT_NEEDS_REVIEW = "SOURCE_CONFLICT_NEEDS_REVIEW"
EVIDENCE_CACHE_MISS = "EVIDENCE_CACHE_MISS"
EVIDENCE_CACHE_STALE = "EVIDENCE_CACHE_STALE"
HUMAN_EVIDENCE_REQUIRED = "HUMAN_EVIDENCE_REQUIRED"
SOURCE_CONFIDENCE_BASE: dict[SourceTier, float] = {
SourceTier.OFFICIAL_PRIMARY: 0.95,
SourceTier.OFFICIAL_SECONDARY: 0.90,
SourceTier.AUTHORITATIVE_MEDIA: 0.80,
SourceTier.SPECIALIZED_DATABASE: 0.78,
SourceTier.GENERAL_WEB: 0.55,
SourceTier.USER_SUPPLIED_EVIDENCE: 0.50,
}
@dataclass(frozen=True)
class EvidenceSource:
url: str
title: str
tier: SourceTier
extracted_at: str
publisher: str | None = None
published_at: str | None = None
evidence_id: str | None = None
extraction_method: str | None = None
metadata: dict[str, Any] = field(default_factory=dict)
@property
def base_confidence(self) -> float:
return SOURCE_CONFIDENCE_BASE[self.tier]
@dataclass(frozen=True)
class SearchGatewayResult:
answer: dict[str, Any]
sources: tuple[EvidenceSource, ...]
failure_codes: tuple[FailureCode, ...] = ()
fallback_used: bool = False
evidence_ids: tuple[str, ...] = ()
verified_at: str = field(default_factory=lambda: datetime.now(timezone.utc).isoformat())
confidence: float = 0.0
confidence_level: str = "low"
notes: tuple[str, ...] = ()
def confidence_level(score: float) -> str:
if score >= 0.85:
return "high"
if score >= 0.65:
return "medium"
return "low"
def score_sources(sources: tuple[EvidenceSource, ...], failure_codes: tuple[FailureCode, ...] = ()) -> float:
"""Compute a conservative confidence score from source tiers and failures."""
if not sources:
return 0.0
best = max(source.base_confidence for source in sources)
official_count = sum(
1 for source in sources if source.tier in {SourceTier.OFFICIAL_PRIMARY, SourceTier.OFFICIAL_SECONDARY}
)
corroboration_bonus = min(0.08, max(0, len(sources) - 1) * 0.03)
official_bonus = 0.03 if official_count >= 1 and len(sources) >= 2 else 0.0
penalty = 0.0
if FailureCode.SOURCE_CONFLICT_NEEDS_REVIEW in failure_codes:
penalty += 0.25
if FailureCode.SEARCH_RESULT_TRUNCATED in failure_codes:
penalty += 0.05
if FailureCode.SOURCE_EXTRACTION_FAILED in failure_codes:
penalty += 0.08
if FailureCode.EVIDENCE_CACHE_STALE in failure_codes:
penalty += 0.10
if FailureCode.HUMAN_EVIDENCE_REQUIRED in failure_codes:
penalty += 0.15
score = best + corroboration_bonus + official_bonus - penalty
return max(0.0, min(0.99, round(score, 3)))
def build_result(
answer: dict[str, Any],
sources: tuple[EvidenceSource, ...],
failure_codes: tuple[FailureCode, ...] = (),
fallback_used: bool = False,
notes: tuple[str, ...] = (),
) -> SearchGatewayResult:
score = score_sources(sources, failure_codes)
evidence_ids = tuple(source.evidence_id for source in sources if source.evidence_id)
return SearchGatewayResult(
answer=answer,
sources=sources,
failure_codes=failure_codes,
fallback_used=fallback_used,
evidence_ids=evidence_ids,
confidence=score,
confidence_level=confidence_level(score),
notes=notes,
)
def classify_http_or_extraction_failure(status_code: int | None = None, *, dynamic_page: bool = False, captcha: bool = False) -> FailureCode:
if captcha:
return FailureCode.SEARCH_PROVIDER_CAPTCHA
if status_code == 403:
return FailureCode.SOURCE_BLOCKED_BY_CDN
if status_code == 404:
return FailureCode.SOURCE_404_OR_REMOVED
if dynamic_page:
return FailureCode.SOURCE_DYNAMIC_PAGE_UNREADABLE
return FailureCode.SOURCE_EXTRACTION_FAILED
@@ -0,0 +1,93 @@
from runtime.search_gateway.resilience_contract import (
EvidenceSource,
FailureCode,
SourceTier,
build_result,
classify_http_or_extraction_failure,
confidence_level,
score_sources,
)
def source(tier: SourceTier, evidence_id: str = "ev-1") -> EvidenceSource:
return EvidenceSource(
url="https://example.test/source",
title="Example source",
tier=tier,
extracted_at="2026-07-10T00:00:00+00:00",
evidence_id=evidence_id,
)
def test_official_primary_scores_high():
result = build_result(
answer={"value": 1},
sources=(source(SourceTier.OFFICIAL_PRIMARY),),
)
assert result.confidence >= 0.85
assert result.confidence_level == "high"
def test_general_web_scores_low_without_corroboration():
result = build_result(
answer={"value": 1},
sources=(source(SourceTier.GENERAL_WEB),),
)
assert result.confidence_level == "low"
def test_multiple_sources_add_corroboration_bonus():
one = score_sources((source(SourceTier.AUTHORITATIVE_MEDIA, "ev-1"),))
two = score_sources((source(SourceTier.AUTHORITATIVE_MEDIA, "ev-1"), source(SourceTier.GENERAL_WEB, "ev-2")))
assert two > one
def test_conflict_penalty_lowers_score():
clean = score_sources((source(SourceTier.OFFICIAL_PRIMARY),))
conflicted = score_sources((source(SourceTier.OFFICIAL_PRIMARY),), (FailureCode.SOURCE_CONFLICT_NEEDS_REVIEW,))
assert conflicted < clean
def test_build_result_carries_failure_codes_and_fallback_flag():
result = build_result(
answer={"value": 1},
sources=(source(SourceTier.OFFICIAL_SECONDARY),),
failure_codes=(FailureCode.SEARCH_RESULT_TRUNCATED,),
fallback_used=True,
)
assert result.fallback_used is True
assert FailureCode.SEARCH_RESULT_TRUNCATED in result.failure_codes
def test_build_result_collects_evidence_ids():
result = build_result(
answer={"value": 1},
sources=(source(SourceTier.OFFICIAL_PRIMARY, "ev-a"), source(SourceTier.AUTHORITATIVE_MEDIA, "ev-b")),
)
assert result.evidence_ids == ("ev-a", "ev-b")
def test_confidence_level_boundaries():
assert confidence_level(0.85) == "high"
assert confidence_level(0.65) == "medium"
assert confidence_level(0.64) == "low"
def test_classify_403_as_cdn_blocked():
assert classify_http_or_extraction_failure(status_code=403) == FailureCode.SOURCE_BLOCKED_BY_CDN
def test_classify_404_as_removed():
assert classify_http_or_extraction_failure(status_code=404) == FailureCode.SOURCE_404_OR_REMOVED
def test_classify_dynamic_page_as_unreadable():
assert classify_http_or_extraction_failure(dynamic_page=True) == FailureCode.SOURCE_DYNAMIC_PAGE_UNREADABLE
def test_classify_captcha_as_provider_captcha():
assert classify_http_or_extraction_failure(captcha=True) == FailureCode.SEARCH_PROVIDER_CAPTCHA
def test_no_sources_scores_zero():
assert score_sources(()) == 0.0
@@ -0,0 +1,47 @@
{
"sr_id": "V1R4C4-SR02",
"title": "Generic Search Gateway Resilience",
"baseline": "V1R4C3",
"scope": "generic_search_capability",
"status": "merge_candidate",
"risk": "medium",
"owner_gate_required": true,
"summary": "Upgrade Search Gateway into a domain-agnostic, evidence-first retrieval pipeline with multi-source fallback, failure diagnosis, evidence cache, and confidence scoring.",
"non_goals": [
"No education-specific hardcoding",
"No geography-specific hardcoding",
"No CAPTCHA bypass",
"No access-control bypass",
"No single search engine dependency"
],
"capabilities": [
"multi_source_fallback",
"source_tier_classification",
"failure_diagnosis_codes",
"evidence_cache_contract",
"confidence_scoring",
"human_evidence_intake"
],
"files": [
"policies/search-gateway-resilience.yaml",
"runtime/search_gateway/resilience_contract.py",
"tests/test_search_gateway_resilience_contract.py",
"docs/search-gateway-resilience.md"
],
"acceptance_tests": [
"tests/test_search_gateway_resilience_contract.py"
],
"acceptance_signals": [
"AIOS_SEARCH_GATEWAY_MULTI_SOURCE_READY",
"AIOS_SEARCH_GATEWAY_FAILURE_DIAGNOSIS_READY",
"AIOS_SEARCH_GATEWAY_EVIDENCE_CACHE_READY",
"AIOS_SEARCH_GATEWAY_CONFIDENCE_SCORING_READY",
"AIOS_SEARCH_GATEWAY_GENERIC_RESILIENCE_READY"
],
"next_steps_after_merge": [
"Bind concrete providers to the generic contract",
"Implement evidence cache storage adapter",
"Add provider-level fallback orchestration",
"Run existing Task Center and Search Gateway regression suite"
]
}