Python (rag_plusplus.slice)

┌─────────────────────────────────────────────────────────────────┐
│                        RAG++ Retrieval                          │
├─────────────────────────────┬───────────────────────────────────┤
│       Slice-RAG             │          Global-RAG               │
│   (Admissibility-Gated)     │      (Explicit Exception)         │
├─────────────────────────────┼───────────────────────────────────┤
│ • Anchor-first              │ • No anchor required              │
│ • Slice-scoped candidates   │ • Whole-fabric search             │
│ • is_admissible = true      │ • is_admissible = false           │
│ • Can trigger promotions    │ • Cannot trigger promotions       │
│ • Full provenance chain     │ • Provenance (but marked global)  │
└─────────────────────────────┴───────────────────────────────────┘

┌──────────────────────────────────────────────────────────────────────────┐
│                              Client Application                           │
│                    (trajectory-search, Orbit, Claude Code)                │
└─────────────────────────────────┬────────────────────────────────────────┘
                                  │
                                  ▼
┌──────────────────────────────────────────────────────────────────────────┐
│                           RAG++ Service (Python)                          │
│                              Port 8000                                    │
│  ┌─────────────────┐  ┌─────────────────┐  ┌─────────────────────────┐   │
│  │  SliceClient    │  │ MemoryRetriever │  │   DualPlaneRetriever    │   │
│  │                 │  │                 │  │                         │   │
│  │ • get_slice()   │  │ • search_slice()│  │ • raw_turns            │   │
│  │ • get_batch()   │  │ • search_global()│ │ • atoms/proposals      │   │
│  └────────┬────────┘  └────────┬────────┘  └────────────────────────┘   │
│           │                    │                                         │
│           │                    ▼                                         │
│           │         ┌─────────────────────┐                              │
│           │         │ RetrievalProvenance │                              │
│           │         │ • query_id          │                              │
│           │         │ • slice_id          │                              │
│           │         │ • is_admissible     │                              │
│           │         └─────────────────────┘                              │
└───────────┼──────────────────────────────────────────────────────────────┘
            │
            ▼
┌──────────────────────────────────────────────────────────────────────────┐
│                     Graph Kernel Service (Rust)                           │
│                              Port 8001                                    │
│  ┌─────────────────┐  ┌─────────────────┐  ┌─────────────────────────┐   │
│  │  ContextSlicer  │  │ PolicyRegistry  │  │     PostgresGraphStore  │   │
│  │                 │  │                 │  │                         │   │
│  │ • slice()       │  │ • register()    │  │ • get_turn()            │   │
│  │ • fingerprint   │  │ • resolve()     │  │ • get_parents()         │   │
│  └─────────────────┘  └─────────────────┘  └─────────────────────────┘   │
└──────────────────────────────────────────────────────────────────────────┘
            │
            ▼
┌──────────────────────────────────────────────────────────────────────────┐
│                           Supabase (PostgreSQL)                           │
│  ┌─────────────────┐  ┌─────────────────┐  ┌─────────────────────────┐   │
│  │  memory_turns   │  │ semantic_atoms  │  │   semantic_proposals    │   │
│  │                 │  │                 │  │                         │   │
│  │ • id (UUID)     │  │ • source_turn_ids│ │ • source_atoms          │   │
│  │ • embedding     │  │ • source_slice_id│ │ • source_slice_id       │   │
│  └─────────────────┘  └─────────────────┘  └─────────────────────────┘   │
└──────────────────────────────────────────────────────────────────────────┘

1. Client sends search request with anchor_turn_id
                    │
                    ▼
2. RAG++ SliceClient calls Graph Kernel: POST /api/slice
                    │
                    ▼
3. Graph Kernel builds SliceExport:
   • Expands from anchor using policy
   • Computes deterministic fingerprint
   • Returns ordered turn_ids + edges
                    │
                    ▼
4. RAG++ MemoryRetriever.search_slice():
   • Filters candidates to admissible_turn_ids
   • Executes vector search
   • Attaches provenance chain
                    │
                    ▼
5. Results returned with full provenance:
   • query_id, slice_id, anchor_turn_id
   • is_admissible = true
   • Can be used for semantic promotions

