ADR-0039: Knowledge Substrate Minimal Interface¶

from typing import Protocol, runtime_checkable

@runtime_checkable
class KnowledgeStore(Protocol):
    """Minimal protocol for durable knowledge persistence.

    Implementations:
      - NullKnowledgeStore: no-op (library default, zero overhead)
      - MemoryKnowledgeStore: in-process dict (testing, short-lived scripts)
      - SQLiteKnowledgeStore: local file (Studio mode)
      - RemoteKnowledgeStore: governed remote store
    """

    async def write_claim(
        self,
        claim: "Claim",
    ) -> str:
        """Persist a claim. Returns claim_id."""
        ...

    async def query(
        self,
        question: str,
        *,
        scope: "Scope | None" = None,
        status: "list[str] | None" = None,
        limit: int = 10,
    ) -> "list[Claim]":
        """Semantic retrieval of relevant claims."""
        ...

    async def transition(
        self,
        claim_id: str,
        new_status: str,
        *,
        evidence: "list[EvidenceRef] | None" = None,
        reason: str | None = None,
    ) -> None:
        """Move a claim through its lifecycle."""
        ...

    async def supersede(
        self,
        old_id: str,
        new_id: str,
    ) -> None:
        """Mark old claim superseded by new one."""
        ...

    async def history(
        self,
        claim_id: str,
    ) -> "list[ClaimEvent]":
        """Return lifecycle events for a claim."""
        ...

from dataclasses import dataclass, field
from uuid import uuid4

@dataclass
class Claim:
    """A unit of learned knowledge with evidence and lifecycle."""

    text: str
    evidence: list[EvidenceRef]  # MUST be non-empty (enforced at construction)
    claim_status: str = "observed"
    confidence: float = 1.0
    # Confidence in the CLAIM being true (the fact itself).
    # Distinct from StateReason.confidence in ADR-0033, which measures
    # confidence in a reason explaining an operational state.

    # Auto-populated by Branch.learn()
    id: str = field(default_factory=lambda: str(uuid4()))
    project: str | None = None
    scope_type: str | None = None  # project | repo | agent | run
    scope_id: str | None = None
    session_id: str | None = None
    branch_id: str | None = None
    actor_type: str = "agent"     # agent | user | system | tool
    actor_id: str | None = None
    created_at: float = 0.0

    # Optional structured metadata
    domain: str | None = None     # e.g., "codebase", "api", "policy"
    tags: list[str] = field(default_factory=list)
    supersedes: str | None = None

    def __post_init__(self):
        if not self.evidence:
            raise ValueError(
                "A Claim requires at least one EvidenceRef. "
                "If evidence is truly absent, this is a hypothesis — "
                "use kind='model_inference' with low confidence."
            )

class Branch:
    def __init__(
        self,
        ...,
        knowledge_store: KnowledgeStore | None = None,
    ):
        ...
        self._knowledge_store = knowledge_store or NullKnowledgeStore()

    async def learn(
        self,
        claim: str,
        evidence: list[EvidenceRef],
        *,
        status: str = "observed",
        confidence: float = 1.0,
        domain: str | None = None,
        tags: list[str] | None = None,
    ) -> str | None:
        """Record a learned fact with evidence. Returns claim_id.

        This is the primary knowledge-write API. It enforces:
        - Non-empty evidence (raises ValueError)
        - Auto-populates session/branch/actor context
        - Delegates to the configured KnowledgeStore

        Returns None if store is NullKnowledgeStore (library mode).
        """
        c = Claim(
            text=claim,
            evidence=evidence,
            claim_status=status,
            confidence=confidence,
            session_id=str(self._session_id) if self._session_id else None,
            branch_id=str(self.id),
            domain=domain,
            tags=tags or [],
            created_at=now_utc().timestamp(),
        )
        return await self._knowledge_store.write_claim(c)

    async def recall(
        self,
        question: str,
        *,
        limit: int = 5,
        status: list[str] | None = None,
    ) -> list[Claim]:
        """Retrieve relevant knowledge claims.

        Default: returns only active claims (observed, inferred, verified).
        Pass status=["disputed"] to include contested knowledge.
        """
        if status is None:
            status = ["observed", "inferred", "verified"]
        return await self._knowledge_store.query(
            question, status=status, limit=limit
        )

    async def verify(
        self, claim_id: str, evidence: list[EvidenceRef] | None = None
    ) -> None:
        """Upgrade a claim to verified status."""
        await self._knowledge_store.transition(
            claim_id, "verified", evidence=evidence
        )

    async def dispute(
        self, claim_id: str, reason: str, evidence: list[EvidenceRef] | None = None
    ) -> None:
        """Mark a claim as disputed with reason."""
        await self._knowledge_store.transition(
            claim_id, "disputed", evidence=evidence, reason=reason
        )

