Code Story: Creating GitHub Copilot’s Context Engine with Codex 2 and Tree-sitter in 2026

import os
import sys
import json
import logging
from typing import List, Dict, Optional, Tuple
from dataclasses import dataclass

# Initialize logging for error tracking
logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
logger = logging.getLogger(__name__)

# Data class to hold extracted context snippets
@dataclass
class ContextSnippet:
    file_path: str
    language: str
    node_type: str
    start_line: int
    end_line: int
    content: str
    relevance_score: float = 0.0

class TreeSitterContextExtractor:
    """Extracts structured context from code files using Tree-sitter grammars"""

    SUPPORTED_LANGUAGES = {
        "python": "python",
        "javascript": "javascript",
        "typescript": "typescript",
        "go": "go",
        "rust": "rust"
    }

    def __init__(self, grammar_dir: str = "/usr/local/lib/tree-sitter-grammars"):
        self.grammar_dir = grammar_dir
        self.language_cache = {}
        # Validate grammar directory exists
        if not os.path.isdir(grammar_dir):
            logger.warning(f"Grammar directory {grammar_dir} not found, falling back to default")
            self.grammar_dir = os.path.join(os.path.dirname(__file__), "grammars")
            os.makedirs(self.grammar_dir, exist_ok=True)

    def _load_language(self, lang_name: str) -> Optional[object]:
        """Load a Tree-sitter language grammar with caching"""
        if lang_name in self.language_cache:
            return self.language_cache[lang_name]
        if lang_name not in self.SUPPORTED_LANGUAGES:
            logger.error(f"Unsupported language: {lang_name}")
            return None

        try:
            # Load grammar from precompiled .so files (matches https://github.com/tree-sitter/tree-sitter distribution)
            grammar_path = os.path.join(self.grammar_dir, f"tree-sitter-{lang_name}.so")
            if not os.path.exists(grammar_path):
                logger.error(f"Grammar file not found for {lang_name} at {grammar_path}")
                return None
            language = tree_sitter.Language(grammar_path, lang_name)
            self.language_cache[lang_name] = language
            return language
        except Exception as e:
            logger.error(f"Failed to load language {lang_name}: {str(e)}")
            return None

    def extract_context(self, file_path: str, code_content: str) -> List[ContextSnippet]:
        """Parse code content and extract relevant context snippets (functions, classes, imports)"""
        # Determine language from file extension
        ext = os.path.splitext(file_path)[1].lstrip(".")
        lang_name = next((k for k, v in self.SUPPORTED_LANGUAGES.items() if v == ext or k == ext), None)
        if not lang_name:
            logger.warning(f"No supported language found for file {file_path}")
            return []

        language = self._load_language(lang_name)
        if not language:
            return []

        try:
            parser = tree_sitter.Parser()
            parser.set_language(language)
            tree = parser.parse(bytes(code_content, "utf-8"))
        except Exception as e:
            logger.error(f"Failed to parse {file_path}: {str(e)}")
            return []

        snippets = []
        # Define node types to extract per language
        target_nodes = {
            "python": ["function_definition", "class_definition", "import_statement", "import_from_statement"],
            "javascript": ["function_declaration", "class_declaration", "import_statement", "export_statement"],
            "typescript": ["function_declaration", "class_declaration", "interface_declaration", "import_statement"]
        }.get(lang_name, [])

        def traverse(node):
            if node.type in target_nodes:
                start_line = node.start_point[0] + 1  # Tree-sitter uses 0-indexed lines
                end_line = node.end_point[0] + 1
                snippet_content = code_content[node.start_byte:node.end_byte]
                snippets.append(ContextSnippet(
                    file_path=file_path,
                    language=lang_name,
                    node_type=node.type,
                    start_line=start_line,
                    end_line=end_line,
                    content=snippet_content
                ))
            for child in node.children:
                traverse(child)

        traverse(tree.root_node)
        logger.info(f"Extracted {len(snippets)} context snippets from {file_path}")
        return snippets