// Rust (cc-graph-kernel)
pub struct SliceExport {
    pub slice_id: SliceFingerprint,     // Deterministic hash
    pub anchor_turn_id: TurnId,          // Anchor turn
    pub turns: Vec<TurnSnapshot>,        // Ordered turns
    pub edges: Vec<Edge>,                // Ordered edges
    pub policy_id: String,               // e.g., "slice_policy_v1"
    pub policy_params_hash: String,      // xxHash64 of policy JSON
    pub schema_version: String,          // Graph Kernel schema version
}

# Python (rag_plusplus.slice)
@dataclass
class SliceExport:
    slice_id: str
    anchor_turn_id: str
    turn_ids: List[str]
    edge_count: int
    policy_id: str
    policy_params_hash: str
    schema_version: str

    def is_admissible(self, turn_id: str) -> bool:
        """Check if a turn is in this slice."""
        return turn_id in self._turn_id_set

@dataclass(frozen=True)
class PolicyRef:
    policy_id: str      # "slice_policy_v1"
    params_hash: str    # xxHash64 of canonical JSON

@dataclass
class RetrievalProvenance:
    query_id: str                    # Unique per request
    retrieval_mode: Literal["slice", "global"]
    query_text: str
    retrieved_turn_ids: List[str]
    timestamp: str
    elapsed_ms: float

    # Slice-specific (None for global)
    slice_info: Optional[SliceInfo]

    # For replay verification
    query_embedding_hash: Optional[str]
    result_hash: Optional[str]

    @property
    def is_admissible(self) -> bool:
        """Only slice-mode results are admissible."""
        return self.retrieval_mode == "slice" and self.slice_info is not None

core/cc-graph-kernel/
├── Cargo.toml                 # Added 'service' feature
├── Dockerfile.service         # Production container
└── src/
    ├── lib.rs                 # Exports service module
    ├── bin/
    │   └── graph_kernel_service.rs  # Service binary
    └── service/
        ├── mod.rs             # Module entry point
        ├── state.rs           # PolicyRegistry, ServiceState
        └── routes.rs          # Axum handlers

pub struct PolicyRegistry {
    policies: BTreeMap<PolicyRef, SlicePolicyV1>,
    registry_fingerprint: String,
}

impl PolicyRegistry {
    /// Register a policy, returning its reference.
    pub fn register(&mut self, policy: SlicePolicyV1) -> PolicyRef {
        let policy_ref = PolicyRef::from_policy(&policy);
        if !self.policies.contains_key(&policy_ref) {
            self.policies.insert(policy_ref.clone(), policy);
            self.update_fingerprint();
        }
        policy_ref
    }

    /// Resolve a reference to the actual policy.
    pub fn resolve(&self, policy_ref: &PolicyRef) -> Option<&SlicePolicyV1> {
        self.policies.get(policy_ref)
    }
}

{
  "anchor_turn_id": "550e8400-e29b-41d4-a716-446655440000",
  "policy_ref": {
    "policy_id": "slice_policy_v1",
    "params_hash": "abc123..."
  }
}

{
  "slice": {
    "slice_id": "slice_fingerprint_hash",
    "anchor_turn_id": "550e8400-...",
    "turn_ids": ["turn1", "turn2", "turn3"],
    "edge_count": 2,
    "policy_id": "slice_policy_v1",
    "policy_params_hash": "abc123...",
    "schema_version": "1.0.0"
  },
  "policy_ref": {
    "policy_id": "slice_policy_v1",
    "params_hash": "abc123..."
  }
}

# Development
cd core/cc-graph-kernel
DATABASE_URL=postgresql://... cargo run --bin graph_kernel_service --features service

# Production
docker build -f Dockerfile.service -t graph-kernel .
docker run -p 8001:8001 -e DATABASE_URL=... graph-kernel

rag_plusplus/slice/
├── __init__.py      # Exports SliceClient, SliceExport, PolicyRef, AnchorResolver
├── client.py        # SliceClient for Graph Kernel communication
└── anchor.py        # AnchorResolver for query-to-anchor mapping

from rag_plusplus.slice import SliceClient, PolicyRef