session = Session(
    knowledge_store=SQLiteKnowledgeStore("~/.lionagi/knowledge.db"),
)
branch = session.new_branch(system="You are a researcher")
# branch._knowledge_store == session._knowledge_store

isolated_branch = session.new_branch(
    system="Sandbox agent",
    knowledge_store=NullKnowledgeStore(),  # override
)

class NullKnowledgeStore:
    """No-op store. Library mode. Zero overhead."""

    async def write_claim(self, claim: Claim) -> str | None:
        return None

    async def query(self, question, **kwargs) -> list[Claim]:
        return []

    async def transition(self, claim_id, new_status, **kwargs) -> None:
        pass

    async def supersede(self, old_id, new_id) -> None:
        pass

    async def history(self, claim_id) -> list:
        return []

{
  "event": "knowledge.learn" | "knowledge.verify" | "knowledge.dispute" | "knowledge.supersede",
  "claim_id": str,
  "claim_status": str,
  "evidence_count": int,
  "evidence_kinds": list[str],
  "scope_type": str | None,
  "scope_id": str | None,
  "actor_type": str,
  "actor_id": str | None,
  "branch_id": str,
  "session_id": str | None,
  "timestamp": float,
}

class MemoryKnowledgeStore:
    """In-process dict store. For tests and short-lived scripts."""

    def __init__(self):
        self._claims: dict[str, Claim] = {}

    async def write_claim(self, claim: Claim) -> str:
        self._claims[claim.id] = claim
        return claim.id

    async def query(self, question, *, scope=None, status=None, limit=10):
        results = list(self._claims.values())
        if status:
            results = [c for c in results if c.claim_status in status]
        # Simple substring match for in-memory mode
        if question:
            results = [c for c in results if question.lower() in c.text.lower()]
        return results[:limit]

    async def transition(self, claim_id, new_status, *, evidence=None, reason=None):
        if claim_id in self._claims:
            self._claims[claim_id].claim_status = new_status

    async def supersede(self, old_id, new_id):
        if old_id in self._claims:
            self._claims[old_id].claim_status = "superseded"
            self._claims[old_id].supersedes = new_id

    async def history(self, claim_id):
        return []  # In-memory mode doesn't track history

def from_message(session_id: str, message_id: str, quote: str | None = None) -> EvidenceRef:
    """Evidence from a chat message. The `quote` becomes EvidenceRef.detail."""
    return EvidenceRef(kind="message", session_id=session_id, message_id=message_id, detail=quote)

def from_tool_result(tool_call_id: str, summary: str | None = None) -> EvidenceRef:
    """Evidence from a tool execution result. The `summary` becomes EvidenceRef.detail."""
    return EvidenceRef(kind="tool_result", tool_call_id=tool_call_id, detail=summary)

def from_user_statement(session_id: str, message_id: str) -> EvidenceRef:
    """Evidence from an explicit user statement."""
    return EvidenceRef(kind="user_statement", session_id=session_id, message_id=message_id)

def from_artifact(artifact_id: str, path: str | None = None) -> EvidenceRef:
    """Evidence from a produced artifact (file/report/output)."""
    return EvidenceRef(kind="artifact", artifact_id=artifact_id, path=path)

def from_file(path: str, repo: str | None = None, commit: str | None = None) -> EvidenceRef:
    """Evidence from a file on disk or in a repo."""
    return EvidenceRef(kind="file", path=path, repo=repo, commit_sha=commit)

def from_url(url: str, fetched_at: float | None = None, content_hash: str | None = None) -> EvidenceRef:
    """Evidence from a web resource."""
    return EvidenceRef(kind="url", url=url, fetched_at=fetched_at, content_hash=content_hash)

def from_model_inference(session_id: str, message_id: str, rationale: str) -> EvidenceRef:
    """Evidence from model reasoning. The weakest kind — be honest about confidence."""
    return EvidenceRef(kind="model_inference", session_id=session_id, message_id=message_id, detail=rationale)