if __name__ == "__main__":
    # Example usage with a sample Python file
    extractor = TreeSitterContextExtractor()
    sample_code = """
import os
from typing import List

class ContextEngine:
    def __init__(self, cache_size: int = 1024):
        self.cache = {}
        self.cache_size = cache_size

    def get_context(self, file_path: str) -> List[str]:
        return self.cache.get(file_path, [])
"""
    snippets = extractor.extract_context("sample.py", sample_code)
    print(json.dumps([s.__dict__ for s in snippets], indent=2))

import hashlib
import time
import json
import logging
from typing import List, Dict, Optional, Tuple
from dataclasses import dataclass, asdict
import redis
from tenacity import retry, stop_after_attempt, wait_exponential

# Configure logging
logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
logger = logging.getLogger(__name__)

# Reference to Codex 2 client: https://github.com/openai/codex-2
try:
    from openai import Codex2Client
except ImportError:
    logger.warning("Codex2Client not installed, using mock client for demonstration")
    class Codex2Client:
        def __init__(self, api_key: str):
            self.api_key = api_key
        def generate_completion(self, prompt: str, context: List[str], max_tokens: int = 256):
            return {"choices": [{"text": "# Mock completion for testing"}]}

@dataclass
class ContextWindow:
    """Structured context window for Codex 2 inference"""
    window_id: str
    file_path: str
    language: str
    context_snippets: List[Dict]
    max_tokens: int = 4096
    created_at: float = time.time()

    def to_prompt_string(self) -> str:
        """Convert context window to a formatted prompt string for Codex 2"""
        prompt_parts = [
            f"# Context for file: {self.file_path}",
            f"# Language: {self.language}",
            "# Relevant code snippets:"
        ]
        current_tokens = 0
        for snippet in self.context_snippets:
            snippet_str = f"# Lines {snippet['start_line']}-{snippet['end_line']} ({snippet['node_type']})\n{snippet['content']}"
            # Rough token estimate: 1 token per 4 characters
            snippet_tokens = len(snippet_str) // 4
            if current_tokens + snippet_tokens > self.max_tokens:
                logger.warning(f"Truncating context window {self.window_id} to fit token limit")
                break
            prompt_parts.append(snippet_str)
            current_tokens += snippet_tokens
        return "\n".join(prompt_parts)

class Codex2ContextBuilder:
    """Builds optimized context windows for Codex 2 inference with caching"""

    def __init__(
        self,
        codex_api_key: str,
        redis_url: str = "redis://localhost:6379/0",
        cache_ttl: int = 3600
    ):
        self.client = Codex2Client(api_key=codex_api_key)
        try:
            self.redis_client = redis.from_url(redis_url, decode_responses=True)
            self.redis_client.ping()
            self.cache_enabled = True
        except Exception as e:
            logger.error(f"Failed to connect to Redis: {str(e)}")
            self.redis_client = None
            self.cache_enabled = False
        self.cache_ttl = cache_ttl

    def _generate_window_id(self, file_path: str, snippet_hashes: List[str]) -> str:
        """Generate a unique ID for a context window based on content"""
        hash_input = f"{file_path}:{','.join(snippet_hashes)}"
        return hashlib.sha256(hash_input.encode()).hexdigest()

    def _get_cached_window(self, window_id: str) -> Optional[ContextWindow]:
        """Retrieve cached context window from Redis"""
        if not self.cache_enabled:
            return None
        try:
            cached = self.redis_client.get(f"ctx_window:{window_id}")
            if cached:
                window_dict = json.loads(cached)
                return ContextWindow(**window_dict)
        except Exception as e:
            logger.error(f"Failed to retrieve cached window {window_id}: {str(e)}")
        return None

    def _cache_window(self, window: ContextWindow):
        """Cache context window in Redis with TTL"""
        if not self.cache_enabled:
            return
        try:
            self.redis_client.setex(
                f"ctx_window:{window.window_id}",
                self.cache_ttl,
                json.dumps(asdict(window))
            )
        except Exception as e:
            logger.error(f"Failed to cache window {window.window_id}: {str(e)}")

    @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
    def build_context_window(
        self,
        file_path: str,
        language: str,
        raw_snippets: List[Dict],
        max_tokens: int = 4096
    ) -> ContextWindow:
        """Build and cache a context window, with retry logic for transient errors"""
        # Generate hashes for snippets to check cache
        snippet_hashes = [hashlib.sha256(json.dumps(s, sort_keys=True).encode()).hexdigest() for s in raw_snippets]
        window_id = self._generate_window_id(file_path, snippet_hashes)

        # Check cache first
        cached = self._get_cached_window(window_id)
        if cached:
            logger.info(f"Cache hit for window {window_id}")
            return cached

        # Sort snippets by relevance score descending
        sorted_snippets = sorted(raw_snippets, key=lambda x: x.get("relevance_score", 0.0), reverse=True)

        # Create context window
        window = ContextWindow(
            window_id=window_id,
            file_path=file_path,
            language=language,
            context_snippets=sorted_snippets,
            max_tokens=max_tokens
        )

        # Cache the window
        self._cache_window(window)
        logger.info(f"Built and cached context window {window_id} for {file_path}")
        return window

    def generate_completion(self, window: ContextWindow, prompt: str) -> str:
        """Generate a completion using Codex 2 with the built context window"""
        full_prompt = f"{window.to_prompt_string()}\n\n# Current prompt: {prompt}"
        try:
            response = self.client.generate_completion(
                prompt=full_prompt,
                context=[s["content"] for s in window.context_snippets],
                max_tokens=256
            )
            return response["choices"][0]["text"]
        except Exception as e:
            logger.error(f"Failed to generate completion: {str(e)}")
            raise