client = SliceClient(
    base_url="http://localhost:8001",  # Or GRAPH_KERNEL_URL env var
    timeout=30.0,
)

# Get a slice for an anchor
slice_export = await client.get_slice(
    anchor_turn_id="550e8400-e29b-41d4-a716-446655440000",
    policy_ref=PolicyRef("slice_policy_v1", "abc123"),  # Optional
)

# Batch slicing for atlas operations
slices = await client.get_slice_batch(
    anchor_turn_ids=["turn1", "turn2", "turn3"],
    policy_ref=policy_ref,
)

# Register a custom policy
custom_policy_ref = await client.register_policy(
    max_nodes=128,
    max_radius=5,
    include_siblings=True,
)

from rag_plusplus.slice import AnchorResolver

resolver = AnchorResolver(supabase_client)

# Explicit anchor
anchor = await resolver.resolve_explicit("turn_uuid")

# From session (latest turn)
anchor = await resolver.resolve_from_session("session_id")

# Semantic match (find best anchor for query)
anchor = await resolver.resolve_from_query(
    query="authentication flow",
    embedding=query_embedding,
)

# Temporal (most recent turns)
anchors = await resolver.resolve_temporal(limit=5, project_id="proj_123")

from rag_plusplus.slice.anchor import AutoAnchorResolver

auto_resolver = AutoAnchorResolver(resolver)

# Tries in order: explicit → session → semantic → temporal
anchor = await auto_resolver.resolve(
    turn_id=None,        # No explicit anchor
    session_id="sess_1", # Try session first
    query="auth flow",   # Fall back to semantic
    embedding=embedding,
)

from rag_plusplus.retrieval import SliceScopedQuery

# Create from SliceExport
query = SliceScopedQuery.from_slice_export(
    text="authentication flow",
    slice_export=slice_export,
    limit=10,
)

# Check admissibility
assert query.is_admissible("turn_001") == True
assert query.is_admissible("turn_999") == False  # Not in slice

# Filter candidates
admissible = query.filter_admissible(["turn_001", "turn_999"])
# Returns: ["turn_001"]

from rag_plusplus.retrieval import MemoryRetriever

retriever = MemoryRetriever()

# Slice-conditioned search
results = retriever.search_slice(
    query="authentication flow",
    slice_export=slice_export,
    limit=10,
)

# Results are SliceScopedResults
assert results.provenance.is_admissible == True
assert results.provenance.slice_id == slice_export.slice_id

for result in results:
    print(f"Turn: {result.turn_id}, Similarity: {result.similarity}")

# Global search (explicit exception mode)
results = retriever.search_global(
    query="authentication flow",
    limit=10,
)

# Results are GlobalResults
assert results.provenance.is_admissible == False
assert results.provenance.slice_id is None

┌─────────────────────────────────────────────────────────────────┐
│                    search_slice() Process                        │
├─────────────────────────────────────────────────────────────────┤
│                                                                  │
│  Stage 1: Admissibility Filtering                               │
│  ─────────────────────────────────                              │
│  • Get slice.turn_ids from SliceExport                          │
│  • Create admissible_turn_ids set                               │
│  • Only these IDs can appear in results                         │
│                                                                  │
│  Stage 2: Vector Search                                         │
│  ──────────────────────                                         │
│  • Generate query embedding                                      │
│  • If |admissible| < 100: Use IN clause (efficient)             │
│  • If |admissible| >= 100: Over-fetch + client filter           │
│  • Rank by similarity within admissible set                      │
│                                                                  │
│  Stage 3: Provenance Construction                               │
│  ────────────────────────────────                               │
│  • Attach slice_id, anchor_turn_id, policy_ref                  │
│  • Hash query embedding for replay verification                  │
│  • Set is_admissible = true                                      │
│                                                                  │
└─────────────────────────────────────────────────────────────────┘

