generated from coulomb/repo-seed
156 lines
4.6 KiB
Python
156 lines
4.6 KiB
Python
"""Protocol interfaces for optional Markitect backends."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
from typing import Any, Protocol, runtime_checkable
|
|
|
|
from markitect_tool.backend.engine import (
|
|
BackendCapabilityCheck,
|
|
DependencyEdge,
|
|
DocumentSnapshot,
|
|
ProvenanceEnvelope,
|
|
)
|
|
|
|
|
|
@runtime_checkable
|
|
class SnapshotBackend(Protocol):
|
|
"""Durable parsed-document snapshot backend."""
|
|
|
|
backend_id: str
|
|
|
|
def capabilities(self) -> BackendCapabilityCheck:
|
|
"""Return supported snapshot capabilities."""
|
|
|
|
def put_document(
|
|
self,
|
|
source_path: str | Path,
|
|
content: str,
|
|
parse_options: dict[str, Any] | None = None,
|
|
) -> str:
|
|
"""Persist a parsed document snapshot and return its snapshot id."""
|
|
|
|
def get_snapshot(self, snapshot_id: str) -> DocumentSnapshot:
|
|
"""Return a previously stored snapshot."""
|
|
|
|
def resolve_source(self, source_path: str | Path) -> str | None:
|
|
"""Return the latest snapshot id for a source path."""
|
|
|
|
def diff_snapshot(self, old_id: str, new_id: str) -> dict[str, Any]:
|
|
"""Return a backend-specific snapshot diff."""
|
|
|
|
|
|
@runtime_checkable
|
|
class IndexBackend(Protocol):
|
|
"""Derived index backend for snapshots."""
|
|
|
|
backend_id: str
|
|
|
|
def capabilities(self) -> BackendCapabilityCheck:
|
|
"""Return supported index capabilities."""
|
|
|
|
def build(self, snapshot_ids: list[str], options: dict[str, Any] | None = None) -> dict[str, Any]:
|
|
"""Build derived indexes for snapshots."""
|
|
|
|
def refresh(self, changed_snapshots: list[str]) -> dict[str, Any]:
|
|
"""Refresh derived indexes for changed snapshots."""
|
|
|
|
def query(self, request: dict[str, Any]) -> dict[str, Any]:
|
|
"""Run a backend query and return a common result envelope."""
|
|
|
|
def explain(self, request: dict[str, Any]) -> dict[str, Any]:
|
|
"""Explain a backend query plan."""
|
|
|
|
|
|
@runtime_checkable
|
|
class QueryAdapter(Protocol):
|
|
"""Adapter from stable Markitect requests to backend execution."""
|
|
|
|
name: str
|
|
|
|
def supports(self, selector_or_query: str, target: str | None = None) -> bool:
|
|
"""Return whether the adapter can execute a query."""
|
|
|
|
def execute(self, request: dict[str, Any]) -> dict[str, Any]:
|
|
"""Execute a query and return common results."""
|
|
|
|
def explain(self, request: dict[str, Any]) -> dict[str, Any]:
|
|
"""Explain how a query would execute."""
|
|
|
|
|
|
@runtime_checkable
|
|
class ContextPackageRegistry(Protocol):
|
|
"""Agent-ready context package registry."""
|
|
|
|
registry_id: str
|
|
|
|
def create_package(
|
|
self,
|
|
query_or_manifest: dict[str, Any],
|
|
budget: dict[str, Any] | None = None,
|
|
policy: dict[str, Any] | None = None,
|
|
) -> str:
|
|
"""Create a context package and return its id."""
|
|
|
|
def activate(self, package_id: str, thread_or_workspace: str) -> str:
|
|
"""Activate a context package and return activation id."""
|
|
|
|
def deactivate(self, activation_id: str) -> None:
|
|
"""Deactivate an active context package."""
|
|
|
|
def refresh(self, package_id: str) -> str:
|
|
"""Refresh a package and return the new package id."""
|
|
|
|
def explain(self, package_id: str) -> dict[str, Any]:
|
|
"""Return package provenance, budget, and retrieval details."""
|
|
|
|
|
|
@runtime_checkable
|
|
class AccessPolicyGateway(Protocol):
|
|
"""Authorization and filtering gateway for backend results."""
|
|
|
|
gateway_id: str
|
|
|
|
def authorize(
|
|
self,
|
|
subject: str,
|
|
action: str,
|
|
object_id: str,
|
|
context: dict[str, Any] | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Authorize one action against one object."""
|
|
|
|
def filter_results(
|
|
self,
|
|
subject: str,
|
|
action: str,
|
|
results: list[dict[str, Any]],
|
|
context: dict[str, Any] | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Filter results and return policy decisions."""
|
|
|
|
def explain_decision(self, decision_id: str) -> dict[str, Any]:
|
|
"""Explain one policy decision."""
|
|
|
|
|
|
@runtime_checkable
|
|
class ProcessorResultStore(Protocol):
|
|
"""Optional store for deterministic or assisted processor outputs."""
|
|
|
|
store_id: str
|
|
|
|
def put_result(
|
|
self,
|
|
processor_name: str,
|
|
input_hash: str,
|
|
result: dict[str, Any],
|
|
provenance: ProvenanceEnvelope,
|
|
) -> str:
|
|
"""Persist a processor result and return its result id."""
|
|
|
|
def get_result(self, result_id: str) -> dict[str, Any]:
|
|
"""Return a processor result."""
|
|
|
|
def dependencies(self, result_id: str) -> list[DependencyEdge]:
|
|
"""Return dependencies for a processor result."""
|