Domain Knowledge Collection — NineS

# Domain Knowledge Collection — NineS

> **Task**: T04 (Research Team L3) | **Generated**: 2026-04-11 | **Status**: Complete

This document collects domain knowledge across four key technical areas required to build NineS: information retrieval APIs, knowledge decomposition patterns, self-improving system design, and lightweight sandbox solutions. Each section provides concrete technical approaches with runnable code examples.

---

## Table of Contents

1. [Area 1: Information Retrieval APIs](#area-1-information-retrieval-apis)
2. [Area 2: Knowledge Decomposition and Abstraction Patterns](#area-2-knowledge-decomposition-and-abstraction-patterns)
3. [Area 3: Self-Improving System Design Patterns](#area-3-self-improving-system-design-patterns)
4. [Area 4: Lightweight Sandbox Solutions](#area-4-lightweight-sandbox-solutions)

---

## Area 1: Information Retrieval APIs

NineS's information collection subsystem needs to pull data from GitHub (repositories, activity, releases), arXiv (papers), and general web sources (blogs, announcements via RSS). This section covers concrete API usage, authentication, and rate-limiting strategies.

### 1.1 GitHub REST API v3

The REST API is best suited for simple, targeted lookups and paginated list operations.

**Base URL**: `https://api.github.com`
**API Version Header**: `X-GitHub-Api-Version: 2022-11-28` (or `2026-03-10` for latest)

#### Authentication

```python
import httpx

GITHUB_TOKEN = "ghp_..."  # Personal access token or GitHub App token

client = httpx.Client(
    base_url="https://api.github.com",
    headers={
        "Accept": "application/vnd.github+json",
        "Authorization": f"Bearer {GITHUB_TOKEN}",
        "X-GitHub-Api-Version": "2022-11-28",
    },
    timeout=30.0,
)
```

#### Repository Search

Search repositories by topic, language, and minimum star count:

```python
def search_repositories(
    client: httpx.Client,
    query: str,
    sort: str = "stars",
    order: str = "desc",
    per_page: int = 30,
    page: int = 1,
) -> dict:
    """Search GitHub repositories. Returns up to 1000 results total."""
    resp = client.get(
        "/search/repositories",
        params={
            "q": query,
            "sort": sort,
            "order": order,
            "per_page": per_page,
            "page": page,
        },
    )
    resp.raise_for_status()
    return resp.json()

# Search for AI agent evaluation frameworks in Python with 50+ stars
results = search_repositories(
    client,
    query="topic:agent-evaluation language:python stars:>50",
)
for repo in results["items"]:
    print(f"{repo['full_name']}: ★{repo['stargazers_count']} | {repo['description']}")
```

#### Star/Fork/Commit Tracking

Track activity over time for a specific repository:

```python
def get_commit_activity(client: httpx.Client, owner: str, repo: str) -> list[dict]:
    """Weekly commit counts for the last year (52 data points)."""
    resp = client.get(f"/repos/{owner}/{repo}/stats/commit_activity")
    resp.raise_for_status()
    return resp.json()

def get_stargazers_with_dates(
    client: httpx.Client, owner: str, repo: str, per_page: int = 100
) -> list[dict]:
    """Stargazers with timestamps for star-growth tracking."""
    resp = client.get(
        f"/repos/{owner}/{repo}/stargazers",
        headers={"Accept": "application/vnd.github.star+json"},
        params={"per_page": per_page},
    )
    resp.raise_for_status()
    return resp.json()

def get_repo_details(client: httpx.Client, owner: str, repo: str) -> dict:
    """Core metrics: stars, forks, watchers, open issues, language."""
    resp = client.get(f"/repos/{owner}/{repo}")
    resp.raise_for_status()
    data = resp.json()
    return {
        "stars": data["stargazers_count"],
        "forks": data["forks_count"],
        "watchers": data["subscribers_count"],
        "open_issues": data["open_issues_count"],
        "language": data["language"],
        "updated_at": data["updated_at"],
        "pushed_at": data["pushed_at"],
    }
```

#### README Fetching

```python
import base64

def get_readme(client: httpx.Client, owner: str, repo: str) -> str:
    """Fetch decoded README content."""
    resp = client.get(f"/repos/{owner}/{repo}/readme")
    resp.raise_for_status()
    data = resp.json()
    return base64.b64decode(data["content"]).decode("utf-8")
```

#### Release Monitoring

```python
def get_latest_releases(
    client: httpx.Client, owner: str, repo: str, count: int = 5
) -> list[dict]:
    """Fetch recent releases for change tracking."""
    resp = client.get(
        f"/repos/{owner}/{repo}/releases",
        params={"per_page": count},
    )
    resp.raise_for_status()
    return [
        {
            "tag": r["tag_name"],
            "name": r["name"],
            "published_at": r["published_at"],
            "body": r["body"][:500],
        }
        for r in resp.json()
    ]
```

### 1.2 GitHub GraphQL API v4

GraphQL is preferred when NineS needs to fetch multiple related fields in a single request, reducing round-trips and staying within rate limits. GraphQL uses a **point-based** rate limit system (5,000 points/hour) rather than per-request counts.

**Endpoint**: `POST https://api.github.com/graphql`

#### Single Repository Deep Fetch

Retrieve stars, forks, recent commits, and releases in one call:

```python
REPO_DEEP_QUERY = """
query($owner: String!, $name: String!) {
  repository(owner: $owner, name: $name) {
    name
    description
    stargazerCount
    forkCount
    primaryLanguage { name }
    defaultBranchRef {
      target {
        ... on Commit {
          history(first: 10) {
            totalCount
            nodes {
              messageHeadline
              committedDate
              author { name }
            }
          }
        }
      }
    }
    releases(last: 5, orderBy: {field: CREATED_AT, direction: DESC}) {
      nodes { tagName name publishedAt description }
    }
    repositoryTopics(first: 10) {
      nodes { topic { name } }
    }
    issues(states: OPEN) { totalCount }
    pullRequests(states: OPEN) { totalCount }
  }
}
"""

def graphql_query(client: httpx.Client, query: str, variables: dict) -> dict:
    resp = client.post(
        "https://api.github.com/graphql",
        json={"query": query, "variables": variables},
    )
    resp.raise_for_status()
    data = resp.json()
    if "errors" in data:
        raise RuntimeError(f"GraphQL errors: {data['errors']}")
    return data["data"]

result = graphql_query(client, REPO_DEEP_QUERY, {
    "owner": "facebook",
    "name": "react",
})
```

#### Batch Repository Search

Search and compare multiple repositories matching a topic:

```python
SEARCH_QUERY = """
query($searchQuery: String!, $first: Int!) {
  search(query: $searchQuery, type: REPOSITORY, first: $first) {
    repositoryCount
    nodes {
      ... on Repository {
        nameWithOwner
        stargazerCount
        forkCount
        description
        updatedAt
        primaryLanguage { name }
        defaultBranchRef {
          target {
            ... on Commit {
              history(first: 1) {
                nodes { committedDate }
              }
            }
          }
        }
      }
    }
  }
}
"""

results = graphql_query(client, SEARCH_QUERY, {
    "searchQuery": "topic:llm-evaluation language:python stars:>100",
    "first": 20,
})
```

#### Multi-Repo Comparison with Fragments

Compare specific repositories side-by-side:

```python
COMPARE_QUERY = """
fragment RepoMetrics on Repository {
  stargazerCount
  forkCount
  issues(states: OPEN) { totalCount }
  pullRequests(states: OPEN) { totalCount }
  releases(last: 1) { nodes { tagName publishedAt } }
}

query {
  swebench: repository(owner: "princeton-nlp", name: "SWE-bench") { ...RepoMetrics }
  humaneval: repository(owner: "openai", name: "human-eval") { ...RepoMetrics }
  bigcode: repository(owner: "bigcode-project", name: "bigcodebench") { ...RepoMetrics }
}
"""
```

### 1.3 arXiv API

arXiv provides an Atom-based API for searching and retrieving paper metadata. NineS uses this to track AI evaluation research.

**Base URL**: `http://export.arxiv.org/api/query`

#### Direct API Usage

```python
import urllib.parse
import xml.etree.ElementTree as ET
from dataclasses import dataclass
from datetime import datetime

import httpx

ARXIV_NS = {"atom": "http://www.w3.org/2005/Atom"}

@dataclass
class ArxivPaper:
    arxiv_id: str
    title: str
    authors: list[str]
    abstract: str
    categories: list[str]
    published: datetime
    updated: datetime
    pdf_url: str

def search_arxiv(
    query: str,
    start: int = 0,
    max_results: int = 10,
    sort_by: str = "submittedDate",
    sort_order: str = "descending",
) -> list[ArxivPaper]:
    """
    Search arXiv. Query syntax:
      - ti:keyword      (title)
      - au:author        (author)
      - abs:keyword      (abstract)
      - cat:cs.AI        (category)
      - all:keyword      (all fields)
    Combine with AND, OR, ANDNOT.
    """
    params = {
        "search_query": query,
        "start": start,
        "max_results": max_results,
        "sortBy": sort_by,
        "sortOrder": sort_order,
    }
    url = f"http://export.arxiv.org/api/query?{urllib.parse.urlencode(params)}"

    resp = httpx.get(url, timeout=30.0)
    resp.raise_for_status()

    root = ET.fromstring(resp.text)
    papers = []
    for entry in root.findall("atom:entry", ARXIV_NS):
        paper = ArxivPaper(
            arxiv_id=entry.find("atom:id", ARXIV_NS).text.split("/abs/")[-1],
            title=entry.find("atom:title", ARXIV_NS).text.strip(),
            authors=[
                a.find("atom:name", ARXIV_NS).text
                for a in entry.findall("atom:author", ARXIV_NS)
            ],
            abstract=entry.find("atom:summary", ARXIV_NS).text.strip(),
            categories=[
                c.get("term")
                for c in entry.findall("{http://arxiv.org/schemas/atom}category")
            ],
            published=datetime.fromisoformat(
                entry.find("atom:published", ARXIV_NS).text.rstrip("Z")
            ),
            updated=datetime.fromisoformat(
                entry.find("atom:updated", ARXIV_NS).text.rstrip("Z")
            ),
            pdf_url=next(
                (
                    link.get("href")
                    for link in entry.findall("atom:link", ARXIV_NS)
                    if link.get("title") == "pdf"
                ),
                "",
            ),
        )
        papers.append(paper)
    return papers

# Search for recent agent evaluation papers in cs.AI
papers = search_arxiv(
    query="cat:cs.AI AND (ti:agent evaluation OR ti:benchmark)",
    max_results=20,
    sort_by="submittedDate",
)
```

#### Using the `arxiv` Python Library

```python
import arxiv

client = arxiv.Client(
    page_size=50,
    delay_seconds=3.0,  # respect rate limits
    num_retries=3,
)

search = arxiv.Search(
    query='ti:"agent evaluation" AND cat:cs.AI',
    max_results=50,
    sort_by=arxiv.SortCriterion.SubmittedDate,
    sort_order=arxiv.SortOrder.Descending,
)

for result in client.results(search):
    print(f"[{result.entry_id}] {result.title}")
    print(f"  Authors: {', '.join(a.name for a in result.authors)}")
    print(f"  Published: {result.published}")
    print(f"  Categories: {result.categories}")
```

#### Bulk Metadata Collection

For NineS's tracking use case, paginate through results with controlled delays:

```python
import time

def collect_all_papers(
    query: str,
    max_total: int = 500,
    page_size: int = 50,
    delay: float = 3.0,
) -> list[ArxivPaper]:
    """Paginated collection respecting arXiv's rate limits (3s between calls)."""
    all_papers: list[ArxivPaper] = []
    for start in range(0, max_total, page_size):
        batch = search_arxiv(query, start=start, max_results=page_size)
        if not batch:
            break
        all_papers.extend(batch)
        if start + page_size < max_total:
            time.sleep(delay)
    return all_papers
```

### 1.4 RSS Feed Parsing

For tracking blog posts, project announcements, and changelogs:

```python
from dataclasses import dataclass
from datetime import datetime

import feedparser

@dataclass
class FeedEntry:
    title: str
    link: str
    published: str
    summary: str
    source_feed: str

def parse_rss_feed(url: str) -> list[FeedEntry]:
    """Parse an RSS/Atom feed and return structured entries."""
    feed = feedparser.parse(url)
    entries = []
    for entry in feed.entries:
        entries.append(FeedEntry(
            title=entry.get("title", ""),
            link=entry.get("link", ""),
            published=entry.get("published", ""),
            summary=entry.get("summary", "")[:500],
            source_feed=feed.feed.get("title", url),
        ))
    return entries

# Track GitHub release feeds (Atom format)
releases = parse_rss_feed("https://github.com/openai/human-eval/releases.atom")

# Track arXiv new submissions via RSS
arxiv_new = parse_rss_feed("https://rss.arxiv.org/rss/cs.AI")
```

#### Multi-Feed Aggregator

```python
from concurrent.futures import ThreadPoolExecutor, as_completed

TRACKED_FEEDS = [
    "https://github.com/princeton-nlp/SWE-bench/releases.atom",
    "https://github.com/openai/human-eval/releases.atom",
    "https://rss.arxiv.org/rss/cs.AI",
    "https://rss.arxiv.org/rss/cs.SE",
]

def aggregate_feeds(
    feed_urls: list[str],
    max_workers: int = 4,
) -> list[FeedEntry]:
    """Parallel feed fetching with error isolation per feed."""
    all_entries: list[FeedEntry] = []
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = {executor.submit(parse_rss_feed, url): url for url in feed_urls}
        for future in as_completed(futures):
            url = futures[future]
            try:
                entries = future.result()
                all_entries.extend(entries)
            except Exception as exc:
                import logging
                logging.warning("Feed %s failed: %s", url, exc)
    return sorted(all_entries, key=lambda e: e.published, reverse=True)
```

### 1.5 Rate Limiting Strategies

NineS must respect API rate limits across all data sources. The approach uses a token-bucket algorithm with per-source tracking.

```python
import time
import threading
from dataclasses import dataclass, field

@dataclass
class RateLimiter:
    """Token-bucket rate limiter, thread-safe."""
    max_tokens: float
    refill_rate: float  # tokens per second
    _tokens: float = field(init=False)
    _last_refill: float = field(init=False)
    _lock: threading.Lock = field(default_factory=threading.Lock, init=False)

    def __post_init__(self) -> None:
        self._tokens = self.max_tokens
        self._last_refill = time.monotonic()

    def acquire(self, tokens: float = 1.0) -> None:
        """Block until enough tokens are available."""
        while True:
            with self._lock:
                now = time.monotonic()
                elapsed = now - self._last_refill
                self._tokens = min(self.max_tokens, self._tokens + elapsed * self.refill_rate)
                self._last_refill = now
                if self._tokens >= tokens:
                    self._tokens -= tokens
                    return
            time.sleep(0.1)

# Per-source limiters matching documented rate limits
RATE_LIMITERS = {
    "github_rest_search": RateLimiter(max_tokens=30, refill_rate=30 / 60),      # 30/min
    "github_rest_core": RateLimiter(max_tokens=5000, refill_rate=5000 / 3600),   # 5000/hr
    "github_graphql": RateLimiter(max_tokens=5000, refill_rate=5000 / 3600),     # 5000 pts/hr
    "arxiv": RateLimiter(max_tokens=1, refill_rate=1 / 3),                       # 1 per 3s
}
```

#### Adaptive Back-Off Using Response Headers

```python
def github_request_with_backoff(
    client: httpx.Client,
    method: str,
    url: str,
    **kwargs,
) -> httpx.Response:
    """Issue a GitHub API request with rate-limit-aware retry."""
    max_retries = 3
    for attempt in range(max_retries):
        resp = client.request(method, url, **kwargs)

        remaining = int(resp.headers.get("x-ratelimit-remaining", "1"))
        if remaining == 0:
            reset_at = int(resp.headers.get("x-ratelimit-reset", "0"))
            wait = max(0, reset_at - time.time()) + 1
            time.sleep(wait)

        if resp.status_code == 403 and "rate limit" in resp.text.lower():
            reset_at = int(resp.headers.get("x-ratelimit-reset", "0"))
            wait = max(0, reset_at - time.time()) + 1
            time.sleep(wait)
            continue

        if resp.status_code == 429:
            retry_after = int(resp.headers.get("retry-after", "60"))
            time.sleep(retry_after)
            continue

        resp.raise_for_status()
        return resp

    raise RuntimeError(f"Failed after {max_retries} retries: {url}")
```

### 1.6 Authentication Patterns Summary

| Source | Auth Method | Rate Limits (Authenticated) |
|--------|------------|----------------------------|
| GitHub REST | Bearer token (`ghp_...`) or GitHub App JWT | 5,000 req/hr (core), 30 req/min (search) |
| GitHub GraphQL | Bearer token | 5,000 points/hr |
| arXiv | None required | ~1 req/3s (undocumented but enforced) |
| RSS feeds | None required | Varies; respect `Cache-Control` headers |

**Recommended approach for NineS**: Use a GitHub Personal Access Token (fine-grained) for MVP, with scopes limited to `public_repo` read access. Store the token in environment variable `NINES_GITHUB_TOKEN` and load it via the config system.

---

## Area 2: Knowledge Decomposition and Abstraction Patterns

NineS's knowledge analysis engine needs to parse source code, analyze structure, identify architectural patterns, and decompose codebases into reusable knowledge units. This section covers the Python-native tools and algorithms for each step.

### 2.1 AST Analysis in Python

The built-in `ast` module provides full access to Python's abstract syntax tree.

#### Function and Class Extraction

```python
import ast
from dataclasses import dataclass, field
from pathlib import Path

@dataclass
class FunctionInfo:
    name: str
    lineno: int
    end_lineno: int
    args: list[str]
    decorators: list[str]
    docstring: str | None
    is_async: bool
    complexity: int = 0

@dataclass
class ClassInfo:
    name: str
    lineno: int
    bases: list[str]
    methods: list[FunctionInfo]
    docstring: str | None

class CodeExtractor(ast.NodeVisitor):
    """Extract functions, classes, and their metadata from Python source."""

    def __init__(self) -> None:
        self.functions: list[FunctionInfo] = []
        self.classes: list[ClassInfo] = []
        self.imports: list[str] = []
        self._current_class: ClassInfo | None = None

    def visit_FunctionDef(self, node: ast.FunctionDef) -> None:
        info = FunctionInfo(
            name=node.name,
            lineno=node.lineno,
            end_lineno=node.end_lineno or node.lineno,
            args=[arg.arg for arg in node.args.args],
            decorators=[ast.dump(d) for d in node.decorator_list],
            docstring=ast.get_docstring(node),
            is_async=False,
            complexity=self._cyclomatic_complexity(node),
        )
        if self._current_class is not None:
            self._current_class.methods.append(info)
        else:
            self.functions.append(info)
        self.generic_visit(node)

    visit_AsyncFunctionDef = visit_FunctionDef  # same extraction logic

    def visit_ClassDef(self, node: ast.ClassDef) -> None:
        cls = ClassInfo(
            name=node.name,
            lineno=node.lineno,
            bases=[ast.dump(b) for b in node.bases],
            methods=[],
            docstring=ast.get_docstring(node),
        )
        prev = self._current_class
        self._current_class = cls
        self.generic_visit(node)
        self._current_class = prev
        self.classes.append(cls)

    def visit_Import(self, node: ast.Import) -> None:
        for alias in node.names:
            self.imports.append(alias.name)

    def visit_ImportFrom(self, node: ast.ImportFrom) -> None:
        module = node.module or ""
        for alias in node.names:
            self.imports.append(f"{module}.{alias.name}")

    @staticmethod
    def _cyclomatic_complexity(node: ast.AST) -> int:
        """McCabe cyclomatic complexity: count decision points + 1."""
        complexity = 1
        for child in ast.walk(node):
            if isinstance(child, (ast.If, ast.While, ast.For, ast.ExceptHandler)):
                complexity += 1
            elif isinstance(child, ast.BoolOp):
                complexity += len(child.values) - 1
            elif isinstance(child, (ast.Assert, ast.With)):
                complexity += 1
        return complexity

def analyze_file(path: Path) -> CodeExtractor:
    source = path.read_text(encoding="utf-8")
    tree = ast.parse(source, filename=str(path))
    extractor = CodeExtractor()
    extractor.visit(tree)
    return extractor
```

#### Dependency Graph Construction

Build an intra-project import dependency graph:

```python
from pathlib import Path

def build_dependency_graph(project_root: Path) -> dict[str, set[str]]:
    """
    Map each module to the set of project-internal modules it imports.
    Returns adjacency list: module_path -> {imported_module_paths, ...}
    """
    py_files = list(project_root.rglob("*.py"))
    module_map: dict[str, Path] = {}
    for f in py_files:
        rel = f.relative_to(project_root)
        mod_name = str(rel).replace("/", ".").removesuffix(".py").removesuffix(".__init__")
        module_map[mod_name] = f

    graph: dict[str, set[str]] = {mod: set() for mod in module_map}

    for mod_name, filepath in module_map.items():
        try:
            tree = ast.parse(filepath.read_text(encoding="utf-8"))
        except SyntaxError:
            continue
        for node in ast.walk(tree):
            imported: str | None = None
            if isinstance(node, ast.Import):
                for alias in node.names:
                    imported = alias.name
            elif isinstance(node, ast.ImportFrom) and node.module:
                imported = node.module

            if imported is not None:
                for candidate in module_map:
                    if imported == candidate or imported.startswith(candidate + "."):
                        graph[mod_name].add(candidate)
                        break

    return graph
```

#### Coupling Metrics

```python
def compute_coupling_metrics(graph: dict[str, set[str]]) -> dict[str, dict[str, int]]:
    """Afferent (Ca) and efferent (Ce) coupling per module."""
    metrics: dict[str, dict[str, int]] = {}
    for mod in graph:
        ce = len(graph[mod])  # outgoing dependencies
        ca = sum(1 for other in graph if mod in graph[other])  # incoming dependents
        instability = ce / (ca + ce) if (ca + ce) > 0 else 0.0
        metrics[mod] = {"Ca": ca, "Ce": ce, "instability": round(instability, 3)}
    return metrics
```

### 2.2 Directory Structure Analysis

Analyze project layout to detect module boundaries and layering:

```python
from pathlib import Path
from dataclasses import dataclass

@dataclass
class DirectoryNode:
    name: str
    path: Path
    is_package: bool
    children: list["DirectoryNode"]
    py_file_count: int
    total_lines: int

def analyze_directory_structure(root: Path, max_depth: int = 6) -> DirectoryNode:
    """Recursively analyze directory structure and code distribution."""
    def _walk(path: Path, depth: int) -> DirectoryNode:
        children: list[DirectoryNode] = []
        py_count = 0
        total_lines = 0

        if depth < max_depth:
            for child in sorted(path.iterdir()):
                if child.name.startswith((".", "__pycache__", "node_modules")):
                    continue
                if child.is_dir():
                    children.append(_walk(child, depth + 1))
                elif child.suffix == ".py":
                    py_count += 1
                    try:
                        total_lines += len(child.read_text(encoding="utf-8").splitlines())
                    except (OSError, UnicodeDecodeError):
                        pass

        child_py = sum(c.py_file_count for c in children)
        child_lines = sum(c.total_lines for c in children)

        return DirectoryNode(
            name=path.name,
            path=path,
            is_package=(path / "__init__.py").exists(),
            children=children,
            py_file_count=py_count + child_py,
            total_lines=total_lines + child_lines,
        )

    return _walk(root, 0)
```

#### Layer Detection Heuristics

```python
LAYER_INDICATORS = {
    "presentation": {"cli", "api", "web", "ui", "views", "routes", "endpoints", "handlers"},
    "application":  {"services", "usecases", "commands", "orchestrator", "workflows"},
    "domain":       {"models", "entities", "domain", "core", "types"},
    "infrastructure": {"db", "database", "repos", "adapters", "clients", "storage", "external"},
    "testing":      {"tests", "test", "fixtures", "conftest", "mocks"},
}

def detect_layers(root: Path) -> dict[str, list[Path]]:
    """Classify top-level directories into architectural layers."""
    detected: dict[str, list[Path]] = {layer: [] for layer in LAYER_INDICATORS}
    for child in root.iterdir():
        if not child.is_dir() or child.name.startswith("."):
            continue
        name_lower = child.name.lower()
        for layer, indicators in LAYER_INDICATORS.items():
            if name_lower in indicators:
                detected[layer].append(child)
    return detected
```

### 2.3 Architecture Pattern Recognition

Detect common architectural patterns from code structure:

```python
@dataclass
class ArchitectureSignal:
    pattern: str
    confidence: float  # 0.0 - 1.0
    evidence: list[str]

def detect_architecture_patterns(
    root: Path,
    dep_graph: dict[str, set[str]],
) -> list[ArchitectureSignal]:
    signals: list[ArchitectureSignal] = []
    dirs = {d.name.lower() for d in root.iterdir() if d.is_dir()}

    # MVC detection
    mvc_dirs = {"models", "views", "controllers"}
    mvc_overlap = mvc_dirs & dirs
    if len(mvc_overlap) >= 2:
        signals.append(ArchitectureSignal(
            pattern="MVC",
            confidence=len(mvc_overlap) / 3,
            evidence=[f"Found directories: {mvc_overlap}"],
        ))

    # Hexagonal / Ports & Adapters detection
    hex_indicators = {"ports", "adapters", "domain", "core"}
    hex_overlap = hex_indicators & dirs
    if len(hex_overlap) >= 2:
        signals.append(ArchitectureSignal(
            pattern="Hexagonal",
            confidence=len(hex_overlap) / 4,
            evidence=[f"Found directories: {hex_overlap}"],
        ))

    # Layered architecture detection
    layer_dirs = {"presentation", "application", "domain", "infrastructure"}
    layer_overlap = layer_dirs & dirs
    if len(layer_overlap) >= 2:
        signals.append(ArchitectureSignal(
            pattern="Layered",
            confidence=len(layer_overlap) / 4,
            evidence=[f"Found directories: {layer_overlap}"],
        ))

    # Microservices indicators
    service_dirs = [d for d in dirs if "service" in d or "svc" in d]
    docker_compose = (root / "docker-compose.yml").exists() or (root / "docker-compose.yaml").exists()
    if len(service_dirs) >= 2 or (docker_compose and len(service_dirs) >= 1):
        signals.append(ArchitectureSignal(
            pattern="Microservices",
            confidence=min(1.0, len(service_dirs) * 0.3 + (0.3 if docker_compose else 0)),
            evidence=[f"Service dirs: {service_dirs}", f"docker-compose: {docker_compose}"],
        ))

    # Plugin / Extension detection (Protocol-based)
    protocol_count = 0
    for mod in dep_graph:
        path = root / mod.replace(".", "/")
        for suffix in [".py", "/__init__.py"]:
            fpath = Path(str(path) + suffix)
            if fpath.exists():
                try:
                    source = fpath.read_text(encoding="utf-8")
                    tree = ast.parse(source)
                    for node in ast.walk(tree):
                        if isinstance(node, ast.ClassDef):
                            for base in node.bases:
                                if isinstance(base, ast.Name) and base.id == "Protocol":
                                    protocol_count += 1
                except (SyntaxError, OSError):
                    pass

    if protocol_count >= 3:
        signals.append(ArchitectureSignal(
            pattern="Plugin/Extension",
            confidence=min(1.0, protocol_count * 0.15),
            evidence=[f"Found {protocol_count} Protocol-based interfaces"],
        ))

    return signals
```

### 2.4 Abstraction Strategies

Three decomposition strategies that NineS can apply:

#### Functional Decomposition

Break code into units by what each function/class *does*:

```python
@dataclass
class KnowledgeUnit:
    id: str
    name: str
    category: str  # "function", "class", "module", "pattern"
    abstraction_level: str  # "concrete", "interface", "concept"
    source_path: str
    line_range: tuple[int, int]
    dependencies: list[str]
    description: str
    tags: list[str]

def functional_decomposition(extractor: CodeExtractor, filepath: str) -> list[KnowledgeUnit]:
    """Decompose a file into knowledge units by function/class."""
    units: list[KnowledgeUnit] = []

    for func in extractor.functions:
        units.append(KnowledgeUnit(
            id=f"{filepath}::{func.name}",
            name=func.name,
            category="function",
            abstraction_level="concrete",
            source_path=filepath,
            line_range=(func.lineno, func.end_lineno),
            dependencies=[],  # populated from import analysis
            description=func.docstring or "",
            tags=_infer_tags(func.name, func.docstring),
        ))

    for cls in extractor.classes:
        units.append(KnowledgeUnit(
            id=f"{filepath}::{cls.name}",
            name=cls.name,
            category="class",
            abstraction_level="interface" if _is_abstract(cls) else "concrete",
            source_path=filepath,
            line_range=(cls.lineno, cls.lineno),
            dependencies=cls.bases,
            description=cls.docstring or "",
            tags=_infer_tags(cls.name, cls.docstring),
        ))

    return units

def _infer_tags(name: str, docstring: str | None) -> list[str]:
    """Heuristic tag inference from naming conventions."""
    tags: list[str] = []
    name_lower = name.lower()
    if name_lower.startswith("test"):
        tags.append("test")
    if "parse" in name_lower or "extract" in name_lower:
        tags.append("parsing")
    if "score" in name_lower or "eval" in name_lower:
        tags.append("evaluation")
    if name_lower.startswith("_"):
        tags.append("private")
    return tags

def _is_abstract(cls: ClassInfo) -> bool:
    return any("ABC" in b or "Protocol" in b for b in cls.bases)
```

#### Concern-Based Decomposition

Group code units by cross-cutting concern:

```python
CONCERN_PATTERNS = {
    "error_handling": ["except", "raise", "Error", "Exception"],
    "logging":        ["logger", "logging", "log."],
    "validation":     ["validate", "assert", "check", "verify"],
    "serialization":  ["to_dict", "from_dict", "serialize", "deserialize", "json", "toml"],
    "configuration":  ["config", "settings", "options", "defaults"],
    "io":             ["read", "write", "open", "save", "load", "fetch"],
}

def concern_decomposition(
    units: list[KnowledgeUnit],
    source_map: dict[str, str],
) -> dict[str, list[KnowledgeUnit]]:
    """Group knowledge units by their primary concern."""
    grouped: dict[str, list[KnowledgeUnit]] = {c: [] for c in CONCERN_PATTERNS}
    grouped["core_logic"] = []

    for unit in units:
        source = source_map.get(unit.source_path, "")
        start, end = unit.line_range
        snippet = "\n".join(source.splitlines()[start - 1 : end])

        matched = False
        for concern, patterns in CONCERN_PATTERNS.items():
            if any(p in snippet for p in patterns):
                grouped[concern].append(unit)
                matched = True
                break

        if not matched:
            grouped["core_logic"].append(unit)

    return grouped
```

#### Layer-Based Decomposition

Assign units to architectural layers based on their position and dependencies:

```python
def layer_decomposition(
    units: list[KnowledgeUnit],
    dep_graph: dict[str, set[str]],
) -> dict[str, list[KnowledgeUnit]]:
    """Assign units to layers using dependency direction analysis."""
    layer_assignment: dict[str, list[KnowledgeUnit]] = {
        "interface": [],    # entry points, protocols
        "application": [],  # orchestration, use cases
        "domain": [],       # core business logic
        "infrastructure": [],  # I/O, external services
    }

    for unit in units:
        if unit.abstraction_level == "interface":
            layer_assignment["interface"].append(unit)
        elif "io" in unit.tags or "fetch" in unit.name.lower():
            layer_assignment["infrastructure"].append(unit)
        elif any(kw in unit.name.lower() for kw in ("run", "execute", "pipeline", "orchestrat")):
            layer_assignment["application"].append(unit)
        else:
            layer_assignment["domain"].append(unit)

    return layer_assignment
```

---

## Area 3: Self-Improving System Design Patterns

NineS aims to be a self-iterating tool that measures its own performance, identifies gaps, and plans improvements. This section covers the feedback loop architecture, convergence detection methods, and version-over-version comparison.

### 3.1 Feedback Loop Architecture

The core MAPIM (Measure → Analyze → Plan → Improve → Measure) loop:

```python
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum

class IterationPhase(Enum):
    MEASURE = "measure"
    ANALYZE = "analyze"
    PLAN = "plan"
    IMPROVE = "improve"

@dataclass
class MeasurementSnapshot:
    version: str
    timestamp: datetime
    metrics: dict[str, float]
    dimensions: dict[str, dict[str, float]]

@dataclass
class GapAnalysis:
    dimension: str
    current_score: float
    target_score: float
    gap: float
    severity: str  # "critical", "major", "minor", "acceptable"
    root_causes: list[str]

@dataclass
class ImprovementPlan:
    version: str
    gaps: list[GapAnalysis]
    actions: list["ImprovementAction"]
    priority_order: list[str]
    estimated_impact: dict[str, float]

@dataclass
class ImprovementAction:
    id: str
    target_dimension: str
    description: str
    priority: int
    estimated_effort: str  # "small", "medium", "large"
    expected_metric_delta: dict[str, float]

class SelfImprovementLoop:
    """Orchestrates the MAPIM feedback loop."""

    def __init__(
        self,
        measurer: "Measurer",
        analyzer: "GapAnalyzer",
        planner: "ImprovementPlanner",
        history: list[MeasurementSnapshot] = None,
    ) -> None:
        self.measurer = measurer
        self.analyzer = analyzer
        self.planner = planner
        self.history: list[MeasurementSnapshot] = history or []

    def run_iteration(self, version: str) -> ImprovementPlan:
        snapshot = self.measurer.measure(version)
        self.history.append(snapshot)

        baseline = self.history[0] if len(self.history) > 1 else None
        gaps = self.analyzer.analyze(snapshot, baseline)

        plan = self.planner.plan(gaps, self.history)
        return plan

    def has_converged(self, window: int = 5, threshold: float = 0.01) -> bool:
        """Check if recent iterations show diminishing returns."""
        if len(self.history) < window:
            return False
        recent = self.history[-window:]
        deltas = []
        for i in range(1, len(recent)):
            prev_avg = sum(recent[i - 1].metrics.values()) / len(recent[i - 1].metrics)
            curr_avg = sum(recent[i].metrics.values()) / len(recent[i].metrics)
            deltas.append(abs(curr_avg - prev_avg))
        return max(deltas) < threshold


class Measurer(ABC):
    @abstractmethod
    def measure(self, version: str) -> MeasurementSnapshot: ...

class GapAnalyzer(ABC):
    @abstractmethod
    def analyze(
        self, current: MeasurementSnapshot, baseline: MeasurementSnapshot | None
    ) -> list[GapAnalysis]: ...

class ImprovementPlanner(ABC):
    @abstractmethod
    def plan(
        self, gaps: list[GapAnalysis], history: list[MeasurementSnapshot]
    ) -> ImprovementPlan: ...
```

### 3.2 Meta-Learning Concepts for Tool Improvement

Meta-learning ("learning to learn") applies to NineS as a system that can improve its own evaluation and analysis strategies.

**Key applicable concepts:**

```python
@dataclass
class StrategyPerformance:
    strategy_id: str
    task_type: str
    success_rate: float
    avg_score: float
    sample_count: int

class StrategySelector:
    """
    Select the best analysis/evaluation strategy based on accumulated performance data.
    Implements a simplified multi-armed bandit (epsilon-greedy) approach.
    """

    def __init__(self, epsilon: float = 0.1) -> None:
        self.epsilon = epsilon
        self.performance: dict[str, list[StrategyPerformance]] = {}

    def record(self, perf: StrategyPerformance) -> None:
        self.performance.setdefault(perf.strategy_id, []).append(perf)

    def select(self, task_type: str, available_strategies: list[str]) -> str:
        """Epsilon-greedy selection: exploit best known strategy, explore with probability epsilon."""
        import random

        if random.random() < self.epsilon:
            return random.choice(available_strategies)

        best_strategy = available_strategies[0]
        best_score = -1.0
        for strategy in available_strategies:
            records = [
                p for p in self.performance.get(strategy, [])
                if p.task_type == task_type
            ]
            if records:
                avg = sum(r.avg_score for r in records) / len(records)
                if avg > best_score:
                    best_score = avg
                    best_strategy = strategy

        return best_strategy

    def get_improvement_rate(self, strategy_id: str) -> float | None:
        """Compute the learning rate: how much a strategy improves over time."""
        records = self.performance.get(strategy_id, [])
        if len(records) < 2:
            return None
        first_half = records[: len(records) // 2]
        second_half = records[len(records) // 2 :]
        avg_first = sum(r.avg_score for r in first_half) / len(first_half)
        avg_second = sum(r.avg_score for r in second_half) / len(second_half)
        return (avg_second - avg_first) / max(avg_first, 1e-6)
```

### 3.3 Auto-Curriculum: Progressively Harder Evaluation Tasks

NineS should generate evaluation tasks of increasing difficulty to push its own capabilities:

```python
@dataclass
class DifficultyLevel:
    level: int
    name: str
    criteria: dict[str, float]

DIFFICULTY_SCALE = [
    DifficultyLevel(1, "trivial", {"max_files": 1, "max_complexity": 5, "max_dependencies": 0}),
    DifficultyLevel(2, "simple", {"max_files": 3, "max_complexity": 10, "max_dependencies": 2}),
    DifficultyLevel(3, "moderate", {"max_files": 10, "max_complexity": 20, "max_dependencies": 5}),
    DifficultyLevel(4, "complex", {"max_files": 30, "max_complexity": 50, "max_dependencies": 15}),
    DifficultyLevel(5, "expert", {"max_files": 100, "max_complexity": 100, "max_dependencies": 50}),
]

class AutoCurriculumGenerator:
    """Generate evaluation tasks at the appropriate difficulty level."""

    def __init__(self, mastery_threshold: float = 0.85) -> None:
        self.mastery_threshold = mastery_threshold
        self.level_scores: dict[int, list[float]] = {}

    def record_score(self, level: int, score: float) -> None:
        self.level_scores.setdefault(level, []).append(score)

    def current_level(self) -> int:
        """Find the highest level where mastery has been demonstrated."""
        for level_def in reversed(DIFFICULTY_SCALE):
            scores = self.level_scores.get(level_def.level, [])
            if len(scores) >= 3:
                recent_avg = sum(scores[-3:]) / 3
                if recent_avg >= self.mastery_threshold:
                    next_level = min(level_def.level + 1, DIFFICULTY_SCALE[-1].level)
                    return next_level
        return 1

    def should_advance(self) -> bool:
        """Check if performance at current level justifies advancement."""
        level = self.current_level()
        scores = self.level_scores.get(level, [])
        if len(scores) < 3:
            return False
        return sum(scores[-3:]) / 3 >= self.mastery_threshold
```

### 3.4 Convergence Detection

Statistical methods to determine when improvement has plateaued. This is critical for NineS to know when a particular self-improvement cycle should terminate.

#### Method 1: Sliding Window Variance

```python
import math

def sliding_window_convergence(
    scores: list[float],
    window_size: int = 5,
    variance_threshold: float = 0.001,
) -> bool:
    """
    Converged = the variance of the last `window_size` scores is below threshold.
    Works well for metrics that stabilize around a value.
    """
    if len(scores) < window_size:
        return False
    window = scores[-window_size:]
    mean = sum(window) / len(window)
    variance = sum((x - mean) ** 2 for x in window) / len(window)
    return variance < variance_threshold
```

#### Method 2: Relative Improvement Rate

```python
def relative_improvement_convergence(
    scores: list[float],
    window_size: int = 3,
    min_improvement: float = 0.005,
) -> bool:
    """
    Converged = average relative improvement over last `window_size` steps
    is below `min_improvement` (0.5% default).
    """
    if len(scores) < window_size + 1:
        return False
    improvements = []
    for i in range(-window_size, 0):
        prev = scores[i - 1]
        curr = scores[i]
        if prev != 0:
            improvements.append((curr - prev) / abs(prev))
        else:
            improvements.append(0.0)
    avg_improvement = sum(improvements) / len(improvements)
    return avg_improvement < min_improvement
```

#### Method 3: Mann-Kendall Trend Test

A non-parametric statistical test to determine if there's a monotonic trend:

```python
def mann_kendall_trend(scores: list[float]) -> tuple[float, bool]:
    """
    Mann-Kendall trend test.
    Returns (tau statistic, has_significant_trend).
    tau ≈ 0 and no significant trend → convergence.
    """
    n = len(scores)
    if n < 4:
        return 0.0, False

    s = 0
    for i in range(n - 1):
        for j in range(i + 1, n):
            diff = scores[j] - scores[i]
            if diff > 0:
                s += 1
            elif diff < 0:
                s -= 1

    tau = s / (n * (n - 1) / 2)

    variance_s = n * (n - 1) * (2 * n + 5) / 18
    if s > 0:
        z = (s - 1) / math.sqrt(variance_s)
    elif s < 0:
        z = (s + 1) / math.sqrt(variance_s)
    else:
        z = 0.0

    significant = abs(z) > 1.96  # 95% confidence
    return tau, significant

def is_converged_mk(scores: list[float], window: int = 10) -> bool:
    """Use Mann-Kendall test on recent window to detect convergence."""
    if len(scores) < window:
        return False
    recent = scores[-window:]
    tau, significant = mann_kendall_trend(recent)
    return not significant  # no significant trend = converged
```

#### Method 4: CUSUM (Cumulative Sum) Change Detection

Detect if recent scores have shifted from the running mean, indicating active improvement or degradation:

```python
def cusum_change_detected(
    scores: list[float],
    threshold: float = 1.0,
    drift: float = 0.5,
) -> bool:
    """
    CUSUM change detection. Returns True if a change (improvement or regression)
    is ongoing. Returns False when the process is stable (converged).
    """
    if len(scores) < 5:
        return True  # not enough data

    target = sum(scores[:5]) / 5  # reference mean from initial window
    s_pos = 0.0
    s_neg = 0.0

    for x in scores[5:]:
        s_pos = max(0, s_pos + (x - target) - drift)
        s_neg = max(0, s_neg - (x - target) - drift)

    return s_pos > threshold or s_neg > threshold
```

#### Composite Convergence Checker

NineS should combine multiple methods for robust convergence detection:

```python
@dataclass
class ConvergenceReport:
    is_converged: bool
    confidence: float
    methods_agreeing: int
    total_methods: int
    details: dict[str, bool]

def composite_convergence_check(
    scores: list[float],
    window: int = 5,
) -> ConvergenceReport:
    """Combine multiple statistical methods for robust convergence detection."""
    checks = {
        "sliding_variance": sliding_window_convergence(scores, window_size=window),
        "relative_improvement": relative_improvement_convergence(scores, window_size=min(3, window)),
        "mann_kendall": is_converged_mk(scores, window=max(window, 4)),
        "cusum_stable": not cusum_change_detected(scores),
    }
    agreeing = sum(checks.values())
    total = len(checks)
    return ConvergenceReport(
        is_converged=agreeing >= 3,  # majority vote
        confidence=agreeing / total,
        methods_agreeing=agreeing,
        total_methods=total,
        details=checks,
    )
```

### 3.5 Version-over-Version Comparison

Compare NineS performance across releases:

```python
from dataclasses import dataclass

@dataclass
class VersionComparison:
    base_version: str
    target_version: str
    improved: list[tuple[str, float, float]]   # (metric, old, new) where new > old
    regressed: list[tuple[str, float, float]]   # (metric, old, new) where new < old
    unchanged: list[tuple[str, float, float]]   # within tolerance
    overall_delta: float

def compare_versions(
    base: MeasurementSnapshot,
    target: MeasurementSnapshot,
    tolerance: float = 0.01,
) -> VersionComparison:
    """Compare two version snapshots, classifying each metric as improved/regressed/unchanged."""
    improved, regressed, unchanged = [], [], []

    all_keys = set(base.metrics) | set(target.metrics)
    for key in sorted(all_keys):
        old_val = base.metrics.get(key, 0.0)
        new_val = target.metrics.get(key, 0.0)
        delta = new_val - old_val

        if abs(delta) <= tolerance:
            unchanged.append((key, old_val, new_val))
        elif delta > 0:
            improved.append((key, old_val, new_val))
        else:
            regressed.append((key, old_val, new_val))

    old_avg = sum(base.metrics.values()) / max(len(base.metrics), 1)
    new_avg = sum(target.metrics.values()) / max(len(target.metrics), 1)
    overall_delta = (new_avg - old_avg) / max(abs(old_avg), 1e-6)

    return VersionComparison(
        base_version=base.version,
        target_version=target.version,
        improved=improved,
        regressed=regressed,
        unchanged=unchanged,
        overall_delta=overall_delta,
    )

def generate_regression_report(comparison: VersionComparison) -> str:
    """Generate a human-readable regression report."""
    lines = [
        f"# Version Comparison: {comparison.base_version} → {comparison.target_version}",
        f"Overall delta: {comparison.overall_delta:+.2%}",
        "",
    ]
    if comparison.regressed:
        lines.append("## ⚠ Regressions")
        for metric, old, new in comparison.regressed:
            lines.append(f"- **{metric}**: {old:.4f} → {new:.4f} ({new - old:+.4f})")
        lines.append("")

    if comparison.improved:
        lines.append("## ✓ Improvements")
        for metric, old, new in comparison.improved:
            lines.append(f"- **{metric}**: {old:.4f} → {new:.4f} ({new - old:+.4f})")
        lines.append("")

    lines.append(f"## Summary: {len(comparison.improved)} improved, "
                 f"{len(comparison.regressed)} regressed, "
                 f"{len(comparison.unchanged)} unchanged")
    return "\n".join(lines)
```

---

## Area 4: Lightweight Sandbox Solutions

NineS needs isolated execution environments for running evaluations without polluting the host system. The design uses a layered approach: process isolation via `subprocess`, filesystem isolation via `tempfile`, and Python environment isolation via `venv`.

### 4.1 Python `venv` Creation and Management (Programmatic)

```python
import venv
import subprocess
import sys
from pathlib import Path

class VenvFactory:
    """Create and manage isolated Python virtual environments."""

    def __init__(self, base_dir: Path) -> None:
        self.base_dir = base_dir
        self.base_dir.mkdir(parents=True, exist_ok=True)

    def create(self, name: str, requirements: list[str] | None = None) -> Path:
        """Create a venv and optionally install requirements."""
        venv_path = self.base_dir / name
        builder = venv.EnvBuilder(
            system_site_packages=False,
            clear=True,
            with_pip=True,
        )
        builder.create(str(venv_path))

        if requirements:
            pip = self._pip_path(venv_path)
            subprocess.run(
                [str(pip), "install", "--quiet", *requirements],
                check=True,
                capture_output=True,
                timeout=120,
            )

        return venv_path

    def destroy(self, name: str) -> None:
        """Remove a venv completely."""
        import shutil
        venv_path = self.base_dir / name
        if venv_path.exists():
            shutil.rmtree(venv_path)

    def python_path(self, venv_path: Path) -> Path:
        """Get the Python interpreter path for a venv."""
        if sys.platform == "win32":
            return venv_path / "Scripts" / "python.exe"
        return venv_path / "bin" / "python"

    def _pip_path(self, venv_path: Path) -> Path:
        if sys.platform == "win32":
            return venv_path / "Scripts" / "pip.exe"
        return venv_path / "bin" / "pip"

    def list_installed(self, venv_path: Path) -> list[str]:
        """List packages installed in the venv."""
        pip = self._pip_path(venv_path)
        result = subprocess.run(
            [str(pip), "list", "--format=freeze"],
            capture_output=True,
            text=True,
            check=True,
        )
        return result.stdout.strip().splitlines()
```

### 4.2 `subprocess` Isolation

Run evaluation code in isolated processes with resource limits:

```python
import subprocess
import json
import os
import signal
from dataclasses import dataclass

@dataclass
class ExecutionResult:
    exit_code: int
    stdout: str
    stderr: str
    duration_ms: float
    timed_out: bool
    memory_exceeded: bool

def run_isolated(
    python_path: str,
    script_path: str,
    working_dir: str,
    timeout_seconds: int = 30,
    max_memory_mb: int = 512,
    env_override: dict[str, str] | None = None,
    seed: int | None = None,
) -> ExecutionResult:
    """
    Run a Python script in a subprocess with timeout and memory constraints.

    Uses resource limits via preexec_fn on Linux to cap memory usage.
    """
    env = os.environ.copy()
    env["PYTHONDONTWRITEBYTECODE"] = "1"
    env["PYTHONHASHSEED"] = str(seed) if seed is not None else "0"

    if seed is not None:
        env["NINES_SEED"] = str(seed)

    if env_override:
        env.update(env_override)

    def set_limits() -> None:
        """Set memory limits via resource module (Linux only)."""
        try:
            import resource
            mem_bytes = max_memory_mb * 1024 * 1024
            resource.setrlimit(resource.RLIMIT_AS, (mem_bytes, mem_bytes))
        except (ImportError, ValueError):
            pass  # not available on this platform

    import time
    start = time.monotonic()
    timed_out = False
    memory_exceeded = False

    try:
        proc = subprocess.run(
            [python_path, script_path],
            cwd=working_dir,
            env=env,
            capture_output=True,
            text=True,
            timeout=timeout_seconds,
            preexec_fn=set_limits,
        )
        exit_code = proc.returncode
        stdout = proc.stdout
        stderr = proc.stderr
    except subprocess.TimeoutExpired as exc:
        timed_out = True
        exit_code = -1
        stdout = exc.stdout.decode("utf-8", errors="replace") if exc.stdout else ""
        stderr = exc.stderr.decode("utf-8", errors="replace") if exc.stderr else ""
    except MemoryError:
        memory_exceeded = True
        exit_code = -2
        stdout = ""
        stderr = "MemoryError: exceeded limit"

    duration_ms = (time.monotonic() - start) * 1000

    return ExecutionResult(
        exit_code=exit_code,
        stdout=stdout,
        stderr=stderr,
        duration_ms=duration_ms,
        timed_out=timed_out,
        memory_exceeded=memory_exceeded,
    )
```

### 4.3 `tempfile`/`tmpdir` Strategies for Filesystem Isolation

```python
import tempfile
import shutil
from contextlib import contextmanager
from pathlib import Path

@contextmanager
def isolated_workspace(
    prefix: str = "nines_sandbox_",
    copy_files: dict[str, str] | None = None,
):
    """
    Context manager providing an isolated temporary workspace.
    Cleans up on exit. Yields a Path to the workspace root.

    copy_files: {destination_relative_path: source_path}
    """
    tmpdir = Path(tempfile.mkdtemp(prefix=prefix))
    try:
        if copy_files:
            for dest_rel, src_path in copy_files.items():
                dest = tmpdir / dest_rel
                dest.parent.mkdir(parents=True, exist_ok=True)
                src = Path(src_path)
                if src.is_dir():
                    shutil.copytree(src, dest)
                else:
                    shutil.copy2(src, dest)

        yield tmpdir
    finally:
        shutil.rmtree(tmpdir, ignore_errors=True)

# Usage:
# with isolated_workspace(copy_files={"task.py": "/path/to/task.py"}) as ws:
#     result = run_isolated(python, str(ws / "task.py"), str(ws))
```

#### Full Sandbox Manager

```python
from dataclasses import dataclass, field
from pathlib import Path
import uuid

@dataclass
class SandboxConfig:
    timeout_seconds: int = 30
    max_memory_mb: int = 512
    requirements: list[str] = field(default_factory=list)
    seed: int | None = None
    keep_on_failure: bool = False

@dataclass
class Sandbox:
    id: str
    workspace: Path
    venv_path: Path
    config: SandboxConfig

class SandboxManager:
    """Lifecycle management for isolated execution sandboxes."""

    def __init__(self, base_dir: Path | None = None) -> None:
        self.base_dir = base_dir or Path(tempfile.gettempdir()) / "nines_sandboxes"
        self.base_dir.mkdir(parents=True, exist_ok=True)
        self.venv_factory = VenvFactory(self.base_dir / "venvs")
        self._active: dict[str, Sandbox] = {}

    def create(self, config: SandboxConfig | None = None) -> Sandbox:
        """Create a new sandbox with isolated workspace and venv."""
        config = config or SandboxConfig()
        sandbox_id = uuid.uuid4().hex[:12]

        workspace = self.base_dir / "workspaces" / sandbox_id
        workspace.mkdir(parents=True, exist_ok=True)

        venv_path = self.venv_factory.create(
            name=sandbox_id,
            requirements=config.requirements,
        )

        sandbox = Sandbox(
            id=sandbox_id,
            workspace=workspace,
            venv_path=venv_path,
            config=config,
        )
        self._active[sandbox_id] = sandbox
        return sandbox

    def execute(self, sandbox: Sandbox, script_content: str) -> ExecutionResult:
        """Write and execute a script inside the sandbox."""
        script_path = sandbox.workspace / "run.py"
        script_path.write_text(script_content, encoding="utf-8")

        python = str(self.venv_factory.python_path(sandbox.venv_path))
        return run_isolated(
            python_path=python,
            script_path=str(script_path),
            working_dir=str(sandbox.workspace),
            timeout_seconds=sandbox.config.timeout_seconds,
            max_memory_mb=sandbox.config.max_memory_mb,
            seed=sandbox.config.seed,
        )

    def destroy(self, sandbox_id: str) -> None:
        """Clean up all sandbox resources."""
        sandbox = self._active.pop(sandbox_id, None)
        if sandbox is None:
            return
        shutil.rmtree(sandbox.workspace, ignore_errors=True)
        self.venv_factory.destroy(sandbox_id)

    def destroy_all(self) -> None:
        for sid in list(self._active):
            self.destroy(sid)
```

### 4.4 Seed Control for Deterministic Execution

```python
import hashlib
import json

def deterministic_env(seed: int) -> dict[str, str]:
    """Environment variables to maximize deterministic behavior."""
    return {
        "PYTHONHASHSEED": str(seed),
        "NINES_SEED": str(seed),
        "CUBLAS_WORKSPACE_CONFIG": ":4096:8",  # deterministic CUDA
        "TF_DETERMINISTIC_OPS": "1",
    }

def seed_init_snippet(seed: int) -> str:
    """Python code to prepend to evaluation scripts for seed control."""
    return f"""\
import random
import os

_SEED = int(os.environ.get("NINES_SEED", {seed}))
random.seed(_SEED)

try:
    import numpy as np
    np.random.seed(_SEED)
except ImportError:
    pass

try:
    import torch
    torch.manual_seed(_SEED)
    torch.backends.cudnn.deterministic = True
    torch.backends.cudnn.benchmark = False
except ImportError:
    pass
"""

def result_fingerprint(result: ExecutionResult) -> str:
    """Compute a deterministic hash of execution output for comparison."""
    content = json.dumps({
        "exit_code": result.exit_code,
        "stdout": result.stdout.strip(),
        "stderr": result.stderr.strip(),
    }, sort_keys=True)
    return hashlib.sha256(content.encode()).hexdigest()
```

### 4.5 Pollution Detection

Verify that sandbox execution didn't modify the host environment:

```python
import os
import hashlib
from pathlib import Path
from dataclasses import dataclass

@dataclass
class EnvironmentSnapshot:
    """Snapshot of observable host state before/after sandbox execution."""
    env_vars: dict[str, str]
    watched_file_hashes: dict[str, str]
    watched_dir_listings: dict[str, list[str]]
    python_path: list[str]

def take_snapshot(
    watched_dirs: list[Path] | None = None,
    watched_files: list[Path] | None = None,
) -> EnvironmentSnapshot:
    """Capture current host environment state."""
    file_hashes: dict[str, str] = {}
    for f in (watched_files or []):
        if f.exists():
            file_hashes[str(f)] = hashlib.sha256(f.read_bytes()).hexdigest()

    dir_listings: dict[str, list[str]] = {}
    for d in (watched_dirs or []):
        if d.exists():
            dir_listings[str(d)] = sorted(str(p) for p in d.rglob("*"))

    return EnvironmentSnapshot(
        env_vars=dict(os.environ),
        watched_file_hashes=file_hashes,
        watched_dir_listings=dir_listings,
        python_path=list(sys.path),
    )

@dataclass
class PollutionReport:
    clean: bool
    env_var_changes: list[str]
    file_changes: list[str]
    dir_changes: list[str]
    path_changes: list[str]

def detect_pollution(
    before: EnvironmentSnapshot,
    after: EnvironmentSnapshot,
) -> PollutionReport:
    """Compare two snapshots to detect host environment changes."""
    env_changes: list[str] = []
    for key in set(before.env_vars) | set(after.env_vars):
        old = before.env_vars.get(key)
        new = after.env_vars.get(key)
        if old != new:
            env_changes.append(f"{key}: {old!r} → {new!r}")

    file_changes: list[str] = []
    for path in set(before.watched_file_hashes) | set(after.watched_file_hashes):
        old_hash = before.watched_file_hashes.get(path)
        new_hash = after.watched_file_hashes.get(path)
        if old_hash != new_hash:
            if old_hash is None:
                file_changes.append(f"CREATED: {path}")
            elif new_hash is None:
                file_changes.append(f"DELETED: {path}")
            else:
                file_changes.append(f"MODIFIED: {path}")

    dir_changes: list[str] = []
    for d in set(before.watched_dir_listings) | set(after.watched_dir_listings):
        old_listing = set(before.watched_dir_listings.get(d, []))
        new_listing = set(after.watched_dir_listings.get(d, []))
        added = new_listing - old_listing
        removed = old_listing - new_listing
        if added:
            dir_changes.append(f"ADDED in {d}: {added}")
        if removed:
            dir_changes.append(f"REMOVED in {d}: {removed}")

    path_changes = []
    if before.python_path != after.python_path:
        added = set(after.python_path) - set(before.python_path)
        removed = set(before.python_path) - set(after.python_path)
        if added:
            path_changes.append(f"sys.path ADDED: {added}")
        if removed:
            path_changes.append(f"sys.path REMOVED: {removed}")

    return PollutionReport(
        clean=not (env_changes or file_changes or dir_changes or path_changes),
        env_var_changes=env_changes,
        file_changes=file_changes,
        dir_changes=dir_changes,
        path_changes=path_changes,
    )
```

#### Integrated Pollution-Checked Execution

```python
def execute_with_pollution_check(
    sandbox_manager: SandboxManager,
    sandbox: Sandbox,
    script: str,
    watched_dirs: list[Path] | None = None,
    watched_files: list[Path] | None = None,
) -> tuple[ExecutionResult, PollutionReport]:
    """Execute in sandbox and verify no host pollution occurred."""
    before = take_snapshot(watched_dirs=watched_dirs, watched_files=watched_files)
    result = sandbox_manager.execute(sandbox, script)
    after = take_snapshot(watched_dirs=watched_dirs, watched_files=watched_files)

    report = detect_pollution(before, after)
    if not report.clean:
        import logging
        logging.error(
            "Sandbox %s caused host pollution: env=%d, files=%d, dirs=%d, path=%d",
            sandbox.id,
            len(report.env_var_changes),
            len(report.file_changes),
            len(report.dir_changes),
            len(report.path_changes),
        )
    return result, report
```

### 4.6 Multi-Round Re-Test Convergence

Verify result stability by running the same evaluation multiple times:

```python
from dataclasses import dataclass
from collections import Counter
import math

@dataclass
class StabilityReport:
    total_runs: int
    unique_results: int
    dominant_result_count: int
    dominant_fingerprint: str
    is_stable: bool
    stability_ratio: float
    fingerprint_distribution: dict[str, int]

def multi_round_stability_test(
    sandbox_manager: SandboxManager,
    config: SandboxConfig,
    script: str,
    rounds: int = 5,
    stability_threshold: float = 1.0,
) -> StabilityReport:
    """
    Run the same script multiple times and verify output convergence.

    Args:
        stability_threshold: Fraction of runs that must produce identical output.
            1.0 = all runs must match (strict determinism).
            0.8 = 80% of runs must agree (allows minor non-determinism).
    """
    fingerprints: list[str] = []

    for _ in range(rounds):
        sandbox = sandbox_manager.create(config)
        try:
            result = sandbox_manager.execute(sandbox, script)
            fp = result_fingerprint(result)
            fingerprints.append(fp)
        finally:
            sandbox_manager.destroy(sandbox.id)

    counts = Counter(fingerprints)
    dominant_fp, dominant_count = counts.most_common(1)[0]

    return StabilityReport(
        total_runs=rounds,
        unique_results=len(counts),
        dominant_result_count=dominant_count,
        dominant_fingerprint=dominant_fp,
        is_stable=(dominant_count / rounds) >= stability_threshold,
        stability_ratio=dominant_count / rounds,
        fingerprint_distribution=dict(counts),
    )

def adaptive_stability_test(
    sandbox_manager: SandboxManager,
    config: SandboxConfig,
    script: str,
    min_rounds: int = 3,
    max_rounds: int = 10,
    confidence: float = 0.95,
) -> StabilityReport:
    """
    Adaptive testing: stop early if results are clearly stable or clearly unstable.
    Uses a sequential probability ratio test (SPRT) approach.
    """
    fingerprints: list[str] = []

    for i in range(max_rounds):
        sandbox = sandbox_manager.create(config)
        try:
            result = sandbox_manager.execute(sandbox, script)
            fingerprints.append(result_fingerprint(result))
        finally:
            sandbox_manager.destroy(sandbox.id)

        if i + 1 >= min_rounds:
            counts = Counter(fingerprints)
            dominant_count = counts.most_common(1)[0][1]
            total = len(fingerprints)

            if dominant_count == total:
                break  # perfectly stable

            # Wilson score lower bound for proportion
            p_hat = dominant_count / total
            z = 1.96  # 95% CI
            denom = 1 + z**2 / total
            center = p_hat + z**2 / (2 * total)
            margin = z * math.sqrt(p_hat * (1 - p_hat) / total + z**2 / (4 * total**2))
            lower_bound = (center - margin) / denom

            if lower_bound >= confidence:
                break  # statistically confident enough

    counts = Counter(fingerprints)
    dominant_fp, dominant_count = counts.most_common(1)[0]

    return StabilityReport(
        total_runs=len(fingerprints),
        unique_results=len(counts),
        dominant_result_count=dominant_count,
        dominant_fingerprint=dominant_fp,
        is_stable=(dominant_count / len(fingerprints)) >= confidence,
        stability_ratio=dominant_count / len(fingerprints),
        fingerprint_distribution=dict(counts),
    )
```

---

## Summary: Key Technical Decisions for NineS

| Area | Recommended Approach | Rationale |
|------|---------------------|-----------|
| GitHub API | GraphQL as primary, REST for simple lookups | Fewer round-trips, single request for complex data |
| arXiv API | `arxiv` Python library + direct API for bulk | Library handles pagination and retries |
| Rate limiting | Token-bucket + response-header adaptation | Respects documented limits while maximizing throughput |
| AST analysis | Built-in `ast` module | Zero dependencies, full Python support |
| Architecture detection | Heuristic multi-signal with confidence scoring | Pragmatic; avoids false positives |
| Decomposition | Three strategies (functional/concern/layer) | Different analyses need different decomposition views |
| Feedback loop | MAPIM with typed intermediate artifacts | Clear phase separation, auditable history |
| Convergence | Composite of 4 statistical methods (majority vote) | Robust against any single method's blind spots |
| Sandbox | `venv` + `subprocess` + `tempfile` (3-layer) | Lightweight, no Docker dependency, sufficient isolation for MVP |
| Determinism | Seed control + output fingerprinting | Verifiable reproducibility without heavyweight tooling |
| Pollution detection | Before/after snapshot diffing | Simple, extensible, covers fs + env + sys.path |
| Stability testing | Adaptive SPRT with Wilson confidence bounds | Minimizes unnecessary re-runs while maintaining statistical rigor |

---

*Last modified: 2026-04-11T00:00:00Z*