provenance = RetrievalProvenance(
    query_id="uuid-v4",
    retrieval_mode="slice",
    query_text="authentication flow",
    retrieved_turn_ids=["turn_001", "turn_002"],
    timestamp="2026-01-01T12:00:00Z",
    elapsed_ms=42.5,

    slice_info=SliceInfo(
        slice_id="slice_abc123",
        anchor_turn_id="turn_anchor",
        turn_count=50,
        edge_count=49,
        policy_ref=PolicyRef("slice_policy_v1", "hash123"),
        schema_version="1.0.0",
    ),

    query_embedding_hash="sha256_prefix",
    result_hash="sha256_prefix",
    filters_applied={"min_salience": 0.5},
    limit_requested=10,
    limit_returned=2,
)

def can_promote_to_atom(result: SearchResult, provenance: RetrievalProvenance) -> bool:
    """Only admissible results can be promoted to semantic atoms."""
    if not provenance.is_admissible:
        return False

    # Verify result is in the slice
    if str(result.turn_id) not in provenance.retrieved_turn_ids:
        return False

    return True

# Usage
if can_promote_to_atom(result, results.provenance):
    atom = create_semantic_atom(result, source_slice_id=provenance.slice_id)

def verify_replay(
    original_provenance: RetrievalProvenance,
    replay_results: SliceScopedResults,
) -> bool:
    """Verify that a replay produces identical results."""

    # Same slice?
    if original_provenance.slice_id != replay_results.provenance.slice_id:
        return False

    # Same query embedding?
    if original_provenance.query_embedding_hash != replay_results.provenance.query_embedding_hash:
        return False

    # Same results?
    original_ids = set(original_provenance.retrieved_turn_ids)
    replay_ids = set(replay_results.provenance.retrieved_turn_ids)

    return original_ids == replay_ids

@dataclass
class SliceInfo:
    slice_id: str              # Selection fingerprint
    anchor_turn_id: str
    turn_count: int
    edge_count: int
    policy_ref: PolicyRef
    schema_version: str
    graph_snapshot_hash: str   # NEW: Content immutability proof
    admissibility_token: str   # NEW: Unforgeable kernel claim

@dataclass
class RetrievalProvenance:
    # ... existing fields ...
    admissibility_shortfall: bool = False  # NEW: Couldn't fill limit

    @property
    def is_admissible(self) -> bool:
        """Four conditions for admissibility."""
        return (
            self.retrieval_mode == "slice" and
            self.slice_info is not None and
            self.slice_info.has_valid_token() and
            not self.admissibility_shortfall
        )

# Quantization factor: 1e6
# Example: 0.9 → 900000

def quantize_float(value: float) -> int:
    return round(value * 1_000_000)

// Kernel issues token (Rust)
let [sensitive field redacted]
    &slice_id,
    &anchor_turn_id,
    &policy_id,
    &policy_params_hash,
    &graph_snapshot_hash,
    &schema_version,
);

// Computed from table statistics
GraphSnapshotHash::from_stats(
    max_updated_at,  // Latest update timestamp
    turn_count,      // Number of turns
    edge_count,      // Number of edges
    schema_version,  // Graph Kernel version
)

CREATE TABLE retrieval_traces (
    id UUID PRIMARY KEY,
    retrieval_mode TEXT CHECK (retrieval_mode IN ('slice', 'global')),
    is_admissible BOOLEAN NOT NULL,
    slice_id TEXT,
    admissibility_token TEXT,
    -- ...
);

CREATE INDEX idx_admissible ON retrieval_traces (is_admissible);

{
  "query": "authentication flow",
  "anchor_turn_id": "550e8400-e29b-41d4-a716-446655440000",
  "policy_ref": {
    "policy_id": "slice_policy_v1",
    "params_hash": "abc123"
  },
  "limit": 10,
  "min_similarity": 0.5
}

{
  "query": "authentication flow",
  "results": [
    {
      "turn_id": "...",
      "content": "...",
      "role": "assistant",
      "similarity": 0.92,
      "trajectory": {
        "depth": 5,
        "sibling_order": 0,
        "homogeneity": 0.8,
        "temporal": 0.3,
        "complexity": 2
      }
    }
  ],
  "slice_info": {
    "slice_id": "slice_fingerprint",
    "anchor_turn_id": "...",
    "turn_count": 50,
    "edge_count": 49,
    "policy_id": "slice_policy_v1",
    "params_hash": "abc123",
    "schema_version": "1.0.0"
  },
  "provenance": {
    "query_id": "...",
    "retrieval_mode": "slice",
    "is_admissible": true,
    "slice_id": "slice_fingerprint",
    "anchor_turn_id": "...",
    "elapsed_ms": 42.5,
    "limit_requested": 10,
    "limit_returned": 5
  }
}