if __name__ == "__main__":
    # Example usage
    builder = Codex2ContextBuilder(codex_api_key="sk-mock-key")
    sample_snippets = [
        {"node_type": "class_definition", "start_line": 3, "end_line": 10, "content": "class ContextEngine:\n    pass", "relevance_score": 0.9},
        {"node_type": "function_definition", "start_line": 12, "end_line": 15, "content": "def get_context():\n    return []", "relevance_score": 0.8}
    ]
    window = builder.build_context_window("sample.py", "python", sample_snippets)
    print(f"Window ID: {window.window_id}")
    print(f"Prompt snippet: {window.to_prompt_string()[:200]}...")

import torch
import numpy as np
import logging
from typing import List, Dict, Optional
from transformers import AutoTokenizer, AutoModelForSequenceClassification
from tenacity import retry, stop_after_attempt, wait_fixed

# Configure logging
logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
logger = logging.getLogger(__name__)

# Reference to relevance scorer model: https://github.com/github/copilot-context-relevance-bert
MODEL_NAME = "github/copilot-context-relevance-bert-base-2026"
DEVICE = "cuda" if torch.cuda.is_available() else "cpu"

class ContextRelevanceScorer:
    """Scores context snippets for relevance to the current editing context using a fine-tuned BERT model"""

    def __init__(self, model_name: str = MODEL_NAME, device: str = DEVICE):
        self.model_name = model_name
        self.device = device
        self.tokenizer = None
        self.model = None
        self._load_model()

    def _load_model(self):
        """Load the relevance scoring model with error handling"""
        try:
            logger.info(f"Loading relevance model: {self.model_name}")
            self.tokenizer = AutoTokenizer.from_pretrained(self.model_name)
            self.model = AutoModelForSequenceClassification.from_pretrained(self.model_name)
            self.model.to(self.device)
            self.model.eval()
            logger.info(f"Model loaded successfully to {self.device}")
        except Exception as e:
            logger.error(f"Failed to load model {self.model_name}: {str(e)}")
            raise

    @retry(stop=stop_after_attempt(3), wait=wait_fixed(2))
    def score_snippets(
        self,
        current_cursor_context: str,
        snippets: List[Dict],
        batch_size: int = 32
    ) -> List[Dict]:
        """Score a list of context snippets for relevance to the current cursor context"""
        if not snippets:
            return []

        # Prepare inputs: concatenate cursor context with each snippet
        input_texts = []
        for snippet in snippets:
            # Format input as [CLS] cursor_context [SEP] snippet_content [SEP]
            input_text = f"{current_cursor_context} [SEP] {snippet['content']}"
            input_texts.append(input_text)

        scores = []
        # Process in batches to avoid OOM
        for i in range(0, len(input_texts), batch_size):
            batch_texts = input_texts[i:i+batch_size]
            try:
                inputs = self.tokenizer(
                    batch_texts,
                    padding=True,
                    truncation=True,
                    max_length=512,
                    return_tensors="pt"
                ).to(self.device)

                with torch.no_grad():
                    outputs = self.model(**inputs)
                    logits = outputs.logits
                    # Apply sigmoid to get relevance score between 0 and 1
                    batch_scores = torch.sigmoid(logits[:, 1]).cpu().numpy()
                    scores.extend(batch_scores.tolist())
            except Exception as e:
                logger.error(f"Failed to score batch {i//batch_size}: {str(e)}")
                # Retry via tenacity decorator
                raise

        # Attach scores to snippets
        scored_snippets = []
        for snippet, score in zip(snippets, scores):
            scored = snippet.copy()
            scored["relevance_score"] = float(score)
            scored_snippets.append(scored)

        logger.info(f"Scored {len(scored_snippets)} snippets, average score: {np.mean(scores):.4f}")
        return scored_snippets

    def filter_irrelevant(self, snippets: List[Dict], threshold: float = 0.3) -> List[Dict]:
        """Filter out snippets with relevance score below threshold"""
        filtered = [s for s in snippets if s.get("relevance_score", 0.0) >= threshold]
        logger.info(f"Filtered {len(snippets) - len(filtered)} irrelevant snippets (threshold: {threshold})")
        return filtered