def from_human_assertion(user_id: str, asserted_at: float, detail: str | None = None) -> EvidenceRef:
    """Evidence from a human's external assertion (e.g., 'I verified this manually')."""
    return EvidenceRef(kind="human_assertion", id=user_id, fetched_at=asserted_at, detail=detail)

# Example: auto-extract knowledge from tool results
async def extract_knowledge_hook(branch, message):
    """Hook that fires on every ActionResponse to detect learnable facts."""
    if not isinstance(message, ActionResponse):
        return
    # Custom extraction logic per tool
    for result in message.content.results:
        if result.tool_name == "read_file" and result.output:
            # Application-specific: extract facts from file reads
            pass

# Registration
branch.msgs._on_message_added.append(extract_knowledge_hook)

# Tier 1: Library mode (default)
branch = Branch(system="You are a researcher")
# knowledge_store = NullKnowledgeStore() implicitly

# Tier 2: Local persistence (Studio mode)
from lionagi.knowledge import SQLiteKnowledgeStore
branch = Branch(
    system="...",
    knowledge_store=SQLiteKnowledgeStore("~/.lionagi/knowledge.db"),
)

from lionagi import Branch, Session
from lionagi.knowledge import SQLiteKnowledgeStore, from_file, from_tool_result, from_user_statement

# Session 1, Monday: agent reviews PR #1842
session = Session(
    knowledge_store=SQLiteKnowledgeStore("~/.lionagi/knowledge/payments.db"),
)
branch = session.new_branch(system="You review pull requests for the payments service.")

result = await branch.operate(
    instruction="Review PR #1842, focus on transaction handling",
    tools=[read_file_tool, github_pr_tool],
)

# Agent observes a pattern, records it
claim_id = await branch.learn(
    "payments-service uses the saga pattern for multi-step transactions",
    evidence=[
        from_file("src/sagas/payment_saga.py", repo="acme/payments", commit="abc123"),
        from_tool_result("tc_pr1842_001", "GitHub PR shows saga.compensate() in error path"),
    ],
    domain="architecture",
    tags=["saga", "transactions"],
)
# Persists to SQLite. claim_id is returned.

# --- Session 2, Wednesday: different agent reviews PR #1851 ---
session2 = Session(knowledge_store=SQLiteKnowledgeStore("~/.lionagi/knowledge/payments.db"))
branch2 = session2.new_branch(system="You review pull requests for the payments service.")

# Before reviewing, agent recalls prior knowledge
context = await branch2.recall("What patterns does this codebase use for transactions?")
# Returns: [Claim(text="payments-service uses the saga pattern...", confidence=1.0, status="observed", ...)]

# Agent uses this context, doesn't need to re-discover the pattern

# Later in the review, agent finds the new PR uses a different pattern
new_claim_id = await branch2.learn(
    "PR #1851 introduces 2PC alongside saga for cross-service transactions",
    evidence=[
        from_file("src/coordinators/two_phase.py", repo="acme/payments", commit="def456"),
        from_tool_result("tc_pr1851_003", "PR description: explicitly mentions 2PC for inventory+payment"),
    ],
    domain="architecture",
    tags=["2pc", "transactions"],
)

# --- Session 3, Friday: operator reviews knowledge ---
# Operator opens Run detail for PR #1851 review
# Knowledge lens shows: "PR #1851 introduces 2PC..."
# Operator clicks "Verify" after checking the PR themselves
await branch2.verify(new_claim_id, evidence=[
    from_user_statement(session_id=str(session2.id), message_id=operator_msg_id),
])
# Status now: verified
# Both claims persist; future agents querying "transactions" get both with appropriate status

def derive_claim_severity(claim: Claim) -> tuple[str, str]:
    """Returns (severity, tone) for a knowledge claim."""

    # Critical: nothing for claims in v1 (claims aren't operational failures)

    # Warning: actively contested or unconfirmed-too-long
    if claim.claim_status == "disputed":
        return ("warning", "warning")
    if claim.claim_status == "hypothesis" and claim.confidence < 0.5:
        return ("warning", "warning")

    # Info: pending validation, active observations
    if claim.claim_status in ("observed", "inferred"):
        return ("info", "info")
    if claim.claim_status == "hypothesis":
        return ("info", "neutral")

    # Success: confirmed
    if claim.claim_status == "verified":
        return ("neutral", "success")

    # Neutral: archived
    if claim.claim_status == "superseded":
        return ("neutral", "neutral")

    return ("neutral", "neutral")