API reference

Import path: fg_agent_knowledge. Fully typed (py.typed). All record dataclasses are frozen.

Export surface note: the facade, types, stores, KeywordRetriever, RetrievalQuery/Ranking, signing functions, and the constants CONTRADICTION_WEIGHT / CONFLICT_CONTRADICTION_THRESHOLD / DEFAULT_BRIEFING_LIMIT / DEFAULT_STALENESS_HORIZON_SECONDS plus confidence / staleness_seconds are top-level exports. The remaining scoring helpers, governance internals, eligible_claims, and BaseRetriever are imported from their submodules (fg_agent_knowledge.scoring, .governance, .retrieval, .retrievers).

KnowledgeBase — the facade

The one class a consumer touches.

KnowledgeBase(store: Store, retriever: Retriever | None = None)
# retriever defaults to KeywordRetriever()

Capture

observe(scope: Scope, observer: KeyPair | str, kind: EpisodeKind,
        content: str, refs: tuple[ArtifactRef, ...] = (),
        occurred_at: datetime | None = None) -> Episode

Knowledge lifecycle

propose(scope: Scope, keys: KeyPair, kind: ClaimKind, statement: str,
        topics: tuple[str, ...] = (), refs: tuple[ArtifactRef, ...] = (),
        episodes: tuple[str, ...] = (), supersedes: str | None = None,
        asserted_at: datetime | None = None) -> Promotion

review(promotion_id: str, keys: KeyPair, verdict: ReviewDecision,
       basis: str = "") -> Promotion

endorse(claim_id: str, keys: KeyPair, verdict: EndorsementVerdict,
        basis: str = "", episodes: tuple[str, ...] = (),
        issued_at: datetime | None = None) -> Endorsement

retire(claim_id: str, keys: KeyPair, reason: str) -> Retirement

invalidate(scope: Scope, artifact_uri: str, changed_at: datetime) -> list[str]

Retrieval

brief(scope: Scope, task: str, topics: tuple[str, ...] = (),
      refs: tuple[ArtifactRef, ...] = (), limit: int | None = None,
      source_weight: SourceWeight | None = None) -> Briefing

claim(claim_id: str) -> Claim
claims(scope: Scope) -> list[Claim]
promotions(scope: Scope, status: PromotionStatus | None = None) -> list[Promotion]
conflicts(scope: Scope) -> list[ConflictEvent]

Policy

set_policy(scope: Scope, policy: Policy) -> None
policy(scope: Scope) -> Policy

Types

Scope(space: str, segment: str | None = None)
ArtifactRef(uri: str, kind: str | None = None)
Policy(mode: PolicyMode = "auto", required_approvals: int = 1,
       protected_topics: tuple[str, ...] = (),
       staleness_horizon_seconds: int | None = None)

Records: Episode, ClaimBody, Claim(body, signature, claim_id), EndorsementBody, Endorsement, ReviewVerdictBody, ReviewVerdict, RetirementBody, Retirement, Promotion, ConflictEvent, BriefingItem, Briefing.

Literals:

Constants: SPEC = "fg-agent-knowledge/v1" · MAX_TEXT_BYTES = 65_536.

Signing

DOMAIN = "fg-agent-knowledge/v1"
CONTEXT_CLAIM, CONTEXT_ENDORSEMENT, CONTEXT_VERDICT, CONTEXT_RETIREMENT

signing_input(context: str, payload: Any) -> bytes
    # uint16be(len(tag)) || tag || canonical_json(payload); floats -> ValidationError
record_id(context: str, payload: Any) -> str            # sha256 hex
sign_payload(keys: KeyPair, context: str, payload: Any) -> str    # base64
verify_by_address(address: str, context: str, payload: Any, signature: str) -> None
    # raises SignatureError; rejects non-canonical base64

Record builders / verifiers

# claims
build_claim(keys, scope, kind, statement, topics=(), refs=(), episodes=(),
            asserted_at=None, supersedes=None) -> Claim
claim_id_of(body: ClaimBody) -> str
verify_claim(claim: Claim) -> None    # verifies id-vs-body AND signature-vs-author

# endorsements
build_endorsement(keys, claim_id, verdict, basis="", episodes=(), issued_at=None) -> Endorsement
verify_endorsement(endorsement) -> None

# governance
DEFAULT_POLICY = Policy(mode="auto")
needs_review(claim, policy) -> bool
open_promotion(scope, claim, proposer, policy, now=None) -> Promotion
build_verdict(keys, promotion_id, verdict, basis="", issued_at=None) -> ReviewVerdict
verify_verdict(verdict) -> None
check_reviewable(promotion, reviewer, prior) -> None    # raises PolicyError
decide(promotion, verdicts, policy) -> str | None        # "accepted" | "rejected" | None
build_retirement(keys, claim_id, reason, issued_at=None) -> Retirement
verify_retirement(retirement) -> None

Scoring — pure functions

SourceWeight = Callable[[str], float]

CONTRADICTION_WEIGHT = 2
DEFAULT_STALENESS_HORIZON_SECONDS = 7 * 24 * 3600
REF_OVERLAP_BONUS = 2.0
CONFLICT_CONTRADICTION_THRESHOLD = 2
DEFAULT_BRIEFING_LIMIT = 12

latest_verdicts(endorsements) -> dict[str, Endorsement]
corroborators(endorsements) -> set[str]
contradictors(endorsements) -> set[str]
confidence(corroborations, contradictions, k=CONTRADICTION_WEIGHT) -> float
weighted_support(endorsements, source_weight=None) -> tuple[float, float]
last_corroborated_at(claim, endorsements) -> datetime
staleness_seconds(claim, endorsements, now) -> int
is_suspect(claim, endorsements, invalidations) -> bool
freshness_damping(staleness, horizon_seconds) -> float   # horizon / (horizon + staleness)
briefing_score(relevance, ref_overlap, claim_confidence, staleness, horizon_seconds) -> float

Retrieval seam

@dataclass RetrievalQuery(task: str, topics: tuple[str, ...] = ())
@dataclass Ranking(scores: dict[str, float] = {}, query_empty: bool = False)

class Retriever(Protocol):  # runtime_checkable
    def rank(self, query: RetrievalQuery, claims: Sequence[Claim]) -> Ranking: ...
    def index(self, claim: Claim) -> None: ...
    def remove(self, claim_id: str) -> None: ...

class KeywordRetriever(BaseRetriever):
    def __init__(self, *, k1=1.5, b=0.75, stem=True,
                 remove_stopwords=True, topic_weight=2)
    # Okapi BM25 over Porter-stemmed, stopword-filtered tokens;
    # topic terms weighted; scores normalized to [0,1] per query.

Assembly functions: brief(store, scope, task, ...) -> Briefing, eligible_claims(store, scope).

Store seam

Store (Protocol): add_episode/episode, add_claim/claim/claims, add_endorsement/endorsements/endorsements_for (batched — avoids read-path N+1), add_promotion/promotion/promotions/decide_promotion, add_verdict (returns bool — admits atomically only while pending) /verdicts, add_retirement/retirement, record_invalidation/invalidations, set_policy/policy.

Implementations:

• SQLiteStore(path) — single file, WAL mode, thread-locked, chunked IN-queries.
• InMemoryStore() — dict-backed twin, tested for parity with SQLite.

Errors

KnowledgeError
├── ValidationError    # malformed payloads, floats in signed bodies, oversized text
├── SignatureError     # bad signature or non-canonical encoding
├── NotFoundError      # unknown claim / episode / promotion
├── PolicyError        # self-review, double-review, decided promotions
└── ScopeError         # cross-space episode / supersedes references