Roadmap: AI-Embedded Graph Operating System

Roadmap: AI-Embedded Graph Operating System

From GraFOSa to GraFOSa/AI — Embedding Intelligence Into the Graph Kernel

Date: 2026-03-05
Status: Proposal for review
Scope: 3 layers of work to transform GraFOSa from an externally-consumed analysis tool into a self-aware, AI-embedded graph operating system.

1. Where We Are

GraFOSa already has the infrastructure that most "AI OS" projects would need to build from scratch:

Total existing code: ~25,000+ lines across kernel, 8 layers, IR compiler, executor, pipeline, scanners.

What's Missing

The AI currently lives outside the graph. Claude (or any LLM) consumes MCP tools to query and analyze the graph. The graph itself has no capacity to:

1. Invoke inference as part of a compiled plan
2. Perceive changes continuously (it rebuilds on demand)
3. Act autonomously (it reports findings; humans decide what to do)

These three gaps define the three layers of work.

2. Architecture: The Three-Layer Addition

                    CURRENT                              TARGET
                    ───────                              ──────

              Claude / User                        Claude / User
                   │                                    │
                   ▼                                    ▼
            ┌─────────────┐                      ┌─────────────┐
            │  MCP Tools  │                      │  MCP Tools  │
            │  CLI (50+)  │                      │  CLI (50+)  │
            └──────┬──────┘                      └──────┬──────┘
                   │                                    │
                   ▼                                    ▼
            ┌─────────────┐                      ┌─────────────┐
            │ L8 Strategy │                      │ L8 Strategy │
            │ L7 Intel    │                      │ L7 Intel    │
            │ L6 Sched    │                      │ L7.5 CORTEX │ ◄── NEW: Autonomous reasoning
            │ L5 Exec     │                      │ L6 Sched    │
            │ L4 Proc     │                      │ L5 Exec     │
            │ L3 Net      │                      │ L4 Proc     │
            │ L2 Mod      │                      │ L3 Net      │
            │ L1 FS       │                      │ L2 Mod      │
            └──────┬──────┘                      │ L1 FS       │
                   │                              └──────┬──────┘
                   ▼                                     │
            ┌─────────────┐                              ▼
            │ L0 Kernel   │                      ┌──────────────┐
            │ (GraphKernel│                      │ L0 Kernel    │
            │  + IR)      │                      │ + IR + INFER │ ◄── NEW: Inference as graph op
            └─────────────┘                      └──────┬───────┘
                                                        │
                                                        ▼
                                                 ┌──────────────┐
                                                 │ L-1 SENTINEL │ ◄── NEW: Continuous perception
                                                 │ (file watch, │
                                                 │  git hooks,  │
                                                 │  OTEL ingest)│
                                                 └──────────────┘

3. Layer 1: INFER — Inference as a Graph Operation

3.1 Problem Statement

The Graph IR has 38 OpTypes covering scans, traversals, filters, algorithms, mutations, and control flow. None of them can invoke an LLM. To embed AI in the graph, inference must become a first-class operation — schedulable, optimizable, and composable with other graph operations.

3.2 Design

3.2.1 New OpTypes (add to ir.py)

3.2.2 Inference Backend Registry

Like the kernel has pluggable storage backends, inference gets pluggable model backends:

# New file: src/grafosa/kernels/graph_ir/inference.py

class InferenceBackend(Protocol):
    """Backend-agnostic inference interface."""

    def infer(self, prompt: str, schema: dict | None = None) -> dict: ...
    def embed(self, texts: list[str], dimensions: int = 0) -> list[list[float]]: ...
    def classify(self, text: str, categories: list[str]) -> str: ...

class AnthropicBackend(InferenceBackend):
    """Claude API backend."""
    ...

class OpenAIBackend(InferenceBackend):
    """OpenAI API backend."""
    ...

class OllamaBackend(InferenceBackend):
    """Local Ollama models — zero-cost inference for development."""
    ...

class MockBackend(InferenceBackend):
    """Deterministic mock for testing — returns canned responses."""
    ...

_REGISTRY: dict[str, InferenceBackend] = {}

def register_backend(name: str, backend: InferenceBackend) -> None: ...
def get_backend(name: str = "default") -> InferenceBackend: ...

3.2.3 Compiler Handlers