{
  "query": "authentication flow",
  "results": [...],
  "provenance": {
    "query_id": "...",
    "retrieval_mode": "global",
    "is_admissible": false,
    "slice_id": null,
    "anchor_turn_id": null,
    "elapsed_ms": 35.2,
    "limit_requested": 10,
    "limit_returned": 10
  }
}

# docker-compose.yml (excerpt)
services:
  graph-kernel:
    build:
      context: ../cc-graph-kernel
      dockerfile: Dockerfile.service
    ports:
      - "8001:8001"
    environment:
      - PORT=8001
      - DATABASE_URL=${DATABASE_URL}
      - RUST_LOG=info
    profiles:
      - with-kernel

  backend:
    environment:
      - GRAPH_KERNEL_URL=http://graph-kernel:8001

# Build and push Graph Kernel
cd core/cc-graph-kernel
gcloud builds submit --tag gcr.io/PROJECT/graph-kernel

# Deploy
gcloud run deploy graph-kernel \
  --image gcr.io/PROJECT/graph-kernel \
  --port 8001 \
  --set-env-vars "DATABASE_URL=..." \
  --allow-unauthenticated

# tests/test_slice_retrieval.py

def test_slice_retrieval_determinism():
    """Same anchor + policy -> same slice_id -> same results."""
    query1 = SliceScopedQuery.from_slice_export(
        text="test",
        slice_export=slice_export,
    )
    query2 = SliceScopedQuery.from_slice_export(
        text="test",
        slice_export=slice_export,
    )

    assert query1.slice_id == query2.slice_id
    assert query1.admissible_turn_ids == query2.admissible_turn_ids

def test_global_vs_slice_separation():
    """Global results cannot trigger promotions."""
    slice_prov = RetrievalProvenance.create_slice_provenance(...)
    global_prov = RetrievalProvenance.create_global_provenance(...)

    assert slice_prov.is_admissible == True
    assert global_prov.is_admissible == False

// cc-graph-kernel/tests/service_golden.rs

#[test]
fn test_slice_determinism() {
    let store = InMemoryGraphStore::from_fixture("test_graph.json");
    let policy = SlicePolicyV1::default();
    let slicer = ContextSlicer::new(&store, policy);

    let anchor = TurnId::from_str("anchor_001").unwrap();

    // Run 100 times
    let mut slice_ids = HashSet::new();
    for _ in 0..100 {
        let slice = slicer.slice(anchor.clone()).unwrap();
        slice_ids.insert(slice.slice_id.clone());
    }

    // All runs produce identical slice_id
    assert_eq!(slice_ids.len(), 1);
}

# Python tests
cd core/cc-rag-plus-plus
pytest tests/test_slice_retrieval.py -v

# Rust tests
cd core/cc-graph-kernel
cargo test --features service

results = retriever.search("authentication flow", limit=10)

# 1. Get slice from Graph Kernel
slice_export = await slice_client.get_slice(anchor_turn_id)

# 2. Slice-scoped search
results = retriever.search_slice("authentication flow", slice_export)

# 3. Check admissibility before promotion
if results.provenance.is_admissible:
    promote_to_atom(results)

pub struct SlicePolicyV1 {
    pub version: String,           // "slice_policy_v1"
    pub max_nodes: usize,          // Max turns in slice (256)
    pub max_radius: u32,           // Max hops from anchor (10)
    pub phase_weights: PhaseWeights,
    pub salience_weight: f32,      // [0, 1]
    pub distance_decay: f32,       // Priority decay per hop
    pub include_siblings: bool,
    pub max_siblings_per_node: usize,
}

@dataclass
class SearchResult:
    turn_id: UUID
    content: str
    role: str
    similarity: float
    conversation_id: UUID
    trajectory: TrajectoryCoordinates
    metadata: Dict[str, Any]