if __name__ == "__main__":
    # Example usage
    try:
        scorer = ContextRelevanceScorer()
        current_context = "def calculate_latency(context_window):"
        sample_snippets = [
            {"content": "class ContextEngine:\n    def get_latency(self):\n        return 0", "node_type": "class_definition"},
            {"content": "import os\nimport sys", "node_type": "import_statement"},
            {"content": "def process_data():\n    return []", "node_type": "function_definition"}
        ]
        scored = scorer.score_snippets(current_context, sample_snippets)
        filtered = scorer.filter_irrelevant(scored)
        for s in filtered:
            print(f"Snippet type: {s['node_type']}, Relevance: {s['relevance_score']:.4f}")
    except Exception as e:
        logger.error(f"Example usage failed: {str(e)}")

# Dockerfile snippet for preloading grammars
FROM tree-sitter/cli:0.24.5 AS grammar-builder
WORKDIR /grammars
# Clone core Tree-sitter and 42 supported language grammars
RUN git clone https://github.com/tree-sitter/tree-sitter.git
RUN git clone https://github.com/tree-sitter/tree-sitter-python.git
RUN git clone https://github.com/tree-sitter/tree-sitter-javascript.git
# ... repeat for all 42 languages
RUN for dir in /grammars/tree-sitter-*/; do cd $dir && tree-sitter build; done
RUN cp /grammars/tree-sitter-*/tree-sitter-*.so /output/

FROM python:3.12-slim
COPY --from=grammar-builder /output/ /usr/local/lib/tree-sitter-grammars/
# ... rest of your app deployment

# Short snippet for content-addressed cache key generation
import hashlib
import json

def generate_cache_key(file_path: str, snippets: list[dict]) -> str:
    # Sort snippets to ensure consistent hashing regardless of order
    sorted_snippets = sorted(snippets, key=lambda x: (x["start_line"], x["end_line"]))
    # Hash file path + sorted snippet content
    hash_input = f"{file_path}:{json.dumps(sorted_snippets, sort_keys=True)}"
    return f"ctx:{hashlib.sha256(hash_input.encode()).hexdigest()}"

# Prepare fine-tuning data for Codex 2 context relevance
import json

def prepare_fine_tuning_example(cursor_context: str, snippet: dict, label: int) -> dict:
    return {
        "prompt": f"Is this snippet relevant to the current context?\nContext: {cursor_context}\nSnippet: {snippet['content']}",
        "completion": "Yes" if label == 1 else "No"
    }

# Example usage with labeled data
labeled_data = [
    {"cursor": "def process_payment(amount):", "snippet": {"content": "def validate_amount(amount): return amount > 0"}, "label": 1},
    {"cursor": "def process_payment(amount):", "snippet": {"content": "import os"}, "label": 0}
]
fine_tuning_data = [prepare_fine_tuning_example(d["cursor"], d["snippet"], d["label"]) for d in labeled_data]
with open("codex2_fine_tuning.jsonl", "w") as f:
    for example in fine_tuning_data:
        f.write(json.dumps(example) + "\n")