Add to compiler.py:

def _infer(self, node: IRNode, lines: list[str]) -> str:
    """Compile INFER op: gather input from child, format prompt, call backend."""
    # 1. Compile child op (the graph data to reason about)
    # 2. Format prompt_template with child results
    # 3. Call _inference_backend.infer(prompt, schema)
    # 4. Return structured result

def _embed(self, node: IRNode, lines: list[str]) -> str:
    """Compile EMBED op: extract text from nodes, call embed backend."""
    # 1. Compile child op (set of nodes)
    # 2. Extract text (name, fqn, properties) from each node
    # 3. Call _inference_backend.embed(texts)
    # 4. Store embeddings in kernel metadata or return

def _classify(self, node: IRNode, lines: list[str]) -> str:
    """Compile CLASSIFY op: classify nodes into categories."""
    ...

def _semantic_search(self, node: IRNode, lines: list[str]) -> str:
    """Compile SEMANTIC_SEARCH: query against vector index."""
    ...

def _reason(self, node: IRNode, lines: list[str]) -> str:
    """Compile REASON op: multi-step reasoning with evidence collection."""
    ...

3.2.4 Optimizer Integration

New optimizer pass: defer_inference (Pass 7)

• Move INFER/REASON ops as late as possible in PIPE chains (they're expensive)
• Batch EMBED ops: if multiple EMBED ops target the same model, merge into one batch call
• Parallelize independent INFER ops: if two INFER ops have no data dependency, wrap in PARALLEL
• Cost model: INFER base cost = 100.0 (vs PAGERANK = 5.0), EMBED = 50.0, CLASSIFY = 20.0

3.2.5 Context Windowing

LLMs have finite context. Graph data can be enormous. The compiler needs a context budget:

class ContextBudget:
    """Manages how much graph data gets sent to the LLM."""
    max_tokens: int = 8000          # Default context budget per INFER op
    node_format: str = "compact"    # "compact" | "full" | "minimal"
    priority: str = "pagerank"      # Rank nodes by importance when truncating

    def fit(self, nodes: list[Node], edges: list[Edge]) -> str:
        """Format graph data to fit within token budget."""
        ...

3.2.6 Example: AI-Enhanced Dead Code Detection

# Current: binary detection (reachable or not)
dead = exec_.dead_code()

# With INFER: semantic understanding of "intentionally unused" vs "accidentally dead"
plan = IRNode(OpType.PIPE, children=[
    # Step 1: Find structurally dead symbols
    IRNode(OpType.SCAN_NODES, params={"label": "Symbol", "properties": {"_dead": True}}),

    # Step 2: Ask LLM to classify each one
    IRNode(OpType.CLASSIFY, params={
        "categories": ["accidentally_dead", "intentionally_unused", "test_helper", "deprecated"],
        "model": "default",
    }),

    # Step 3: Filter to only accidentally dead
    IRNode(OpType.FILTER_PROPERTY, params={"key": "_classification", "value": "accidentally_dead"}),
])

result = kernel.execute(plan)

3.3 Files

3.4 Dependencies

• anthropic (optional) — Claude API
• openai (optional) — OpenAI API
• ollama (optional) — Local models
• tiktoken or anthropic.count_tokens (optional) — Token counting
• All backends are optional; MockBackend is always available

3.5 Risk Assessment

4. Layer 2: SENTINEL — Continuous Perception

4.1 Problem Statement

The graph is currently rebuilt on demand (grafosa build or incremental_build()). For an AI-embedded system, the graph must perceive changes continuously — file saves, git commits, OTEL traces — and update itself in real-time.

4.2 Design

4.2.1 Event Sources

4.2.2 Sentinel Daemon

# New file: src/grafosa/sentinel/daemon.py

class Sentinel:
    """Continuous perception daemon — keeps the graph alive."""

    def __init__(
        self,
        kernel: GraphKernel,
        code_stack: dict,
        config: SentinelConfig,
    ) -> None:
        self._kernel = kernel
        self._stack = code_stack
        self._event_queue: asyncio.Queue[GraphEvent] = asyncio.Queue()
        self._watchers: list[EventSource] = []
        self._subscribers: list[Callable[[GraphEvent], Awaitable[None]]] = []

    async def start(self) -> None:
        """Start all watchers and the event processing loop."""
        ...

    async def stop(self) -> None:
        """Gracefully stop all watchers."""
        ...

    async def _process_events(self) -> None:
        """Main loop: dequeue events, update graph, notify subscribers."""
        ...

    def subscribe(self, callback: Callable[[GraphEvent], Awaitable[None]]) -> None:
        """Register a subscriber for graph events."""
        ...

4.2.3 Event Types

# New file: src/grafosa/sentinel/events.py

@dataclass
class GraphEvent:
    """A change observed by the Sentinel."""
    source: str              # "filesystem", "git", "otel", "feedback"
    kind: str                # "file_changed", "commit", "trace", "verdict"
    timestamp: float
    payload: dict[str, Any]

@dataclass
class FileChangeEvent(GraphEvent):
    path: str
    change_type: str         # "created", "modified", "deleted"

@dataclass
class GitEvent(GraphEvent):
    sha: str
    message: str
    files_changed: list[str]

@dataclass
class TraceEvent(GraphEvent):
    trace_id: str
    spans: list[dict]        # OTEL span data

@dataclass
class FeedbackEvent(GraphEvent):
    category: str
    entity: str
    verdict: str             # "true_positive", "false_positive"

4.2.4 Incremental Graph Update Pipeline

When a FileChangeEvent arrives:

FileChangeEvent
    │
    ▼
┌─────────────────────────────────────────────────────────────┐
│ 1. DEBOUNCE (100ms)                                         │
│    Batch rapid successive changes to the same file          │
└────────────────────────┬────────────────────────────────────┘
                         ▼
┌─────────────────────────────────────────────────────────────┐
│ 2. DELTA COMPUTE (existing GraphDelta)                      │
│    Hash comparison → changed/added/deleted classification   │
└────────────────────────┬────────────────────────────────────┘
                         ▼
┌─────────────────────────────────────────────────────────────┐
│ 3. SURGICAL UPDATE                                          │
│    Remove affected File + child Symbols + incident edges    │
│    Re-scan only the changed file (PythonScanner/TSScanner)  │
│    Re-wire IMPORTS, CALLS edges                             │
│    Invalidate caches (generation++)                         │
└────────────────────────┬────────────────────────────────────┘
                         ▼
┌─────────────────────────────────────────────────────────────┐
│ 4. IMPACT ASSESSMENT                                        │
│    Compute blast radius of the change                       │
│    Identify affected services                               │
│    Check if any health thresholds crossed                   │
└────────────────────────┬────────────────────────────────────┘
                         ▼
┌─────────────────────────────────────────────────────────────┐
│ 5. NOTIFY SUBSCRIBERS                                       │
│    Emit GraphEvent to Cortex, MCP clients, CLI watchers     │
└─────────────────────────────────────────────────────────────┘

4.2.5 OTEL Trace Integration

When TraceEvent arrives:

TraceEvent (distributed trace spans)
    │
    ▼
┌─────────────────────────────────────────────────────────────┐
│ 1. SPAN → EDGE CORRELATION                                  │
│    Map span service+operation to Service+Endpoint nodes     │
│    Map span-to-span parent→child to HTTP_DEPENDS edges      │
│    Map messaging spans to NATS_PUBLISHES/SUBSCRIBES edges   │
└────────────────────────┬────────────────────────────────────┘
                         ▼
┌─────────────────────────────────────────────────────────────┐
│ 2. VALIDATION                                               │
│    Compare trace-observed edges vs graph-predicted edges    │
│    True positive: edge exists and trace confirms           │
│    False positive: edge exists but trace never traverses   │
│    False negative: trace shows call not in graph           │
└────────────────────────┬────────────────────────────────────┘
                         ▼
┌─────────────────────────────────────────────────────────────┐
│ 3. FEEDBACK                                                 │
│    Feed TP/FP/FN to AccuracyTracker                        │
│    Update confidence scores on edges and sections          │
│    Enrich edge properties with latency, error_rate         │
└─────────────────────────────────────────────────────────────┘

4.2.6 Health Dashboard Streaming

Instead of point-in-time health_report() calls, Sentinel enables a streaming health model:

class HealthStream:
    """Streaming health updates triggered by graph mutations."""

    def __init__(self, intel: GraphIntel, sentinel: Sentinel) -> None:
        self._intel = intel
        sentinel.subscribe(self._on_graph_change)
        self._current_report: HealthReport | None = None
        self._subscribers: list[Callable] = []

    async def _on_graph_change(self, event: GraphEvent) -> None:
        """Re-run affected health sections when graph changes."""
        affected_sections = self._identify_affected_sections(event)
        for section in affected_sections:
            # Re-run only the affected section (not full report)
            updated = self._intel._run_section(section)
            self._patch_report(section, updated)

        # Notify subscribers (MCP clients, CLI, Cortex)
        for sub in self._subscribers:
            await sub(self._current_report)

4.3 Files

4.4 Dependencies

• watchfiles — Rust-based file watcher (cross-platform, async-native)
• opentelemetry-proto (optional) — OTEL protobuf definitions for trace ingestion

4.5 CLI Integration

# Start the Sentinel daemon
grafosa sentinel start --graph graph.json --watch /path/to/project

# Show live health dashboard (terminal UI)
grafosa sentinel dashboard

# Show event stream
grafosa sentinel events --tail

# Stop
grafosa sentinel stop

4.6 MCP Integration

@mcp.tool()
def sentinel_status() -> str:
    """Show Sentinel daemon status, event counts, last update time."""
    ...

@mcp.tool()
def sentinel_events(limit: int = 20) -> str:
    """Show recent graph events from the Sentinel."""
    ...

5. Layer 3: CORTEX — Autonomous Reasoning Loop

5.1 Problem Statement

With INFER ops in the kernel and Sentinel feeding continuous updates, the final piece is a reasoning loop that connects perception → analysis → decision → action → observation. This is "L7.5 Cortex" — it sits between Intel/Strategy and the execution engine, autonomously deciding when and how to act.

5.2 Design

5.2.1 The Reasoning Loop

         ┌──────────────────────────────────────┐
         │                                      │
         ▼                                      │
    ┌─────────┐     ┌──────────┐     ┌─────────┤
    │ PERCEIVE│────▶│  REASON  │────▶│   ACT   │
    └─────────┘     └──────────┘     └─────────┘
         ▲                                │
         │          ┌──────────┐          │
         └──────────│ OBSERVE  │◀─────────┘
                    └──────────┘

PERCEIVE: Sentinel emits GraphEvent. Cortex receives it.

REASON: Cortex uses INFER ops + Intel analysis + Strategy scoring to decide:

• Is this change significant enough to act on?
• What are the options? (approach ranking from Strategy)
• What's the risk? (blast radius, confidence)
• Should a human be consulted?

ACT: Execute the chosen action:

• Auto-fix (for high-confidence, low-risk, reversible actions)
• Recommend (for medium-confidence or higher-risk actions)
• Alert (for critical issues requiring immediate human attention)
• Learn (update AccuracyTracker, adjust thresholds)

OBSERVE: After acting, verify the outcome:

• Did the fix resolve the issue? (re-run health check on affected section)
• Did anything regress? (compare before/after scores)
• Feed results back to AccuracyTracker

5.2.2 Action Types

5.2.3 Cortex Core

# New file: src/grafosa/cortex/engine.py

class Cortex:
    """Autonomous reasoning engine — the graph's 'brain'."""

    def __init__(
        self,
        code_stack: dict,
        sentinel: Sentinel,
        config: CortexConfig,
    ) -> None:
        self._stack = code_stack
        self._intel = GraphIntel.from_code_stack(code_stack)
        self._strategy = code_stack.get("strategy")
        self._sentinel = sentinel
        self._config = config
        self._action_log: list[CortexAction] = []

        # Subscribe to Sentinel events
        sentinel.subscribe(self._on_event)

    async def _on_event(self, event: GraphEvent) -> None:
        """Main entry point: perceive → reason → act → observe."""

        # 1. PERCEIVE: Is this event significant?
        significance = self._assess_significance(event)
        if significance < self._config.min_significance:
            return

        # 2. REASON: What should we do?
        decision = await self._reason(event, significance)

        # 3. ACT: Execute the decision
        result = await self._act(decision)

        # 4. OBSERVE: Verify the outcome
        await self._observe(decision, result)

    def _assess_significance(self, event: GraphEvent) -> float:
        """Score event significance 0-1 based on blast radius and issue impact."""
        ...

    async def _reason(self, event: GraphEvent, significance: float) -> CortexDecision:
        """Use Intel + Strategy + optional INFER to decide on action."""
        ...

    async def _act(self, decision: CortexDecision) -> ActionResult:
        """Execute the chosen action."""
        ...

    async def _observe(self, decision: CortexDecision, result: ActionResult) -> None:
        """Verify outcome and feed back to AccuracyTracker."""
        ...

5.2.4 Reasoning Strategies

Cortex supports multiple reasoning strategies, configurable per event type:

1. Rule-Based (default, zero-cost)

class RuleBasedReasoner:
    """If-then rules derived from Intel + Strategy analysis."""

    def reason(self, event, health_delta) -> CortexDecision:
        # New critical issue introduced → Alert
        if health_delta.new_criticals > 0:
            return CortexDecision(action="alert", confidence=0.95)

        # Score dropped >10 points → Recommend
        if health_delta.score_change < -10:
            return CortexDecision(action="recommend", confidence=0.8)

        # New dead code introduced → Annotate
        if health_delta.new_dead_code:
            return CortexDecision(action="annotate", confidence=0.9)

        # Threshold auto-adjustment validated by git → Learn
        if health_delta.accuracy_signal:
            return CortexDecision(action="learn", confidence=0.85)

        return CortexDecision(action="none")

2. LLM-Augmented (uses INFER ops)

class LLMReasoner:
    """Uses graph IR INFER ops for nuanced reasoning."""

    async def reason(self, event, context_nodes) -> CortexDecision:
        plan = IRNode(OpType.PIPE, children=[
            # Gather context: affected nodes + neighbors + recent history
            IRNode(OpType.TRAVERSE_BFS, params={"max_depth": 2}),

            # Ask LLM to reason about the change
            IRNode(OpType.REASON, params={
                "question": f"A {event.kind} occurred affecting {event.payload}. "
                            "Given the graph context, what action should be taken?",
                "strategy": "chain_of_thought",
            }),
        ])
        result = self._kernel.execute(plan)
        return CortexDecision.from_llm_response(result)

3. Ensemble (combine both)

class EnsembleReasoner:
    """Rule-based for speed, LLM for ambiguous cases."""

    async def reason(self, event, context) -> CortexDecision:
        rule_decision = self._rules.reason(event, context)

        # If rules are confident (>0.8), use them (fast, free)
        if rule_decision.confidence > 0.8:
            return rule_decision

        # Otherwise, escalate to LLM (slower, costs tokens)
        llm_decision = await self._llm.reason(event, context)

        # If they agree, boost confidence
        if rule_decision.action == llm_decision.action:
            llm_decision.confidence = min(1.0, llm_decision.confidence + 0.1)

        return llm_decision

5.2.5 Approval Model

Actions above a risk threshold require human approval:

@dataclass
class CortexConfig:
    """Cortex autonomy configuration."""

    # Significance threshold (0-1). Events below this are ignored.
    min_significance: float = 0.3

    # Reasoning strategy
    reasoner: str = "ensemble"  # "rule_based", "llm", "ensemble"

    # Action approval thresholds
    auto_approve: list[str] = field(default_factory=lambda: ["learn", "annotate"])
    notify_only: list[str] = field(default_factory=lambda: ["recommend", "alert"])
    require_approval: list[str] = field(default_factory=lambda: ["auto_fix", "block"])

    # Safety limits
    max_auto_actions_per_hour: int = 10
    max_token_budget_per_hour: int = 50000

    # Feedback
    auto_observe: bool = True  # Re-check after acting

5.2.6 Integration with Existing Feedback Loop

Cortex doesn't replace the existing AccuracyTracker → SelfValidator → PipelineConfig loop. It drives it:

Current (manual):
  User runs validate-self → sees results → manually adjusts config

With Cortex (autonomous):
  Sentinel detects git commit → Cortex runs validate-self automatically
  → If correlation strengthens, Cortex calls "learn" action
  → AccuracyTracker adjusts severity
  → Next health_report() uses updated thresholds
  → Cortex observes score change → feeds back

5.2.7 Audit Trail

Every Cortex action is logged:

@dataclass
class CortexAction:
    """Immutable record of a Cortex decision and action."""
    timestamp: float
    event: GraphEvent
    significance: float
    decision: CortexDecision
    reasoner: str
    result: ActionResult
    observation: ObservationResult | None

    def to_dict(self) -> dict: ...

Stored in SQLite (alongside AccuracyTracker) for full auditability.

5.3 Files

5.4 CLI Integration

# Start Cortex (requires Sentinel)
grafosa cortex start --graph graph.json --watch /path/to/project

# Show Cortex audit log
grafosa cortex log --limit 50

# Show current Cortex config
grafosa cortex config

# Adjust autonomy level
grafosa cortex config --set reasoner=rule_based
grafosa cortex config --set min_significance=0.5

5.5 MCP Integration

@mcp.tool()
def cortex_status() -> str:
    """Show Cortex status, recent decisions, action counts."""
    ...

@mcp.tool()
def cortex_log(limit: int = 20) -> str:
    """Show recent Cortex actions with reasoning traces."""
    ...

@mcp.tool()
def cortex_approve(action_id: str) -> str:
    """Approve a pending Cortex action that requires human approval."""
    ...

@mcp.tool()
def cortex_reject(action_id: str, reason: str = "") -> str:
    """Reject a pending Cortex action."""
    ...

6. Implementation Phases

Phase 1: INFER (Weeks 1-3)

Verification:

python -m pytest tests/test_ir_inference.py -x -q
# AI-enhanced dead code POC
grafosa --graph graph.json ai-dead-code --model ollama/llama3

Phase 2: SENTINEL (Weeks 4-6)

Verification:

# Start sentinel watching maestro-platform
grafosa sentinel start --graph graph.json --watch .

# Edit a file, see graph update in <1s
echo "# test" >> symposium/src/conductor.py

# Check health stream
grafosa sentinel dashboard

Phase 3: CORTEX (Weeks 7-10)

Verification:

# Start full AI-embedded stack
grafosa cortex start --graph graph.json --watch .

# Introduce a bug, watch Cortex respond
echo "import nonexistent_module" >> symposium/src/new_file.py
# → Cortex detects file change (Sentinel)
# → Identifies dangling import (Intel)
# → Assesses significance (blast radius)
# → Recommends fix (Strategy)
# → Logs decision (Audit)

# Check what happened
grafosa cortex log

7. What This Enables

7.1 Immediate Capabilities (Post Phase 3)

7.2 Future Capabilities (Post Phase 3 Extensions)

8. Principles

1. AI is optional. Every INFER op has a rule-based fallback. GraFOSa works without any LLM API key. AI enhances; it doesn't gate.

2. Determinism first. Structural analysis (cycles, dead code, blast radius) is always deterministic. AI operations are clearly marked and auditable.

3. The graph is the truth. AI doesn't hallucinate graph structure. It reasons about the graph, using graph data as grounding context. The graph constrains the AI.

4. Feedback loops close. Every AI decision feeds back through AccuracyTracker. Bad decisions get corrected. The system improves over time.

5. Humans approve risky actions. Cortex can learn and annotate freely. Anything that modifies code requires explicit approval. The autonomy dial is configurable.

6. Cost-aware. Every INFER op has a token budget. Rule-based reasoning handles 80% of cases at zero cost. LLM is only invoked for ambiguous situations.

9. Estimated Effort

For context: the existing codebase is ~25,000+ lines built over ~2 weeks. This roadmap adds ~17% more code to achieve a qualitative leap in capability.

10. Decision Points for Review

1. INFER backend priority: Start with Anthropic (Claude) or Ollama (local/free)? Recommendation: Ollama first for rapid iteration, Anthropic for quality validation.

2. Sentinel scope: Full OTEL integration in Phase 2, or defer to Phase 3 extension? Recommendation: Defer OTEL; filesystem + git covers 90% of the value.

3. Cortex autonomy default: How autonomous out of the box? Recommendation: Conservative — learn + annotate auto-approved, everything else notify-only.

4. Naming: "Cortex" for the reasoning engine, "Sentinel" for the perception daemon. Alternatives considered: "Nexus"/"Overseer" (Cortex), "Watcher"/"Observer" (Sentinel).

5. Merge strategy: Should the Graph IR worktree be merged first as a prerequisite? Recommendation: Yes, it's the foundation for INFER ops.