Occam's razor

Vibe Coding to Production

2026-02-26T00:00:00+00:00

I’ve been using Claude Code and Cursor heavily for the past few months. The productivity gain is real - I built a working prototype of a monitoring system in an afternoon that would have taken me a week. But I also shipped a bug to production because the AI-generated code silently swallowed an exception. Lessons were learned.

Here’s what I’ve figured out about the gap between “vibe coded prototype” and “thing that runs in production without waking me up at 3am.”

How I Actually Vibe Code

The trick is telling the AI what you want to exist, not how to build it. When I say “create a Flask app with psycopg2 that queries PostgreSQL,” I’ve already made a bunch of decisions. When I say “I need an API for storing user profiles with CRUD operations,” the AI picks reasonable defaults and I can focus on whether the behavior is right.

When something breaks, I describe the problem instead of debugging it myself. “The create endpoint works but get-by-ID returns a 500, logs show connection already closed.” The AI has context about what it just wrote. It’ll usually spot the issue faster than I would and explain why it happened.

The trap is letting the AI make decisions you should be making. “Handle errors appropriately” gets you generic try/catch blocks. “When payment fails, retry 3 times with backoff, then mark the order failed and email the user” gets you code that does what you actually need.

I stop vibe coding when I notice I’m re-explaining the same context, or when fixes start introducing new bugs, or when I’m not sure what the code is doing anymore. At that point I have a working prototype and it’s time to actually engineer it.

What’s Usually Missing

After vibe coding a few projects to production, the gaps are predictable:

No tests. The AI optimizes for “does this look right” not “will this break.” I tested it manually, it worked, ship it. Then edge cases happen.

Optimistic error handling. Everything is wrapped in try/catch that logs and returns 500. No distinction between “you sent bad data” and “our database is down.” Users see “internal server error” for everything.

print() debugging left in. Or worse, no logging at all. Something breaks in production and I’m flying blind.

Hardcoded config. The AI puts in working values to make the example run. Database URLs, API keys, timeouts - all sitting in the code.

Security as an afterthought. “I’ll add auth later” is in the commit history of every prototype I’ve written.

The Fixes I Actually Apply

Tests, But Not Everywhere

I don’t aim for coverage percentages. I write tests for:

The happy path, end-to-end. Does the main flow actually work?

The business logic I wrote myself. The custom stuff, not the CRUD boilerplate.

The edge cases that have bitten me before. Empty strings. Nulls where I expected values. Concurrent requests to the same resource.

I ask the AI to generate tests, but I’m specific: “Test payment processing. Include: successful payment, insufficient funds, expired card, two concurrent payments for the same order.” Generic “write tests for this module” gets you generic tests.

The other thing: vibe-coded prototypes are often untestable because everything is coupled together. The class creates its own database connection in init. You can’t mock it. I’ve started asking the AI to “refactor this to use dependency injection so I can test it” before I write any tests.

Error Handling That Helps

The minimum viable improvement:

class ValidationError(Exception):
    """User sent bad data - 400"""
    pass

class NotFoundError(Exception):
    """Thing doesn't exist - 404"""
    pass

# Catch these specifically and return appropriate status codes
# Everything else is a 500, but log it properly

The thing that actually matters is separating “errors I should show to users” from “errors I should hide from users.” Validation failures, not found, conflict - users can see those. Database connection errors, null pointer exceptions, third-party API failures - users get a generic message, but I get the full stack trace in logs.

Logging That’s Actually Useful

I switched to structured logging after spending an hour grep-ing through text logs trying to trace a request. Now every log line is JSON with a request ID:

# Before
print(f"Processing order {order_id}")
logger.info("Payment complete")

# After
logger.info("payment_processed", order_id=order.id, amount=amount, method=method)

The structured version is searchable. I can query “show me all logs for request abc-123” or “show me all payment failures over $1000 in the last hour.” The print statements are write-only.

Request IDs are the key. Generate a UUID at the start of each request, attach it to every log line, return it in error responses. When a user reports “I got an error,” they give you the request ID and you can see exactly what happened.

Security Stuff I Keep Forgetting

Input validation. The AI is optimistic about inputs. It assumes request.json[“email”] exists and is a string. Production users will send you nulls, arrays, and SQL injection attempts. I use Pydantic now for everything that comes from outside:

class CreateUserRequest(BaseModel):
    email: EmailStr
    name: str = Field(min_length=1, max_length=100)

If it doesn’t validate, it throws before my code runs. No more checking for None everywhere.

Auth on every endpoint. I’ve shipped endpoints that forgot the auth decorator. Now I use a before_request hook that requires auth by default and I explicitly mark public endpoints:

PUBLIC_PATHS = ["/health", "/login"]

@app.before_request
def require_auth():
    if request.path in PUBLIC_PATHS:
        return
    # ... validate token

Secrets out of code. The AI puts in placeholder values to make examples work. I’ve committed API keys. Now I make secrets required from environment with no defaults - the app crashes on startup if they’re missing, which is better than running with test credentials.

Config

Hardcoded values are the silent killer. The prototype works with 5 second timeouts and 10 database connections. Production needs different numbers. I externalize anything that might vary:

DATABASE_URL = os.environ["DATABASE_URL"]  # Required, no default
REQUEST_TIMEOUT = int(os.environ.get("REQUEST_TIMEOUT", "30"))  # Optional with default
DEBUG = os.environ.get("DEBUG", "").lower() == "true"  # Off unless explicitly on

The required ones fail fast if missing. The optional ones have sensible defaults. Debug is never on by accident.

Making It Deployable

Health check endpoints. Kubernetes (or whatever) needs to know if your app is alive:

@app.get("/health")
def health():
    return {"status": "ok"}

@app.get("/ready")
def ready():
    # Check database, redis, whatever
    # Return 503 if something's down

Graceful shutdown. When the container gets SIGTERM, finish in-flight requests before dying. The AI never adds this:

import signal

def shutdown(signum, frame):
    # Stop accepting new requests
    # Wait for current requests to finish
    # Then exit

signal.signal(signal.SIGTERM, shutdown)

The Actual Checklist

Before I deploy something I vibe coded:

Can I trace a request through the logs? Do errors tell me what went wrong? Is there anything hardcoded that shouldn’t be? Did I test the main flow and the obvious edge cases? Is auth required everywhere it should be? Will it start up correctly in production, or crash because of missing config?

That’s it. It’s not comprehensive, but it catches the stuff that actually wakes me up at night.

The Point

Vibe coding is great for getting something working. It’s not great for keeping it working. The gap is predictable - tests, errors, logging, security, config. Once you know what to look for, it takes maybe an hour to harden a prototype. That hour saves you the 3am pages.

The engineers who get the most out of these tools are the ones who know what production code looks like. The AI gets you 80% of the way there fast. The last 20% is still on you.

Building a Computer-Use Agent in Python - Part 2: Multi-Agent Architecture

2026-01-18T00:00:00+00:00

In Part 1, we built a single GUI agent that could control a computer through screenshots and mouse/keyboard actions. That works for simple tasks, but complex workflows benefit from specialization. A pixel-clicking agent is inefficient for web automation (fragile coordinates vs. DOM selectors), and high-level task planning differs from low-level action generation.

This post covers the multi-agent architecture in kyros: a BossAgent that orchestrates, specialized worker agents for different domains, and how they coordinate through a hierarchical delegation pattern.

Why Multi-Agent?

Single agents hit practical limits:

Prompt complexity: One prompt can’t effectively cover task planning, GUI control, browser automation, and shell commands. Each domain has different tools, context, and reasoning patterns.
Context efficiency: A browser agent doesn’t need window lists, and a GUI agent doesn’t need DOM trees. Splitting agents means each gets relevant context.
Model selection: Planning benefits from stronger reasoning models. Action generation works fine with smaller, faster models. Multi-agent lets you mix models.

The Boss-Worker Pattern

The architecture is hierarchical:

User
  │
  ▼
BossAgent (orchestrator)
  │
  ├── GUIAgent (desktop control)
  ├── BrowserActionAgent (web automation)
  ├── ShellAgent (command execution)
  └── ResearchAgent (web search)

The BossAgent receives tasks from the user, creates plans, and delegates subtasks to specialized agents. It doesn’t execute actions directly - it coordinates.

BossAgent: The Orchestrator

The BossAgent’s job is to understand what the user wants, break it into subtasks, and route them to the right worker:

class BossAgent(BaseAgent):
    def __init__(self, ...):
        super().__init__(agent_name="boss", ...)
        self.subagents: Dict[str, Any] = {}

    def get_system_prompt(self) -> str:
        return """# Identity

You are a computer-use agent. You coordinate tasks by delegating to specialized agents:

## Available Agents

1. **BrowserBossAgent**: Browser automation and web interactions
   - Use for: navigating websites, filling forms, clicking web elements

2. **GUIAgent**: Mouse and keyboard interactions with GUI
   - Use for: clicking, typing, hotkeys, scrolling

3. **ShellAgent**: Executes shell commands
   - Use for: running terminal commands, file operations

4. **ResearchAgent**: Researches information using Tavily
   - Use for: searching the web, gathering information

## Your Responsibilities

1. Analyze the user's request and the current screenshot
2. For multi-step tasks: create a plan and get user approval first
3. Delegate manageable subtasks to sub-agents
4. Handle agent responses and coordinate multi-step workflows
5. Report results back to the user

## Response Format

{
  "thought": "Your reasoning about what to do next",
  "action": {
    "type": "delegate",
    "agent": "BrowserBossAgent",
    "message": "Open https://example.com and fill the login form"
  }
}
"""

The BossAgent outputs structured JSON with a thought process and an action. Actions can be:

delegate: Send subtask to a worker agent
message: Communicate with the user (often to present a plan)
exit: Task complete, report results

Agent Creation and Reuse

Workers are created on-demand and reused across delegations:

def get_or_create_agent(self, agent_type: str) -> Any:
    """Get existing agent or create a new one"""
    if agent_type in self.subagents:
        return self.subagents[agent_type]

    if agent_type == "BrowserActionAgent":
        from agents.browser_action_agent import BrowserActionAgent
        agent = BrowserActionAgent(
            websocket_callback=self.websocket_callback,
            config_dict=self.config_dict
        )
    elif agent_type == "GUIAgent":
        from agents.gui_agent import GUIAgent
        agent = GUIAgent(
            websocket_callback=self.websocket_callback,
            config_dict=self.config_dict
        )
    # ... other agent types

    self.subagents[agent_type] = agent
    return agent

This is important for the BrowserActionAgent - it keeps the browser instance alive across multiple delegations, so we don’t keep opening and closing browsers.

Browser Automation with Playwright

The GUIAgent clicks pixels, which is fragile for web automation. The BrowserActionAgent uses Playwright for proper DOM interaction.

Setup

from playwright.async_api import async_playwright, Browser, BrowserContext, Page

class BrowserActionAgent(BaseAgent):
    def __init__(self, ...):
        super().__init__(agent_name="browser", ...)
        self.playwright = None
        self.browser: Optional[Browser] = None
        self.context: Optional[BrowserContext] = None
        self.page: Optional[Page] = None

    async def _launch(self, name: str = "chromium") -> Dict[str, Any]:
        """Launch a new browser"""
        if self.playwright is None:
            self.playwright = await async_playwright().start()

        self.browser = await self.playwright.chromium.launch(
            headless=False,
            args=[
                '--no-sandbox',
                '--disable-setuid-sandbox',
                '--disable-dev-shm-usage'
            ]
        )
        self.context = await self.browser.new_context(no_viewport=True)
        self.page = await self.context.new_page()

        return {"success": True, "message": "Browser launched"}

XPath-Based Actions

The agent uses XPath selectors rather than coordinates. This is more reliable because elements can be identified by their semantic structure:

async def _click(self, xpath: str, button: str = "left", click_count: int = 1) -> Dict[str, Any]:
    """Click on an element by XPath"""
    if not self.page:
        return {"success": False, "error": "No active browser page"}

    element = await self.page.query_selector(f"xpath={xpath}")
    if not element:
        return {"success": False, "error": f"No element found: {xpath}"}

    await element.click(button=button, click_count=click_count)
    return {"success": True, "message": f"Clicked: {xpath}"}

async def _fill(self, xpath: str, text: str) -> Dict[str, Any]:
    """Fill an element with text"""
    element = await self.page.query_selector(f"xpath={xpath}")
    if not element:
        return {"success": False, "error": f"No element found: {xpath}"}

    await element.fill(text)
    return {"success": True, "message": f"Filled: {xpath}"}

The LLM generates XPath selectors based on the page structure. For example:

//input[@name='email'] - input with name attribute
//button[contains(text(), 'Submit')] - button containing text
//div[@class='search-box']//input - input inside a div with class

System Prompt

def get_system_prompt(self) -> str:
    return """# Identity

You are a Browser Action Agent that automates browser interactions using Playwright.

# Available Tools

- launch(name): Launch browser ("chromium" or "firefox")
- navigate(url): Navigate to URL
- click(xpath): Click element by XPath
- fill(xpath, text): Fill element with text
- input_text(xpath, text): Type text character by character
- press_key(xpath, key): Press a key ("Enter", "Tab", "Escape")
- hover(xpath): Hover over element
- get_text(xpath): Get text content
- wait_for_element(xpath, timeout): Wait for element to appear
- scroll_into_view(xpath): Scroll element into view
- wait(seconds): Wait
- exit(summary, exitCode): Exit when finished

# Response Format

{
  "thought": "Your reasoning about what to do next",
  "action": {
    "tool": "click",
    "args": {"xpath": "//button[@id='submit']"}
  }
}
"""

Action Dispatch

The agent has a dispatch method that routes tool calls to implementations:

async def execute_action(self, action: Dict[str, Any]) -> Dict[str, Any]:
    """Execute a browser action"""
    tool = action.get("tool")
    args = action.get("args", {})

    if tool == "launch":
        return await self._launch(**args)
    elif tool == "navigate":
        return await self._navigate(**args)
    elif tool == "click":
        return await self._click(**args)
    elif tool == "fill":
        return await self._fill(**args)
    elif tool == "press_key":
        return await self._press_key(**args)
    elif tool == "wait_for_element":
        return await self._wait_for_element(**args)
    elif tool == "exit":
        return self._exit(**args)
    else:
        return {"success": False, "error": f"Unknown tool: {tool}"}

Context Management

Agents need context to make good decisions, but context has costs (tokens, latency). The system manages this through context passing and compaction.

Context Passing Between Agents

When the BossAgent delegates to a worker, it passes relevant context in the message:

# BossAgent builds context for worker
context_parts = []
if self.compacted_context:
    context_parts.append(f"Previous Context:\n{self.compacted_context}")
context_parts.append(f"User Request: {message.get('content', '')}")

# Include agent responses from previous steps
for agent_resp in message.get('agent_responses', []):
    agent_type = agent_resp.get("agent_type")
    response = agent_resp.get("response", {})
    context_parts.append(f"{agent_type} response: {response}")

Context Compaction

Both BossAgent and worker agents compact their context when it gets too large:

# Check if compaction is needed
context_text = str(self.history)
word_count = count_words(context_text)

if self.step_count >= trigger_steps or word_count >= trigger_words:
    self.compacted_context = compact_context(
        self.history,
        task,
        self.config_dict,
        self.websocket_callback
    )
    self.history = []  # Clear after compaction
    self.step_count = 0

The compact_context function uses a fast LLM to summarize the action history, preserving key information while reducing token count.

Exit Summaries

When a worker agent finishes, it returns an exit summary that captures what was accomplished:

def _exit(self, summary: str = None, exitCode: int = 0) -> Dict[str, Any]:
    """Exit the agent"""
    return {
        "success": True,
        "exit": True,
        "summary": summary or "Agent completed",
        "exitCode": exitCode
    }

This summary gets passed back to the BossAgent, which uses it as context for the next step. For example:

BrowserActionAgent response: "Filled login form with email user@example.com
and password, clicked Submit button. Login successful - now on dashboard page."

This gives the BossAgent enough context to decide the next action without re-analyzing the full history.

Full Orchestration

Here’s how a multi-step task flows through the system:

User: "Log into example.com and download my invoice"

1. BossAgent receives task, takes screenshot
2. BossAgent creates plan, asks user for approval:
   "I'll help with that. Plan:
    1. Open browser and navigate to example.com
    2. Fill login form with your credentials
    3. Navigate to invoices section
    4. Download the latest invoice
    Does this look good?"

3. User approves

4. BossAgent delegates to BrowserActionAgent:
   "Open browser and navigate to example.com/login"

5. BrowserActionAgent executes:
   - launch(name="chromium")
   - navigate(url="https://example.com/login")
   - exit(summary="Browser launched, navigated to login page")

6. BossAgent receives result, delegates next step:
   "Fill login form with email user@example.com, password ****"

7. BrowserActionAgent executes:
   - fill(xpath="//input[@name='email']", text="user@example.com")
   - fill(xpath="//input[@name='password']", text="****")
   - click(xpath="//button[@type='submit']")
   - wait_for_element(xpath="//div[@class='dashboard']")
   - exit(summary="Logged in successfully, on dashboard")

8. ... continues until task complete

9. BossAgent reports to user:
   "Downloaded invoice_2024_01.pdf to ~/Downloads"

Real-Time Updates

The system uses WebSocket callbacks to send real-time updates to the frontend:

def send_llm_update(self, event_type: str, data: Dict[str, Any]):
    """Send update via WebSocket"""
    if self.websocket_callback:
        self.websocket_callback({
            "type": event_type,
            "agent_id": self.agent_id,
            "agent_type": self.__class__.__name__,
            "data": data
        })

Events include:

screenshot: New screenshot captured
llm_call_start: LLM request started
llm_content_chunk: Streaming response token
action_execute: About to execute action
action_result: Action completed

Tradeoffs and Extensions

What Works

Specialization pays off: Browser agent with Playwright is much more reliable than pixel-clicking for web tasks.
Hierarchical delegation: BossAgent can focus on planning while workers handle execution details.
Context compaction: Keeps long sessions from hitting token limits.

What’s Tricky

Coordination overhead: More agents means more LLM calls. A simple task might be faster with a single agent.
Error recovery: When a worker fails, the BossAgent needs to decide whether to retry, try a different approach, or give up. This is hard to get right.
State sharing: The browser persists across delegations, but the BossAgent needs to track what state it’s in (logged in? which page?).

Extensions

Memory: Persistent memory across sessions (what credentials work, common workflows).
Self-improvement: Agent analyzes its failures and updates its prompts or tool usage.
Human-in-the-loop: Ask for help when stuck rather than failing silently.
Tool creation: Agent writes new tools when it needs capabilities it doesn’t have.

The multi-agent pattern is a good foundation for these extensions because each capability can be added to the appropriate agent without affecting others.

Building a Computer-Use Agent in Python - Part 1: The Agent Loop

2025-12-14T00:00:00+00:00

Computer-use agents are AI systems that can control a computer the way a human would - through mouse clicks, keyboard input, and visual observation. Unlike API-based automation, these agents work with any application that has a GUI, making them surprisingly general-purpose.

I built one called kyros as a side project. This post covers the core architecture: how the agent observes the screen, decides what to do, executes actions, and verifies results. Part 2 will cover multi-agent coordination.

The Core Loop

Every computer-use agent follows the same fundamental pattern:

while task_not_complete:
Observe  - Take a screenshot, get window list
Think   - Send to LLM, get next action
Act     - Execute the action (click, type, etc.)
Verify  - Compare before/after screenshots

The key insight is that this loop is reactive - the agent doesn’t plan out every step in advance. It observes the current state, takes one action, observes the result, and decides the next action. This makes it robust to unexpected UI states.

Here’s the simplified loop from the GUI agent:

async def process_message(self, message: Dict[str, Any]) -> Dict[str, Any]:
    task = message.get("content", "")
    self.running = True
    history = []

    while self.running and iteration < max_iterations:
        # 1. Observe
        screenshot = self.get_screenshot_base64()
        active_windows = self.get_active_windows()

        # 2. Think - Build context and call LLM
        messages = self.build_context(task, screenshot, active_windows, history)
        action_code = self.generate_action(messages, system=self.get_system_prompt())

        # 3. Act
        screenshot_before = screenshot
        exec_result = self.execute_action(action_code)

        if not self.running:  # Exit action was called
            break

        # 4. Verify
        time.sleep(0.5)  # Wait for UI to update
        screenshot_after = self.get_screenshot_base64()
        verification = await self.verify_action(screenshot_before, screenshot_after, action_code)

        history.append({
            "action": action_code,
            "result": exec_result,
            "verification": verification
        })

System Control Tools

The agent needs low-level control over mouse, keyboard, and screen capture. On Linux, I use Xlib for mouse control (it’s more reliable than pyautogui for X11), pyautogui for keyboard, and ImageMagick’s import command for screenshots.

Mouse Control with Xlib

Xlib gives direct access to the X11 server. The trick is using relative coordinates (0-1 range) so the agent doesn’t need to know screen resolution:

import Xlib.display
import Xlib.X
import Xlib.ext.xtest

def click(x: float, y: float, button: int = 1, clicks: int = 1) -> dict:
    """Click at relative coordinates (0-1 range)"""
    display = Xlib.display.Display(os.environ.get('DISPLAY', ':0'))
    screen = display.screen()
    width = screen.width_in_pixels
    height = screen.height_in_pixels

    # Convert relative to absolute coordinates
    abs_x = int(x * width)
    abs_y = int(y * height)

    # Move mouse
    root = screen.root
    root.warp_pointer(abs_x, abs_y)
    display.sync()

    # Perform clicks using XTEST extension
    for _ in range(clicks):
        Xlib.ext.xtest.fake_input(display, Xlib.X.ButtonPress, button)
        display.sync()
        Xlib.ext.xtest.fake_input(display, Xlib.X.ButtonRelease, button)
        display.sync()
        if clicks > 1:
            time.sleep(0.05)  # Small delay between multiple clicks

    display.close()
    return {"stdout": "", "stderr": "", "exitCode": 0}

The XTEST extension is key - it injects synthetic input events that applications can’t distinguish from real input.

Keyboard with pyautogui

For keyboard input, pyautogui works well enough:

import pyautogui

def type(text: str) -> dict:
    """Type text character by character"""
    pyautogui.write(text, interval=0.01)
    return {"stdout": "", "stderr": "", "exitCode": 0}

def hotkey(keys: str) -> dict:
    """Execute hotkey combination. Example: 'super+r' or 'ctrl+alt+t'"""
    key_parts = keys.split('+')

    key_map = {
        'super': 'winleft',
        'ctrl': 'ctrl',
        'alt': 'alt',
        'shift': 'shift'
    }

    mapped_keys = [key_map.get(k.lower(), k.lower()) for k in key_parts]
    pyautogui.hotkey(*mapped_keys)
    return {"stdout": "", "stderr": "", "exitCode": 0}

Screenshot Capture

For screenshots, I use ImageMagick’s import command which captures the X11 root window:

def get_screenshot_base64(self) -> str:
    """Capture screenshot and return as base64-encoded JPEG"""
    temp_fd, temp_path = tempfile.mkstemp(suffix='.png')
    os.close(temp_fd)

    try:
        env = os.environ.copy()
        env['DISPLAY'] = env.get('DISPLAY', ':0')

        subprocess.run(
            ["import", "-window", "root", temp_path],
            capture_output=True,
            timeout=2,
            env=env
        )

        # Convert to JPEG for smaller payload
        screenshot = Image.open(temp_path)
        buffer = BytesIO()
        screenshot.save(buffer, format="JPEG", quality=75)
        buffer.seek(0)

        img_base64 = base64.b64encode(buffer.read()).decode('utf-8')
        return f"data:image/jpeg;base64,{img_base64}"
    finally:
        if os.path.exists(temp_path):
            os.unlink(temp_path)

I also overlay the cursor position onto the screenshot so the LLM knows where the mouse currently is:

# Get cursor position
display = Xlib.display.Display(':0')
root = display.screen().root
pointer = root.query_pointer()
cursor_x, cursor_y = pointer.root_x, pointer.root_y

# Overlay cursor image
cursor_img = Image.open('./cursor.png')
screenshot.paste(cursor_img, (cursor_x, cursor_y), cursor_img)

Window Management

The agent needs to know what windows are open and be able to focus them. wmctrl is perfect for this:

def get_active_windows(self) -> str:
    """Get list of active windows"""
    result = subprocess.run(
        ["wmctrl", "-l"],
        capture_output=True,
        text=True,
        timeout=2
    )
    return result.stdout  # Format: window_id desktop host title

def focus_window(window_id: str) -> dict:
    """Focus a window by its ID"""
    subprocess.run(["wmctrl", "-i", "-a", window_id], timeout=2)

    # Move mouse to center of window
    geom_result = subprocess.run(["wmctrl", "-l", "-G"], capture_output=True, text=True)
    # Parse geometry and move mouse...

The Base Agent Class

All agents inherit from a base class that handles LLM communication:

class BaseAgent(ABC):
    def __init__(
        self,
        agent_id: str = None,
        api_key: str = None,
        model: str = None,
        websocket_callback: Optional[Callable] = None,
        agent_name: str = None,
        config_dict: Dict[str, Any] = None
    ):
        self.agent_id = agent_id or str(uuid.uuid4())
        self.agent_name = agent_name

        # Load config for this agent type
        agent_config = config.get_agent_config(agent_name, config_dict)
        self.api_key = api_key or agent_config.get("api_key")
        self.model = model or agent_config.get("model")
        self.api_provider = agent_config.get("api_provider", "openai")

        # Initialize client based on provider
        if self.api_provider == "anthropic":
            self.client = Anthropic(api_key=self.api_key)
        else:
            self.client = OpenAI(api_key=self.api_key, base_url=self.base_url)

    @abstractmethod
    def get_system_prompt(self) -> str:
        pass

    @abstractmethod
    async def process_message(self, message: Dict[str, Any]) -> Dict[str, Any]:
        pass

The call_llm method handles both OpenAI and Anthropic APIs, with streaming support:

def call_llm(
    self,
    messages: List[Dict[str, Any]],
    system: str = None,
    temperature: float = None,
    max_tokens: int = None,
    stream: bool = True
) -> str:
    if self.api_provider == "anthropic":
        anthropic_messages = self._convert_to_anthropic_format(messages)
        with self.client.messages.stream(
            model=self.model,
            messages=anthropic_messages,
            system=system or "",
            temperature=temperature,
            max_tokens=max_tokens
        ) as response:
            response_text = ""
            for text in response.text_stream:
                response_text += text
                self.send_llm_update("llm_content_chunk", {"content": text})
    else:
        # OpenAI streaming...
        pass

    return response_text

The GUI Agent

The GUI agent is the workhorse - it takes a task, observes the screen, and generates Python code to execute actions.

System Prompt

The system prompt defines what tools are available and how to use them:

def get_system_prompt(self) -> str:
    return """
# Identity

You are a GUI Agent. Your job is to analyze the given screenshot and execute
the given TASK by performing step-by-step actions.

# Tools

- tools.focus_window(window_id): Focus a window by its ID
- tools.move(x, y): Move mouse to relative coordinates (0-1 range)
- tools.click(x, y, button=1, clicks=1): Click at relative coordinates
- tools.scroll(amount): Scroll (positive=down, negative=up)
- tools.type(text): Type the given text
- tools.hotkey(keys): Press a hotkey combination (e.g., 'super+r', 'ctrl+c')
- tools.wait(n): Wait for n seconds
- tools.exit(summary, exitCode): Exit when finished

# Rules

- Respond with executable Python code that calls ONE of these tools
- Only generate 1 action at a time
- Don't repeat the same action again and again
- Look at the "Currently active windows" list to determine which window to focus

# Example

```python
# Focus the Firefox browser window
tools.focus_window("0x02400003")

”””

<br>
### Action Execution

The agent generates Python code, which we execute in a controlled namespace:

```python
def execute_action(self, action_code: str) -> dict:
    """Execute the generated Python code"""
    # Extract code from markdown blocks
    if "```python" in action_code:
        code = action_code.split("```python")[1].split("```")[0].strip()

    result = {"stdout": "", "stderr": "", "exitCode": 0}

    try:
        namespace = {"tools": tools, "__action_result__": None}

        # Capture tool call results
        lines = code.split('\n')
        modified_lines = []
        for line in lines:
            stripped = line.strip()
            if stripped.startswith('tools.') and not stripped.startswith('tools.exit('):
                indent = len(line) - len(line.lstrip())
                modified_lines.append(' ' * indent + f'__action_result__ = {stripped}')
            else:
                modified_lines.append(line)

        exec('\n'.join(modified_lines), namespace)

        if namespace.get("__action_result__"):
            tool_result = namespace["__action_result__"]
            result["stdout"] = tool_result.get("stdout", "")
            result["stderr"] = tool_result.get("stderr", "")
            result["exitCode"] = tool_result.get("exitCode", 0)

    except tools.ExitException as e:
        self.running = False
        result["stdout"] = e.message
        result["exitCode"] = e.exit_code

    return result

Verification Loop

After each action, the agent compares before/after screenshots to verify success:

async def verify_action(self, screenshot_before: str, screenshot_after: str, action: str) -> str:
    """Verify if action succeeded by comparing screenshots"""
    messages = [
        {
            "role": "user",
            "content": [
                {"type": "text", "text": f"Action performed: {action}\n\nDid the action succeed?"},
                {"type": "image_url", "image_url": {"url": screenshot_before}},
                {"type": "text", "text": "Screenshot after:"},
                {"type": "image_url", "image_url": {"url": screenshot_after}}
            ]
        }
    ]

    verification = self.call_llm(
        messages=messages,
        system="Compare two screenshots and determine if the action succeeded. Be concise.",
        max_tokens=500
    )
    return verification

This verification feeds back into the agent’s context for the next iteration, helping it understand what worked and what didn’t.

Context Compaction

As the agent takes more actions, the context grows. To prevent hitting token limits, we periodically compact the history:

# Check if compaction is needed
if self.step_count >= trigger_steps or word_count >= trigger_words:
    self.compacted_context = compact_context(
        history,
        task,
        self.config_dict,
        self.websocket_callback
    )
    history = []  # Clear history after compaction
    self.step_count = 0

The compaction uses a smaller/faster LLM to summarize the action history into a condensed form that preserves the essential information.

Running It

Here’s what a simple task execution looks like:

from agents.gui_agent import GUIAgent

agent = GUIAgent(config_dict=config)
result = await agent.process_message({
    "content": "Open Firefox and navigate to google.com",
    "max_iterations": 20
})

The agent might produce actions like:

# Step 1: Press Super key to open app launcher
tools.hotkey("super")

# Step 2: Type "firefox" to search for the app
tools.type("firefox")

# Step 3: Press Enter to launch Firefox
tools.hotkey("enter")

# Step 4: Wait for Firefox to open
tools.wait(2)

# Step 5: Click on the URL bar
tools.click(0.5, 0.05)

# Step 6: Type the URL
tools.type("google.com")

# Step 7: Press Enter to navigate
tools.hotkey("enter")

# Step 8: Task complete
tools.exit(summary="Opened Firefox and navigated to google.com", exitCode=0)

Next Steps

This single-agent architecture works, but it has limitations. The GUI agent is good at pixel-level interactions but not at high-level reasoning about complex tasks. It also can’t efficiently handle browser automation (clicking by coordinates is fragile compared to DOM selectors).

In Part 2, we’ll look at the multi-agent architecture: a BossAgent that breaks down tasks, a BrowserActionAgent that uses Playwright for web automation, and how they coordinate to handle complex workflows.

Fine-Tuning Qwen3 0.6B for Chess Move Legality

2025-10-25T00:00:00+00:00

Can a 600M parameter language model learn to play chess? Not quite - but it can learn the rules. I fine-tuned Qwen3-0.6B on 776k chess positions to predict whether a move is legal. The goal isn’t to build a chess engine (Stockfish exists), but to teach a small LLM to reason about board states and constraints.

This is a stepping stone toward building models that can generate best moves with chain-of-thought reasoning. First teach the rules, then teach strategy.

Why Chess Move Legality?

Chess engines like Stockfish use tree search and hand-coded evaluation functions. They’re superhuman but not general-purpose - you can’t ask Stockfish to explain its reasoning or apply chess concepts to analogous problems.

Language models, on the other hand, can learn patterns from data and potentially generate human-like reasoning. But they need to understand the fundamentals first. Move legality is that foundation:

Simpler than move generation: “Is this move legal?” is easier than “What’s the best move?”
Well-defined: Legal moves follow strict rules, no ambiguity
Rich in structure: Pieces have constraints, the board has geometry, special moves have conditions

If a model can’t learn move legality, it won’t learn strategy.

The Dataset

Each training example has three components:

1. Board Position - A text description of piece placement:

a8:black rook
b8:black knight
c8:black bishop
d8:black queen
e8:black king
...
e2:white pawn
f1:white king

2. Query - A question about move legality:

Is it legal for the white bishop at c4 to move to (or capture) f7?
Answer only yes or no

3. Output - The answer, optionally with reasoning:

<think>
The white bishop on c4 can move diagonally. f7 is on the diagonal from c4.
There's a black pawn on f7, so this is a capture. The move is legal.
</think>

yes

The <think> tags are for chain-of-thought training. Not all examples have it - some just answer “yes” or “no” directly.

Data Composition

The 776k examples cover:

Legal vs illegal moves: ~50/50 split
All piece types: pawns, knights, bishops, rooks, queens, kings
Special moves: castling, en passant, pawn promotion
Edge cases: pinned pieces, discovered checks, blocked paths
Captures vs quiet moves: both attacking and non-attacking moves

I generated this using python-chess, starting with real game positions from the Lichess database and synthetically creating move queries (both legal and illegal).

Training Setup

Base Model: Qwen3-0.6B - a 600M parameter causal language model. Small enough to train on consumer hardware, large enough to learn patterns.

Framework: TRL (Transformer Reinforcement Learning) for supervised fine-tuning, Unsloth for memory optimization.

Hardware: AMD RX 7900 XT (24GB VRAM). Full fine-tuning uses ~20GB, LoRA uses ~8GB.

Hyperparameters

BATCH_SIZE = 64
GRADIENT_ACCUMULATION_STEPS = 2  # Effective batch = 128
LEARNING_RATE = 2e-4
NUM_EPOCHS = 3
MAX_SEQ_LENGTH = 1024
WEIGHT_DECAY = 0.001
OPTIMIZER = "paged_adamw_8bit"
PRECISION = "bfloat16"

I used full parameter fine-tuning, not LoRA. The model is small enough that full tuning fits in VRAM and trains in reasonable time (~6 hours for 3 epochs).

Chat Template

Qwen3 uses a chat template for instruction following:

<|im_start|>user
Consider the position below and answer the query:

[board position]

Query: Is it legal for the white knight at f3 to move to g5?
Answer only yes or no
<|im_end|>
<|im_start|>assistant
yes
<|im_end|>

The tokenizer handles this formatting via tokenizer.apply_chat_template(), which wraps the prompt and output in the correct structure.

Code

The training loop is straightforward with TRL’s SFTTrainer:

from trl import SFTTrainer, SFTConfig
from transformers import AutoTokenizer, AutoModelForCausalLM
from datasets import load_dataset

# Load model and tokenizer
model = AutoModelForCausalLM.from_pretrained(
    "unsloth/Qwen3-0.6B",
    torch_dtype=torch.bfloat16,
    device_map="auto"
)
tokenizer = AutoTokenizer.from_pretrained("unsloth/Qwen3-0.6B")

# Load dataset
dataset = load_dataset("json", data_files="data.jsonl", split="train")

# Format into chat template
def format_example(example):
    messages = [
        {"role": "user", "content": example["prompt"]},
        {"role": "assistant", "content": example["output"]}
    ]
    return {"text": tokenizer.apply_chat_template(messages, tokenize=False)}

dataset = dataset.map(format_example)

# Train
trainer = SFTTrainer(
    model=model,
    tokenizer=tokenizer,
    train_dataset=dataset,
    args=SFTConfig(
        output_dir="./output",
        num_train_epochs=3,
        per_device_train_batch_size=64,
        gradient_accumulation_steps=2,
        learning_rate=2e-4,
        bf16=True,
        logging_steps=10,
        save_strategy="steps",
        save_steps=500,
    ),
    dataset_text_field="text",
    max_seq_length=1024,
)

trainer.train()

The key is dataset_text_field="text" - SFTTrainer expects the formatted text in a field called “text”.

Training Observations

Loss curve: Dropped quickly in the first epoch (from 1.2 to 0.3), then slowly decreased in epochs 2-3. By epoch 3, loss was ~0.15.

Overfitting: Minimal. The task is large enough (776k examples) and diverse enough that the model didn’t memorize. Validation loss tracked training loss closely.

Speed: With Unsloth optimizations, training ran at ~120 examples/second on the RX 7900 XT. Full 3-epoch run took about 6 hours.

Memory: Full fine-tuning used 19GB VRAM. LoRA would have used ~7GB but I wanted to avoid the merge step later.

Inference

Using the trained model:

from transformers import AutoTokenizer, AutoModelForCausalLM

model = AutoModelForCausalLM.from_pretrained(
    "./output/chess-sft-qwen3-0.6b",
    torch_dtype="auto",
    device_map="auto"
)
tokenizer = AutoTokenizer.from_pretrained("./output/chess-sft-qwen3-0.6b")

# Format a chess position question
position = """a8:black rook
b8:black knight
...
[full board]
"""

prompt = f"""Consider the position below and answer the query:

{position}

Query: Is it legal for the white bishop at c4 to move to f7?
Answer only yes or no"""

messages = [{"role": "user", "content": prompt}]
text = tokenizer.apply_chat_template(messages, tokenize=False, add_generation_prompt=True)

inputs = tokenizer.encode(text, return_tensors="pt").to(model.device)
outputs = model.generate(inputs, max_new_tokens=50, temperature=0.1)
response = tokenizer.decode(outputs[0], skip_special_tokens=True)

print(response)
# => "yes"

The model responds with just “yes” or “no” most of the time. Occasionally it includes reasoning in <think> tags if the position is complex.

Evaluation

I held out 10k examples for testing. The model achieved:

Exact match accuracy: ~85%
Token overlap: ~91% (counts partial matches)

The 15% error rate breaks down as:

10%: Wrong answer (said “yes” when illegal, or vice versa)
5%: Formatting issues (extra text, didn’t follow template)

The common failure modes:

En passant: The model struggles with this special case
Castling through check: Sometimes allows illegal castling
Pinned pieces: Occasionally misses pins that prevent movement

These are all edge cases. For basic moves (pawn pushes, knight jumps, bishop diagonals), accuracy is >95%.

What This Enables

This model is a foundation for:

1. Best move generation: Fine-tune on (position, best_move) pairs. The model already understands legal moves, now teach it which are good.

2. Chain-of-thought reasoning: The <think> examples teach the model to explain its reasoning. Scaling this up could produce models that explain why a move is best.

3. RLHF/DPO: Use this as a reward model for reinforcement learning. The model can score move quality, which can train a policy model.

4. Multimodal chess: Extend to take board images as input instead of text. Vision models + this text understanding = models that “see” chess positions.

Limitations

Not a chess engine: This model verifies legality, it doesn’t play well. It has no concept of strategy, tactics, or winning positions. It’s like teaching someone the rules without teaching them how to win.

Small model: At 600M parameters, there’s limited capacity for complex reasoning. Scaling to 1.5B or 7B would likely improve both accuracy and reasoning quality.

No position evaluation: The model can’t tell you if a position is winning or losing. It only knows legal vs illegal.

Inference speed: Generating text is slower than Stockfish’s tree search. For move generation in real games, you’d need batching or faster inference.

Next Steps

I’m planning to:

Train on best move prediction with reasoning (SFT on annotated games)
Apply DPO using engine evaluations as preference labels
Scale to larger models (1.5B, 7B) to see if reasoning improves
Experiment with vision inputs (board images → moves)

The trained model is at chess-sft-qwen3-0.6b on Hugging Face. Training code is at github - it’s a simple TRL script that anyone can run on consumer hardware.

The Digital Immune System: Building an AI-Powered Security Monitor in Go

2025-06-19T00:00:00+00:00

Your immune system doesn’t wait for you to notice you’re sick. It continuously monitors for threats, identifies anomalies, and responds automatically. What if your servers worked the same way?

This post builds a “digital immune system” in Go - a monitor that watches system activity, uses local AI to detect anomalies, and automatically remediates threats. No cloud APIs, no security vendor lock-in, just a binary that runs on your servers.

Architecture

The system has three layers:

┌─────────────────────────────────────────────────────┐
│                    Collectors                        │
│  (processes, network, filesystem, auth logs)        │
└──────────────────────┬──────────────────────────────┘
                       │ events
                       ▼
┌─────────────────────────────────────────────────────┐
│              Anomaly Detection (ML)                  │
│  (statistical models, fast scoring, filtering)      │
└──────────────────────┬──────────────────────────────┘
                       │ anomalies
                       ▼
┌─────────────────────────────────────────────────────┐
│              AI Reasoning (Ollama)                   │
│  (threat analysis, remediation planning)            │
└──────────────────────┬──────────────────────────────┘
                       │ actions
                       ▼
┌─────────────────────────────────────────────────────┐
│                  Remediation                         │
│  (kill process, block IP, restore file, alert)      │
└─────────────────────────────────────────────────────┘

The ML layer does fast, cheap filtering. Only anomalies that score above a threshold get sent to the LLM for deeper analysis. This keeps latency low and prevents the LLM from being overwhelmed.

Data Structures

First, define the core types:

package main

import (
    "time"
)

type EventType string

const (
    EventProcess    EventType = "process"
    EventNetwork    EventType = "network"
    EventFilesystem EventType = "filesystem"
    EventAuth       EventType = "auth"
)

type Event struct {
    Type      EventType         `json:"type"`
    Timestamp time.Time         `json:"timestamp"`
    Data      map[string]any    `json:"data"`
    Source    string            `json:"source"`
}

type Anomaly struct {
    Event       Event     `json:"event"`
    Score       float64   `json:"score"`       // 0-1, higher = more anomalous
    Reason      string    `json:"reason"`      // ML explanation
    Threat      string    `json:"threat"`      // LLM threat assessment
    Remediation string    `json:"remediation"` // LLM suggested action
}

type Action struct {
    Type    string         `json:"type"`    // kill, block, restore, alert
    Target  string         `json:"target"`  // PID, IP, filepath
    Params  map[string]any `json:"params"`
    DryRun  bool           `json:"dry_run"`
}

Collectors

Each collector watches a different aspect of the system and emits events.

Process Collector

Monitors for suspicious processes - unusual binaries, crypto miners, reverse shells:

package collectors

import (
    "bufio"
    "fmt"
    "os"
    "path/filepath"
    "strconv"
    "strings"
    "time"
)

type ProcessInfo struct {
    PID        int
    Name       string
    Cmdline    string
    User       string
    CPU        float64
    Memory     float64
    StartTime  time.Time
    ParentPID  int
    Executable string
}

type ProcessCollector struct {
    events   chan Event
    known    map[int]ProcessInfo // Track known processes
    interval time.Duration
}

func NewProcessCollector(events chan Event) *ProcessCollector {
    return &ProcessCollector{
        events:   events,
        known:    make(map[int]ProcessInfo),
        interval: 5 * time.Second,
    }
}

func (c *ProcessCollector) Run() {
    ticker := time.NewTicker(c.interval)
    for range ticker.C {
        c.collect()
    }
}

func (c *ProcessCollector) collect() {
    procs, _ := filepath.Glob("/proc/[0-9]*")

    currentPIDs := make(map[int]bool)

    for _, procPath := range procs {
        pid, _ := strconv.Atoi(filepath.Base(procPath))
        currentPIDs[pid] = true

        info, err := c.getProcessInfo(pid)
        if err != nil {
            continue
        }

        // Check if this is a new process
        if _, exists := c.known[pid]; !exists {
            c.events <- Event{
                Type:      EventProcess,
                Timestamp: time.Now(),
                Source:    "process_collector",
                Data: map[string]any{
                    "action":     "spawn",
                    "pid":        info.PID,
                    "name":       info.Name,
                    "cmdline":    info.Cmdline,
                    "user":       info.User,
                    "parent_pid": info.ParentPID,
                    "executable": info.Executable,
                },
            }
        }

        c.known[pid] = info
    }

    // Detect terminated processes
    for pid := range c.known {
        if !currentPIDs[pid] {
            delete(c.known, pid)
        }
    }
}

func (c *ProcessCollector) getProcessInfo(pid int) (ProcessInfo, error) {
    info := ProcessInfo{PID: pid}

    // Read comm (process name)
    comm, err := os.ReadFile(fmt.Sprintf("/proc/%d/comm", pid))
    if err != nil {
        return info, err
    }
    info.Name = strings.TrimSpace(string(comm))

    // Read cmdline
    cmdline, _ := os.ReadFile(fmt.Sprintf("/proc/%d/cmdline", pid))
    info.Cmdline = strings.ReplaceAll(string(cmdline), "\x00", " ")

    // Read executable path
    exe, _ := os.Readlink(fmt.Sprintf("/proc/%d/exe", pid))
    info.Executable = exe

    // Read status for user and parent PID
    status, err := os.Open(fmt.Sprintf("/proc/%d/status", pid))
    if err == nil {
        defer status.Close()
        scanner := bufio.NewScanner(status)
        for scanner.Scan() {
            line := scanner.Text()
            if strings.HasPrefix(line, "Uid:") {
                fields := strings.Fields(line)
                if len(fields) >= 2 {
                    uid, _ := strconv.Atoi(fields[1])
                    info.User = fmt.Sprintf("%d", uid)
                }
            }
            if strings.HasPrefix(line, "PPid:") {
                fields := strings.Fields(line)
                if len(fields) >= 2 {
                    info.ParentPID, _ = strconv.Atoi(fields[1])
                }
            }
        }
    }

    return info, nil
}

Network Collector

Watches for suspicious connections - unusual ports, known bad IPs, data exfiltration patterns:

package collectors

import (
    "bufio"
    "encoding/hex"
    "fmt"
    "net"
    "os"
    "strconv"
    "strings"
    "time"
)

type Connection struct {
    LocalAddr  string
    LocalPort  int
    RemoteAddr string
    RemotePort int
    State      string
    PID        int
}

type NetworkCollector struct {
    events     chan Event
    known      map[string]Connection
    interval   time.Duration
    suspPorts  map[int]bool // Suspicious ports to watch
}

func NewNetworkCollector(events chan Event) *NetworkCollector {
    return &NetworkCollector{
        events:   events,
        known:    make(map[string]Connection),
        interval: 3 * time.Second,
        suspPorts: map[int]bool{
            4444: true,  // Metasploit default
            5555: true,  // Common backdoor
            6666: true,  // IRC (often C2)
            1337: true,  // Leet port
            31337: true, // Back Orifice
        },
    }
}

func (c *NetworkCollector) Run() {
    ticker := time.NewTicker(c.interval)
    for range ticker.C {
        c.collect()
    }
}

func (c *NetworkCollector) collect() {
    // Parse /proc/net/tcp and /proc/net/tcp6
    for _, path := range []string{"/proc/net/tcp", "/proc/net/tcp6"} {
        c.parseNetFile(path)
    }
}

func (c *NetworkCollector) parseNetFile(path string) {
    file, err := os.Open(path)
    if err != nil {
        return
    }
    defer file.Close()

    scanner := bufio.NewScanner(file)
    scanner.Scan() // Skip header

    for scanner.Scan() {
        conn := c.parseLine(scanner.Text())
        if conn == nil {
            continue
        }

        key := fmt.Sprintf("%s:%d-%s:%d",
            conn.LocalAddr, conn.LocalPort,
            conn.RemoteAddr, conn.RemotePort)

        // New connection
        if _, exists := c.known[key]; !exists && conn.State == "ESTABLISHED" {
            c.events <- Event{
                Type:      EventNetwork,
                Timestamp: time.Now(),
                Source:    "network_collector",
                Data: map[string]any{
                    "action":      "connect",
                    "local_addr":  conn.LocalAddr,
                    "local_port":  conn.LocalPort,
                    "remote_addr": conn.RemoteAddr,
                    "remote_port": conn.RemotePort,
                    "pid":         conn.PID,
                },
            }
            c.known[key] = *conn
        }
    }
}

func (c *NetworkCollector) parseLine(line string) *Connection {
    fields := strings.Fields(line)
    if len(fields) < 10 {
        return nil
    }

    localAddr, localPort := parseAddr(fields[1])
    remoteAddr, remotePort := parseAddr(fields[2])
    state := parseState(fields[3])

    return &Connection{
        LocalAddr:  localAddr,
        LocalPort:  localPort,
        RemoteAddr: remoteAddr,
        RemotePort: remotePort,
        State:      state,
    }
}

func parseAddr(s string) (string, int) {
    parts := strings.Split(s, ":")
    if len(parts) != 2 {
        return "", 0
    }

    // IP is in hex, reversed
    ipHex := parts[0]
    ip := make(net.IP, 4)
    for i := 0; i < 4; i++ {
        b, _ := hex.DecodeString(ipHex[i*2 : i*2+2])
        ip[3-i] = b[0]
    }

    port, _ := strconv.ParseInt(parts[1], 16, 32)
    return ip.String(), int(port)
}

func parseState(s string) string {
    states := map[string]string{
        "01": "ESTABLISHED",
        "02": "SYN_SENT",
        "0A": "LISTEN",
    }
    return states[s]
}

Filesystem Collector

Uses inotify to watch for suspicious file changes:

package collectors

import (
    "crypto/sha256"
    "encoding/hex"
    "io"
    "os"
    "path/filepath"
    "time"

    "github.com/fsnotify/fsnotify"
)

type FilesystemCollector struct {
    events    chan Event
    watcher   *fsnotify.Watcher
    watchDirs []string
    hashes    map[string]string // Track file hashes
}

func NewFilesystemCollector(events chan Event) *FilesystemCollector {
    watcher, _ := fsnotify.NewWatcher()
    return &FilesystemCollector{
        events:  events,
        watcher: watcher,
        watchDirs: []string{
            "/etc",
            "/usr/bin",
            "/usr/sbin",
            "/root",
            "/home",
        },
        hashes: make(map[string]string),
    }
}

func (c *FilesystemCollector) Run() {
    // Add watches
    for _, dir := range c.watchDirs {
        filepath.Walk(dir, func(path string, info os.FileInfo, err error) error {
            if err != nil || info.IsDir() {
                return nil
            }
            c.watcher.Add(path)
            c.hashes[path] = c.hashFile(path)
            return nil
        })
    }

    // Process events
    for {
        select {
        case event := <-c.watcher.Events:
            c.handleEvent(event)
        case err := <-c.watcher.Errors:
            if err != nil {
                // Log error
            }
        }
    }
}

func (c *FilesystemCollector) handleEvent(e fsnotify.Event) {
    eventType := ""
    switch {
    case e.Op&fsnotify.Write == fsnotify.Write:
        eventType = "modify"
    case e.Op&fsnotify.Create == fsnotify.Create:
        eventType = "create"
    case e.Op&fsnotify.Remove == fsnotify.Remove:
        eventType = "delete"
    case e.Op&fsnotify.Chmod == fsnotify.Chmod:
        eventType = "chmod"
    default:
        return
    }

    newHash := ""
    if eventType != "delete" {
        newHash = c.hashFile(e.Name)
    }

    c.events <- Event{
        Type:      EventFilesystem,
        Timestamp: time.Now(),
        Source:    "filesystem_collector",
        Data: map[string]any{
            "action":   eventType,
            "path":     e.Name,
            "old_hash": c.hashes[e.Name],
            "new_hash": newHash,
        },
    }

    c.hashes[e.Name] = newHash
}

func (c *FilesystemCollector) hashFile(path string) string {
    f, err := os.Open(path)
    if err != nil {
        return ""
    }
    defer f.Close()

    h := sha256.New()
    io.Copy(h, f)
    return hex.EncodeToString(h.Sum(nil))
}

Auth Log Collector

Watches authentication logs for failed logins, sudo abuse, SSH anomalies:

package collectors

import (
    "bufio"
    "os"
    "regexp"
    "strings"
    "time"

    "github.com/hpcloud/tail"
)

type AuthCollector struct {
    events   chan Event
    logPath  string
    patterns map[string]*regexp.Regexp
}

func NewAuthCollector(events chan Event) *AuthCollector {
    return &AuthCollector{
        events:  events,
        logPath: "/var/log/auth.log",
        patterns: map[string]*regexp.Regexp{
            "failed_login":   regexp.MustCompile(`Failed password for (\S+) from (\S+)`),
            "invalid_user":   regexp.MustCompile(`Invalid user (\S+) from (\S+)`),
            "sudo":           regexp.MustCompile(`sudo:\s+(\S+).*COMMAND=(.*)`),
            "ssh_accepted":   regexp.MustCompile(`Accepted (\S+) for (\S+) from (\S+)`),
            "ssh_disconnect": regexp.MustCompile(`Disconnected from (\S+)`),
        },
    }
}

func (c *AuthCollector) Run() {
    t, err := tail.TailFile(c.logPath, tail.Config{
        Follow:    true,
        ReOpen:    true,
        MustExist: false,
    })
    if err != nil {
        return
    }

    for line := range t.Lines {
        c.parseLine(line.Text)
    }
}

func (c *AuthCollector) parseLine(line string) {
    for eventType, pattern := range c.patterns {
        matches := pattern.FindStringSubmatch(line)
        if matches == nil {
            continue
        }

        data := map[string]any{
            "action":  eventType,
            "raw_log": line,
        }

        switch eventType {
        case "failed_login":
            data["user"] = matches[1]
            data["source_ip"] = matches[2]
        case "invalid_user":
            data["user"] = matches[1]
            data["source_ip"] = matches[2]
        case "sudo":
            data["user"] = matches[1]
            data["command"] = matches[2]
        case "ssh_accepted":
            data["method"] = matches[1]
            data["user"] = matches[2]
            data["source_ip"] = matches[3]
        }

        c.events <- Event{
            Type:      EventAuth,
            Timestamp: time.Now(),
            Source:    "auth_collector",
            Data:      data,
        }
    }
}

Anomaly Detection with ML

The ML layer does fast anomaly scoring. We use a simple statistical approach - no external dependencies, runs in microseconds.

package detector

import (
    "math"
    "sync"
    "time"
)

// RollingStats tracks statistics for anomaly detection
type RollingStats struct {
    mu       sync.RWMutex
    values   []float64
    maxSize  int
    mean     float64
    variance float64
}

func NewRollingStats(size int) *RollingStats {
    return &RollingStats{
        values:  make([]float64, 0, size),
        maxSize: size,
    }
}

func (s *RollingStats) Add(v float64) {
    s.mu.Lock()
    defer s.mu.Unlock()

    s.values = append(s.values, v)
    if len(s.values) > s.maxSize {
        s.values = s.values[1:]
    }

    // Update running stats
    s.mean = 0
    for _, x := range s.values {
        s.mean += x
    }
    s.mean /= float64(len(s.values))

    s.variance = 0
    for _, x := range s.values {
        s.variance += (x - s.mean) * (x - s.mean)
    }
    s.variance /= float64(len(s.values))
}

func (s *RollingStats) ZScore(v float64) float64 {
    s.mu.RLock()
    defer s.mu.RUnlock()

    if s.variance == 0 {
        return 0
    }
    return (v - s.mean) / math.Sqrt(s.variance)
}

// AnomalyDetector scores events for anomalousness
type AnomalyDetector struct {
    // Track stats per event type
    processRate   *RollingStats
    networkRate   *RollingStats
    authFailRate  *RollingStats

    // Counters for rate calculation
    processCount  int
    networkCount  int
    authFailCount int
    lastReset     time.Time

    // Known bad patterns
    suspiciousProcesses map[string]bool
    suspiciousPorts     map[int]bool
}

func NewAnomalyDetector() *AnomalyDetector {
    return &AnomalyDetector{
        processRate:  NewRollingStats(100),
        networkRate:  NewRollingStats(100),
        authFailRate: NewRollingStats(100),
        lastReset:    time.Now(),

        suspiciousProcesses: map[string]bool{
            "nc":       true,
            "ncat":     true,
            "netcat":   true,
            "nmap":     true,
            "masscan":  true,
            "hydra":    true,
            "john":     true,
            "hashcat":  true,
            "mimikatz": true,
            "xmrig":    true, // Crypto miner
            "minerd":   true,
        },

        suspiciousPorts: map[int]bool{
            4444:  true,
            5555:  true,
            6666:  true,
            1337:  true,
            31337: true,
        },
    }
}

func (d *AnomalyDetector) Score(event Event) (float64, string) {
    switch event.Type {
    case EventProcess:
        return d.scoreProcess(event)
    case EventNetwork:
        return d.scoreNetwork(event)
    case EventAuth:
        return d.scoreAuth(event)
    case EventFilesystem:
        return d.scoreFilesystem(event)
    }
    return 0, ""
}

func (d *AnomalyDetector) scoreProcess(event Event) (float64, string) {
    score := 0.0
    reasons := []string{}

    name, _ := event.Data["name"].(string)
    cmdline, _ := event.Data["cmdline"].(string)
    executable, _ := event.Data["executable"].(string)

    // Check known suspicious processes
    if d.suspiciousProcesses[name] {
        score += 0.7
        reasons = append(reasons, "known_suspicious_process")
    }

    // Check for deleted executables (common malware technique)
    if strings.Contains(executable, "(deleted)") {
        score += 0.8
        reasons = append(reasons, "deleted_executable")
    }

    // Check for processes running from /tmp or /dev/shm
    if strings.HasPrefix(executable, "/tmp") || strings.HasPrefix(executable, "/dev/shm") {
        score += 0.6
        reasons = append(reasons, "execution_from_tmp")
    }

    // Check for base64 in command line (obfuscation)
    if strings.Contains(cmdline, "base64") {
        score += 0.4
        reasons = append(reasons, "base64_in_cmdline")
    }

    // Check for reverse shell patterns
    if strings.Contains(cmdline, "/dev/tcp") || strings.Contains(cmdline, "bash -i") {
        score += 0.9
        reasons = append(reasons, "reverse_shell_pattern")
    }

    // Normalize score
    if score > 1.0 {
        score = 1.0
    }

    return score, strings.Join(reasons, ", ")
}

func (d *AnomalyDetector) scoreNetwork(event Event) (float64, string) {
    score := 0.0
    reasons := []string{}

    remotePort, _ := event.Data["remote_port"].(int)
    remoteAddr, _ := event.Data["remote_addr"].(string)

    // Check suspicious ports
    if d.suspiciousPorts[remotePort] {
        score += 0.7
        reasons = append(reasons, "suspicious_port")
    }

    // Check for connections to private IPs from public-facing services
    // (potential lateral movement)
    if isPrivateIP(remoteAddr) {
        // This might be normal, lower score
        score += 0.1
    }

    // Rate-based anomaly
    d.networkCount++
    elapsed := time.Since(d.lastReset).Seconds()
    if elapsed > 60 {
        rate := float64(d.networkCount) / elapsed
        d.networkRate.Add(rate)
        d.networkCount = 0
        d.lastReset = time.Now()

        zScore := d.networkRate.ZScore(rate)
        if zScore > 3 {
            score += 0.5
            reasons = append(reasons, "unusual_connection_rate")
        }
    }

    if score > 1.0 {
        score = 1.0
    }

    return score, strings.Join(reasons, ", ")
}

func (d *AnomalyDetector) scoreAuth(event Event) (float64, string) {
    score := 0.0
    reasons := []string{}

    action, _ := event.Data["action"].(string)

    if action == "failed_login" || action == "invalid_user" {
        score += 0.3
        reasons = append(reasons, "failed_auth")

        // Track rate of failures
        d.authFailCount++
        if d.authFailCount > 5 {
            score += 0.4
            reasons = append(reasons, "brute_force_pattern")
        }
    }

    if action == "sudo" {
        command, _ := event.Data["command"].(string)
        // Check for suspicious sudo commands
        if strings.Contains(command, "chmod 777") ||
            strings.Contains(command, "rm -rf") ||
            strings.Contains(command, "/etc/shadow") {
            score += 0.6
            reasons = append(reasons, "suspicious_sudo_command")
        }
    }

    if score > 1.0 {
        score = 1.0
    }

    return score, strings.Join(reasons, ", ")
}

func (d *AnomalyDetector) scoreFilesystem(event Event) (float64, string) {
    score := 0.0
    reasons := []string{}

    path, _ := event.Data["path"].(string)
    action, _ := event.Data["action"].(string)

    // Critical files
    criticalFiles := []string{
        "/etc/passwd",
        "/etc/shadow",
        "/etc/sudoers",
        "/root/.ssh/authorized_keys",
    }

    for _, critical := range criticalFiles {
        if path == critical {
            score += 0.8
            reasons = append(reasons, "critical_file_modified")
            break
        }
    }

    // New executables in sensitive directories
    if action == "create" && (strings.HasPrefix(path, "/usr/bin") || strings.HasPrefix(path, "/usr/sbin")) {
        score += 0.7
        reasons = append(reasons, "new_system_binary")
    }

    if score > 1.0 {
        score = 1.0
    }

    return score, strings.Join(reasons, ", ")
}

func isPrivateIP(ip string) bool {
    return strings.HasPrefix(ip, "10.") ||
        strings.HasPrefix(ip, "192.168.") ||
        strings.HasPrefix(ip, "172.16.") ||
        strings.HasPrefix(ip, "172.17.") ||
        strings.HasPrefix(ip, "172.18.") ||
        strings.HasPrefix(ip, "172.19.")
}

AI Reasoning with Ollama

High-scoring anomalies get sent to Ollama for deeper analysis. The LLM determines if it’s a real threat and suggests remediation.

package ai

import (
    "bytes"
    "encoding/json"
    "fmt"
    "net/http"
    "strings"
)

type OllamaClient struct {
    baseURL string
    model   string
}

func NewOllamaClient(baseURL, model string) *OllamaClient {
    return &OllamaClient{
        baseURL: baseURL,
        model:   model,
    }
}

type OllamaRequest struct {
    Model  string `json:"model"`
    Prompt string `json:"prompt"`
    Stream bool   `json:"stream"`
}

type OllamaResponse struct {
    Response string `json:"response"`
}

func (c *OllamaClient) Analyze(anomaly Anomaly) (threat string, remediation string, err error) {
    prompt := fmt.Sprintf(`You are a security analyst AI. Analyze this system anomaly and respond with JSON.

Event Type: %s
Event Data: %v
Anomaly Score: %.2f
Detection Reason: %s

Analyze this event and respond with ONLY a JSON object (no other text):
{
    "is_threat": true/false,
    "threat_level": "critical/high/medium/low/none",
    "explanation": "brief explanation of what this event means",
    "remediation": {
        "action": "kill/block/restore/alert/none",
        "target": "the target (PID, IP, filepath, etc)",
        "reason": "why this action"
    }
}`,
        anomaly.Event.Type,
        anomaly.Event.Data,
        anomaly.Score,
        anomaly.Reason,
    )

    reqBody, _ := json.Marshal(OllamaRequest{
        Model:  c.model,
        Prompt: prompt,
        Stream: false,
    })

    resp, err := http.Post(
        c.baseURL+"/api/generate",
        "application/json",
        bytes.NewBuffer(reqBody),
    )
    if err != nil {
        return "", "", err
    }
    defer resp.Body.Close()

    var ollamaResp OllamaResponse
    if err := json.NewDecoder(resp.Body).Decode(&ollamaResp); err != nil {
        return "", "", err
    }

    // Parse the JSON response
    var analysis struct {
        IsThreat    bool   `json:"is_threat"`
        ThreatLevel string `json:"threat_level"`
        Explanation string `json:"explanation"`
        Remediation struct {
            Action string `json:"action"`
            Target string `json:"target"`
            Reason string `json:"reason"`
        } `json:"remediation"`
    }

    // Extract JSON from response (LLM might include extra text)
    responseText := ollamaResp.Response
    startIdx := strings.Index(responseText, "{")
    endIdx := strings.LastIndex(responseText, "}")
    if startIdx >= 0 && endIdx > startIdx {
        jsonStr := responseText[startIdx : endIdx+1]
        if err := json.Unmarshal([]byte(jsonStr), &analysis); err != nil {
            return "", "", fmt.Errorf("failed to parse LLM response: %w", err)
        }
    }

    threat = fmt.Sprintf("[%s] %s", analysis.ThreatLevel, analysis.Explanation)

    if analysis.IsThreat && analysis.Remediation.Action != "none" {
        remediation = fmt.Sprintf("%s:%s (%s)",
            analysis.Remediation.Action,
            analysis.Remediation.Target,
            analysis.Remediation.Reason,
        )
    }

    return threat, remediation, nil
}

Remediation Engine

The remediation engine executes actions suggested by the AI. It has safety checks and dry-run support.

package remediation

import (
    "fmt"
    "os/exec"
    "strconv"
    "strings"
)

type Engine struct {
    dryRun     bool
    allowlist  map[string]bool // Processes that should never be killed
    blockRules []string        // Track blocked IPs for cleanup
}

func NewEngine(dryRun bool) *Engine {
    return &Engine{
        dryRun: dryRun,
        allowlist: map[string]bool{
            "systemd":   true,
            "init":      true,
            "sshd":      true,
            "kernel":    true,
            "dockerd":   true,
            "containerd": true,
        },
    }
}

func (e *Engine) Execute(action Action) error {
    switch action.Type {
    case "kill":
        return e.killProcess(action)
    case "block":
        return e.blockIP(action)
    case "restore":
        return e.restoreFile(action)
    case "alert":
        return e.sendAlert(action)
    default:
        return fmt.Errorf("unknown action type: %s", action.Type)
    }
}

func (e *Engine) killProcess(action Action) error {
    pidStr := action.Target
    pid, err := strconv.Atoi(pidStr)
    if err != nil {
        return fmt.Errorf("invalid PID: %s", pidStr)
    }

    // Safety check: don't kill system processes
    processName := action.Params["name"].(string)
    if e.allowlist[processName] {
        return fmt.Errorf("refusing to kill allowlisted process: %s", processName)
    }

    // Don't kill PID 1 or low PIDs
    if pid <= 100 {
        return fmt.Errorf("refusing to kill low PID: %d", pid)
    }

    if e.dryRun {
        fmt.Printf("[DRY RUN] Would kill process %d (%s)\n", pid, processName)
        return nil
    }

    cmd := exec.Command("kill", "-9", pidStr)
    return cmd.Run()
}

func (e *Engine) blockIP(action Action) error {
    ip := action.Target

    // Validate IP format
    if !isValidIP(ip) {
        return fmt.Errorf("invalid IP address: %s", ip)
    }

    // Don't block private IPs or localhost
    if strings.HasPrefix(ip, "127.") || strings.HasPrefix(ip, "10.") ||
        strings.HasPrefix(ip, "192.168.") {
        return fmt.Errorf("refusing to block private/localhost IP: %s", ip)
    }

    if e.dryRun {
        fmt.Printf("[DRY RUN] Would block IP %s\n", ip)
        return nil
    }

    // Use iptables to block
    cmd := exec.Command("iptables", "-A", "INPUT", "-s", ip, "-j", "DROP")
    if err := cmd.Run(); err != nil {
        return err
    }

    e.blockRules = append(e.blockRules, ip)
    return nil
}

func (e *Engine) restoreFile(action Action) error {
    path := action.Target
    backupPath := action.Params["backup"].(string)

    if e.dryRun {
        fmt.Printf("[DRY RUN] Would restore %s from %s\n", path, backupPath)
        return nil
    }

    cmd := exec.Command("cp", backupPath, path)
    return cmd.Run()
}

func (e *Engine) sendAlert(action Action) error {
    message := action.Params["message"].(string)

    // Log locally
    fmt.Printf("[ALERT] %s\n", message)

    // Could integrate with PagerDuty, Slack, etc.
    // For now, just write to a file
    if !e.dryRun {
        cmd := exec.Command("logger", "-t", "immune-system", message)
        return cmd.Run()
    }

    return nil
}

func isValidIP(ip string) bool {
    parts := strings.Split(ip, ".")
    if len(parts) != 4 {
        return false
    }
    for _, part := range parts {
        n, err := strconv.Atoi(part)
        if err != nil || n < 0 || n > 255 {
            return false
        }
    }
    return true
}

Main Loop

Tie it all together:

package main

import (
    "flag"
    "fmt"
    "log"
    "time"
)

func main() {
    dryRun := flag.Bool("dry-run", true, "Don't execute remediation actions")
    threshold := flag.Float64("threshold", 0.5, "Anomaly score threshold for AI analysis")
    ollamaURL := flag.String("ollama-url", "http://localhost:11434", "Ollama API URL")
    ollamaModel := flag.String("ollama-model", "llama3.2", "Ollama model to use")
    flag.Parse()

    // Initialize components
    events := make(chan Event, 1000)
    detector := NewAnomalyDetector()
    aiClient := NewOllamaClient(*ollamaURL, *ollamaModel)
    remediator := NewEngine(*dryRun)

    // Start collectors
    go NewProcessCollector(events).Run()
    go NewNetworkCollector(events).Run()
    go NewFilesystemCollector(events).Run()
    go NewAuthCollector(events).Run()

    log.Printf("Digital Immune System started (dry-run=%v, threshold=%.2f)", *dryRun, *threshold)

    // Main processing loop
    for event := range events {
        // Fast ML scoring
        score, reason := detector.Score(event)

        if score < *threshold {
            continue // Not anomalous enough
        }

        anomaly := Anomaly{
            Event:  event,
            Score:  score,
            Reason: reason,
        }

        log.Printf("Anomaly detected: type=%s score=%.2f reason=%s",
            event.Type, score, reason)

        // Send to AI for deeper analysis
        threat, remediation, err := aiClient.Analyze(anomaly)
        if err != nil {
            log.Printf("AI analysis failed: %v", err)
            continue
        }

        anomaly.Threat = threat
        anomaly.Remediation = remediation

        log.Printf("AI analysis: threat=%s remediation=%s", threat, remediation)

        // Execute remediation if suggested
        if remediation != "" {
            action := parseRemediation(remediation, event)
            if action != nil {
                if err := remediator.Execute(*action); err != nil {
                    log.Printf("Remediation failed: %v", err)
                } else {
                    log.Printf("Remediation executed: %s", action.Type)
                }
            }
        }
    }
}

func parseRemediation(remediation string, event Event) *Action {
    parts := strings.SplitN(remediation, ":", 2)
    if len(parts) != 2 {
        return nil
    }

    actionType := parts[0]
    target := strings.Split(parts[1], " ")[0]

    return &Action{
        Type:   actionType,
        Target: target,
        Params: event.Data,
    }
}

Running It

# Start Ollama with a model
ollama pull llama3.2
ollama serve

# Build and run (dry-run mode first!)
go build -o immune-system .
sudo ./immune-system --dry-run=true --threshold=0.5

# Once confident, enable remediation
sudo ./immune-system --dry-run=false --threshold=0.6

Example output:

2024/02/19 10:23:45 Digital Immune System started (dry-run=true, threshold=0.50)
2024/02/19 10:24:01 Anomaly detected: type=process score=0.70 reason=known_suspicious_process
2024/02/19 10:24:02 AI analysis: threat=[high] nmap is a network scanning tool commonly used for reconnaissance remediation=alert:nmap (reconnaissance activity detected)
2024/02/19 10:24:02 Remediation executed: alert
2024/02/19 10:25:33 Anomaly detected: type=auth score=0.70 reason=failed_auth, brute_force_pattern
2024/02/19 10:25:34 AI analysis: threat=[critical] Multiple failed SSH attempts from same IP indicates brute force attack remediation=block:192.0.2.50 (block attacking IP)
2024/02/19 10:25:34 [DRY RUN] Would block IP 192.0.2.50

Tradeoffs

What works well:

Fast ML filtering means the LLM only sees interesting events
Local AI means no data leaves your network
Simple collectors are easy to understand and extend

What’s tricky:

LLM response parsing is fragile - models don’t always output valid JSON
Threshold tuning requires baseline data from your specific environment
Remediation actions can cause outages if the AI is wrong

Improvements to consider:

Add a feedback loop - track remediation outcomes to improve scoring
Implement rollback for remediation actions
Add a web UI for reviewing anomalies before auto-remediation
Use embeddings to detect similarity to known attack patterns

Platform Engineering for Small Teams: Building a Lightweight Internal Developer Portal

2024-05-12T00:00:00+00:00

Platform engineering is having a moment. Large companies build dedicated teams to create “golden paths” that let developers deploy without thinking about infrastructure. But what if you’re a team of 5-10 engineers? You don’t have the headcount for a platform team, yet you’re drowning in the same toil: inconsistent environments, flaky CI pipelines, and tribal knowledge about how to deploy things.

This post shows how to build a lightweight Internal Developer Portal (IDP) using Kubernetes-native tools. The goal is maximum automation with minimum maintenance - something a small team can actually sustain.

What We’re Building

An IDP that handles three things:

Service Catalog: What services exist, who owns them, how to use them
Environment Setup: Spin up dev/staging/prod with one command
CI/CD with Gates: Automated pipelines that enforce quality standards

The stack:

ArgoCD for GitOps deployments
Crossplane for infrastructure provisioning
GitHub Actions for CI pipelines
A simple catalog using markdown + conventions (no Backstage needed)

The Service Catalog: Keep It Simple

Backstage is powerful but complex. For small teams, a convention-based catalog works better.

Convention: The `service.yaml` File

Every service repo has a service.yaml at the root:

# service.yaml
apiVersion: platform.example.com/v1
kind: Service
metadata:
  name: user-api
  team: backend
  slack: "#backend-alerts"
spec:
  description: "User authentication and profile management"
  language: go
  framework: gin

  dependencies:
    - name: postgres
      type: database
    - name: redis
      type: cache
    - name: auth-service
      type: service

  endpoints:
    production: https://api.example.com/users
    staging: https://staging-api.example.com/users

  runbook: ./docs/runbook.md

  slos:
    availability: 99.9%
    latency_p99: 200ms

This file is the source of truth. It’s versioned with the code, so it stays up to date.

Catalog Generation

A simple script aggregates all service.yaml files into a searchable catalog:

#!/usr/bin/env python3
# scripts/generate-catalog.py

import yaml
import json
import subprocess
from pathlib import Path

def get_repos():
    """Get all repos from GitHub org"""
    result = subprocess.run(
        ["gh", "repo", "list", "your-org", "--json", "name,url", "-L", "200"],
        capture_output=True, text=True
    )
    return json.loads(result.stdout)

def fetch_service_yaml(repo_name):
    """Fetch service.yaml from repo"""
    result = subprocess.run(
        ["gh", "api", f"/repos/your-org/{repo_name}/contents/service.yaml"],
        capture_output=True, text=True
    )
    if result.returncode != 0:
        return None

    content = json.loads(result.stdout)
    import base64
    return yaml.safe_load(base64.b64decode(content["content"]))

def main():
    catalog = {"services": []}

    for repo in get_repos():
        service = fetch_service_yaml(repo["name"])
        if service:
            service["_repo"] = repo["url"]
            catalog["services"].append(service)

    # Output as JSON for the catalog UI
    print(json.dumps(catalog, indent=2))

    # Also generate markdown for docs
    with open("docs/catalog.md", "w") as f:
        f.write("# Service Catalog\n\n")
        for svc in catalog["services"]:
            meta = svc["metadata"]
            spec = svc["spec"]
            f.write(f"## {meta['name']}\n")
            f.write(f"**Team:** {meta['team']} | **Slack:** {meta['slack']}\n\n")
            f.write(f"{spec['description']}\n\n")
            f.write(f"- **Language:** {spec['language']}\n")
            f.write(f"- **Runbook:** [{spec['runbook']}]({svc['_repo']}/blob/main/{spec['runbook']})\n\n")

if __name__ == "__main__":
    main()

Run this on a cron job (or GitHub Action) to keep the catalog fresh. The output feeds a simple static site or even just a markdown file in your docs repo.

Environment Setup with Crossplane

Crossplane lets you define infrastructure as Kubernetes resources. This means your environment definitions live in Git and deploy through the same ArgoCD pipeline as your applications.

Composite Resource Definition

First, define what an “environment” means for your organization:

# platform/apis/environment-definition.yaml
apiVersion: apiextensions.crossplane.io/v1
kind: CompositeResourceDefinition
metadata:
  name: xenvironments.platform.example.com
spec:
  group: platform.example.com
  names:
    kind: XEnvironment
    plural: xenvironments
  versions:
    - name: v1
      served: true
      referenceable: true
      schema:
        openAPIV3Schema:
          type: object
          properties:
            spec:
              type: object
              properties:
                name:
                  type: string
                  description: "Environment name (dev, staging, prod)"
                size:
                  type: string
                  enum: [small, medium, large]
                  default: small
                services:
                  type: array
                  items:
                    type: string
                  description: "List of services to deploy"
              required:
                - name
                - services

Composition: What Gets Created

The composition defines what infrastructure to provision for each environment:

# platform/compositions/environment-composition.yaml
apiVersion: apiextensions.crossplane.io/v1
kind: Composition
metadata:
  name: environment-aws
spec:
  compositeTypeRef:
    apiVersion: platform.example.com/v1
    kind: XEnvironment

  resources:
    # Namespace for the environment
    - name: namespace
      base:
        apiVersion: kubernetes.crossplane.io/v1alpha1
        kind: Object
        spec:
          forProvider:
            manifest:
              apiVersion: v1
              kind: Namespace
              metadata:
                name: "" # patched
      patches:
        - fromFieldPath: spec.name
          toFieldPath: spec.forProvider.manifest.metadata.name

    # RDS PostgreSQL instance
    - name: database
      base:
        apiVersion: database.aws.crossplane.io/v1beta1
        kind: RDSInstance
        spec:
          forProvider:
            region: us-west-2
            dbInstanceClass: db.t3.micro  # patched based on size
            engine: postgres
            engineVersion: "15"
            masterUsername: admin
            allocatedStorage: 20
            skipFinalSnapshot: true
          writeConnectionSecretToRef:
            namespace: "" # patched
            name: "" # patched
      patches:
        - fromFieldPath: spec.name
          toFieldPath: spec.forProvider.dbName
        - fromFieldPath: spec.name
          toFieldPath: spec.writeConnectionSecretToRef.namespace
        - fromFieldPath: spec.name
          toFieldPath: spec.writeConnectionSecretToRef.name
          transforms:
            - type: string
              string:
                fmt: "%s-db-creds"
        # Size mapping
        - fromFieldPath: spec.size
          toFieldPath: spec.forProvider.dbInstanceClass
          transforms:
            - type: map
              map:
                small: db.t3.micro
                medium: db.t3.small
                large: db.t3.medium

    # Redis ElastiCache
    - name: cache
      base:
        apiVersion: cache.aws.crossplane.io/v1beta1
        kind: ReplicationGroup
        spec:
          forProvider:
            region: us-west-2
            engine: redis
            cacheNodeType: cache.t3.micro  # patched based on size
            numCacheClusters: 1
      patches:
        - fromFieldPath: spec.name
          toFieldPath: spec.forProvider.replicationGroupDescription
        - fromFieldPath: spec.size
          toFieldPath: spec.forProvider.cacheNodeType
          transforms:
            - type: map
              map:
                small: cache.t3.micro
                medium: cache.t3.small
                large: cache.t3.medium

Creating an Environment

Now developers can spin up environments with a simple YAML file:

# environments/staging.yaml
apiVersion: platform.example.com/v1
kind: XEnvironment
metadata:
  name: staging
spec:
  name: staging
  size: medium
  services:
    - user-api
    - order-service
    - notification-service

Commit this to Git, and ArgoCD provisions everything: namespace, database, cache, and deploys the services.

CLI Wrapper

Make it even simpler with a CLI:

#!/bin/bash
# platform-cli

case "$1" in
  env:create)
    ENV_NAME=$2
    SIZE=${3:-small}

    cat <<EOF | kubectl apply -f -
apiVersion: platform.example.com/v1
kind: XEnvironment
metadata:
  name: $ENV_NAME
spec:
  name: $ENV_NAME
  size: $SIZE
  services: []
EOF
    echo "Environment '$ENV_NAME' created. Add services with: platform-cli env:add-service $ENV_NAME <service>"
    ;;

  env:add-service)
    ENV_NAME=$2
    SERVICE=$3
    kubectl patch xenvironment $ENV_NAME --type=json \
      -p="[{\"op\": \"add\", \"path\": \"/spec/services/-\", \"value\": \"$SERVICE\"}]"
    ;;

  env:delete)
    ENV_NAME=$2
    kubectl delete xenvironment $ENV_NAME
    echo "Environment '$ENV_NAME' and all resources will be deleted."
    ;;

  env:list)
    kubectl get xenvironments -o wide
    ;;

  *)
    echo "Usage: platform-cli <command>"
    echo "Commands:"
    echo "  env:create <name> [size]  - Create new environment"
    echo "  env:add-service <env> <svc> - Add service to environment"
    echo "  env:delete <name>         - Delete environment"
    echo "  env:list                  - List all environments"
    ;;
esac

CI/CD with Quality Gates

GitHub Actions handles CI. The key is defining reusable workflows that enforce standards without slowing teams down.

Reusable CI Workflow

Create a shared workflow that all services use:

# .github/workflows/ci-template.yaml (in your platform repo)
name: CI Template

on:
  workflow_call:
    inputs:
      language:
        required: true
        type: string
      run_e2e:
        required: false
        type: boolean
        default: false
    secrets:
      SONAR_TOKEN:
        required: false

jobs:
  lint:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Lint (Go)
        if: inputs.language == 'go'
        run: |
          go install github.com/golangci/golangci-lint/cmd/golangci-lint@latest
          golangci-lint run

      - name: Lint (Python)
        if: inputs.language == 'python'
        run: |
          pip install ruff
          ruff check .

      - name: Lint (TypeScript)
        if: inputs.language == 'typescript'
        run: |
          npm ci
          npm run lint

  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Test (Go)
        if: inputs.language == 'go'
        run: go test -race -coverprofile=coverage.out ./...

      - name: Test (Python)
        if: inputs.language == 'python'
        run: |
          pip install pytest pytest-cov
          pytest --cov=. --cov-report=xml

      - name: Test (TypeScript)
        if: inputs.language == 'typescript'
        run: |
          npm ci
          npm test -- --coverage

      - name: Upload coverage
        uses: codecov/codecov-action@v3

  security:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Run Trivy vulnerability scanner
        uses: aquasecurity/trivy-action@master
        with:
          scan-type: 'fs'
          scan-ref: '.'
          severity: 'CRITICAL,HIGH'
          exit-code: '1'

      - name: Run Semgrep
        uses: returntocorp/semgrep-action@v1
        with:
          config: p/default

  build:
    needs: [lint, test, security]
    runs-on: ubuntu-latest
    outputs:
      image_tag: $
    steps:
      - uses: actions/checkout@v4

      - name: Set up Docker Buildx
        uses: docker/setup-buildx-action@v3

      - name: Login to Registry
        uses: docker/login-action@v3
        with:
          registry: ghcr.io
          username: $
          password: $

      - name: Extract metadata
        id: meta
        uses: docker/metadata-action@v5
        with:
          images: ghcr.io/$
          tags: |
            type=sha,prefix=
            type=ref,event=branch
            type=ref,event=pr

      - name: Build and push
        uses: docker/build-push-action@v5
        with:
          context: .
          push: true
          tags: $
          cache-from: type=gha
          cache-to: type=gha,mode=max

Service-Specific CI

Each service calls the template:

# .github/workflows/ci.yaml (in each service repo)
name: CI

on:
  push:
    branches: [main]
  pull_request:
    branches: [main]

jobs:
  ci:
    uses: your-org/platform/.github/workflows/ci-template.yaml@main
    with:
      language: go
    secrets:
      SONAR_TOKEN: $

That’s it. All the lint, test, security, and build logic is centralized. Update the template, and all services get the improvement.

Quality Gates

Add gates that block merges if standards aren’t met:

# In the template, add a gate job
  gate:
    needs: [lint, test, security, build]
    runs-on: ubuntu-latest
    steps:
      - name: Check coverage threshold
        run: |
          COVERAGE=$(curl -s "https://codecov.io/api/v2/github/your-org/repos/$/commits/$" \
            -H "Authorization: Bearer $" | jq '.totals.coverage')

          if (( $(echo "$COVERAGE < 70" | bc -l) )); then
            echo "Coverage is $COVERAGE%, must be at least 70%"
            exit 1
          fi

      - name: Check for breaking changes
        if: github.event_name == 'pull_request'
        run: |
          # Use oasdiff or similar for API breaking change detection
          # This is a placeholder - implement based on your API spec format
          echo "Checking for breaking API changes..."

GitOps Deployment with ArgoCD

ArgoCD watches Git and deploys changes automatically. The pattern is:

CI builds image, pushes to registry
CI updates image tag in GitOps repo
ArgoCD detects change, deploys to cluster

Application Definition

# gitops/applications/user-api.yaml
apiVersion: argoproj.io/v1alpha1
kind: Application
metadata:
  name: user-api
  namespace: argocd
spec:
  project: default
  source:
    repoURL: https://github.com/your-org/gitops
    targetRevision: HEAD
    path: services/user-api
  destination:
    server: https://kubernetes.default.svc
    namespace: production
  syncPolicy:
    automated:
      prune: true
      selfHeal: true
    syncOptions:
      - CreateNamespace=true

Kustomize Overlays

Structure your GitOps repo with Kustomize for environment-specific config:

gitops/
├── services/
│   └── user-api/
│       ├── base/
│       │   ├── kustomization.yaml
│       │   ├── deployment.yaml
│       │   └── service.yaml
│       └── overlays/
│           ├── staging/
│           │   ├── kustomization.yaml
│           │   └── replicas-patch.yaml
│           └── production/
│               ├── kustomization.yaml
│               └── replicas-patch.yaml

Base deployment:

# gitops/services/user-api/base/deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: user-api
spec:
  replicas: 1
  selector:
    matchLabels:
      app: user-api
  template:
    metadata:
      labels:
        app: user-api
    spec:
      containers:
        - name: user-api
          image: ghcr.io/your-org/user-api:latest  # Updated by CI
          ports:
            - containerPort: 8080
          env:
            - name: DATABASE_URL
              valueFrom:
                secretKeyRef:
                  name: user-api-db-creds
                  key: url

Production overlay with more replicas:

# gitops/services/user-api/overlays/production/kustomization.yaml
apiVersion: kustomize.config.k8s.io/v1beta1
kind: Kustomization
resources:
  - ../../base
patches:
  - path: replicas-patch.yaml
images:
  - name: ghcr.io/your-org/user-api
    newTag: abc123  # Updated by CI

CI Updates GitOps Repo

Add a step to CI that updates the image tag:

  deploy:
    needs: [build]
    if: github.ref == 'refs/heads/main'
    runs-on: ubuntu-latest
    steps:
      - name: Checkout GitOps repo
        uses: actions/checkout@v4
        with:
          repository: your-org/gitops
          token: $

      - name: Update image tag
        run: |
          cd services/$/overlays/production
          kustomize edit set image ghcr.io/your-org/$:$

      - name: Commit and push
        run: |
          git config user.name "CI Bot"
          git config user.email "ci@example.com"
          git add .
          git commit -m "Deploy $:$"
          git push

ArgoCD picks up the change and rolls out the new version.

Putting It Together

The full flow:

Developer creates service.yaml in new repo
Catalog script picks it up, adds to documentation
Developer runs platform-cli env:create dev for a dev environment
Crossplane provisions database, cache, namespace
Developer pushes code
CI runs lint, test, security scan, builds image
CI updates GitOps repo with new image tag
ArgoCD deploys to the environment
Quality gates enforce standards before merge

All of this runs on ~4 tools (Crossplane, ArgoCD, GitHub Actions, a shell script) and requires no dedicated platform team to maintain.

Maintenance Tips

Keep compositions simple: Start with the minimum viable infrastructure. Add complexity only when teams ask for it.
Version your platform APIs: Use apiVersion in your CRDs. When you need breaking changes, create a v2.
Monitor the platform itself: ArgoCD and Crossplane have Prometheus metrics. Set up alerts for sync failures.
Document escape hatches: Teams will need to do things your platform doesn’t support. Document how to work around it rather than blocking them.
Review platform changes: Changes to shared workflows or compositions affect everyone. Treat them like infrastructure changes.

The goal isn’t to build a perfect platform. It’s to eliminate the most common sources of toil so your small team can focus on building product.

Distributing a Key-Value Store with Raft in Go

2023-01-28T00:00:00+00:00

In part 1, we built a single-node key-value store with persistence. It works, but there’s a glaring problem: if that node dies, your data is unavailable. The disk might be fine, but clients can’t reach it until someone restarts the process or provisions a new machine.

The solution is replication - keep copies of the data on multiple nodes so the system survives individual failures. But replication introduces a hard problem: how do you keep the copies consistent? If two clients write to different nodes simultaneously, which write wins?

This is where Raft comes in.

Why Raft?

Distributed consensus has been a solved problem since Paxos in 1989. But Paxos is notoriously difficult to understand and implement correctly. The original paper is dense, and real-world implementations require significant extensions that aren’t well-specified.

Raft was designed in 2014 specifically to be understandable. It provides the same guarantees as Paxos but breaks the problem into manageable pieces: leader election, log replication, and safety. The authors prioritized clarity over cleverness.

The key properties:

Leader-based: One node handles all writes, simplifying coordination
Strong consistency: Reads from the leader always see the latest committed data
Fault tolerant: A cluster of 2n+1 nodes survives n failures

Raft Basics

Before diving into code, let’s understand how Raft works at a high level.

Leader Election

Nodes are in one of three states: follower, candidate, or leader. Time is divided into terms - logical clocks that increase monotonically. Each term has at most one leader.

Followers expect periodic heartbeats from the leader. If a follower’s election timeout expires without hearing from a leader, it becomes a candidate and starts an election. It votes for itself and requests votes from other nodes. If it receives votes from a majority, it becomes the leader.

This ensures the cluster always makes progress as long as a majority of nodes are up.

Log Replication

The leader maintains a log of operations. When a client sends a write, the leader appends it to its log and replicates it to followers. Once a majority of nodes have the entry, it’s committed and the leader applies it to its state machine.

Each log entry has an index and a term. The commit index tracks the highest entry known to be committed. Followers apply entries up to this index to their state machines.

Safety

Raft guarantees that if a log entry is committed, all future leaders will have that entry. This is enforced during elections - a candidate can only win if its log is at least as up-to-date as the majority.

Using hashicorp/raft

Implementing Raft correctly is hard. Edge cases around network partitions, timing, and log compaction are subtle. For production use, battle-tested libraries are the way to go.

HashiCorp’s raft library powers Consul, Nomad, and Vault. It handles the consensus protocol, and we just need to implement the state machine.

import "github.com/hashicorp/raft"

The key interface is FSM (Finite State Machine):

type FSM interface {
    Apply(*raft.Log) interface{}
    Snapshot() (FSMSnapshot, error)
    Restore(io.ReadCloser) error
}

Apply: Called when a log entry is committed - this is where we update our KV store
Snapshot: Returns a point-in-time snapshot for log compaction
Restore: Rebuilds state from a snapshot

Implementing the FSM

Let’s adapt our KV store to work as a Raft state machine. First, we define command types:

type CommandType uint8

const (
    CommandSet CommandType = iota
    CommandDelete
)

type Command struct {
    Type  CommandType
    Key   string
    Value string
}

Now the FSM implementation:

type KVStoreFSM struct {
    mu   sync.RWMutex
    data map[string]string
}

func NewKVStoreFSM() *KVStoreFSM {
    return &KVStoreFSM{
        data: make(map[string]string),
    }
}

func (f *KVStoreFSM) Apply(log *raft.Log) interface{} {
    var cmd Command
    if err := json.Unmarshal(log.Data, &cmd); err != nil {
        return err
    }

    f.mu.Lock()
    defer f.mu.Unlock()

    switch cmd.Type {
    case CommandSet:
        f.data[cmd.Key] = cmd.Value
    case CommandDelete:
        delete(f.data, cmd.Key)
    }

    return nil
}

func (f *KVStoreFSM) Get(key string) (string, bool) {
    f.mu.RLock()
    defer f.mu.RUnlock()
    val, ok := f.data[key]
    return val, ok
}

Note that Apply is the only way to mutate state. Raft guarantees it’s called in the same order on all nodes.

Snapshots prevent the log from growing forever. When the log gets too large, Raft takes a snapshot and discards old entries:

func (f *KVStoreFSM) Snapshot() (raft.FSMSnapshot, error) {
    f.mu.RLock()
    defer f.mu.RUnlock()

    // Clone the data
    data := make(map[string]string, len(f.data))
    for k, v := range f.data {
        data[k] = v
    }

    return &kvSnapshot{data: data}, nil
}

type kvSnapshot struct {
    data map[string]string
}

func (s *kvSnapshot) Persist(sink raft.SnapshotSink) error {
    err := func() error {
        b, err := json.Marshal(s.data)
        if err != nil {
            return err
        }
        if _, err := sink.Write(b); err != nil {
            return err
        }
        return sink.Close()
    }()

    if err != nil {
        sink.Cancel()
    }
    return err
}

func (s *kvSnapshot) Release() {}

Restore rebuilds state from a snapshot:

func (f *KVStoreFSM) Restore(rc io.ReadCloser) error {
    defer rc.Close()

    var data map[string]string
    if err := json.NewDecoder(rc).Decode(&data); err != nil {
        return err
    }

    f.mu.Lock()
    f.data = data
    f.mu.Unlock()

    return nil
}

Cluster Setup

Setting up a Raft cluster requires configuring storage, transport, and the initial cluster membership.

func setupRaft(nodeID string, addr string, fsm raft.FSM, bootstrap bool) (*raft.Raft, error) {
    config := raft.DefaultConfig()
    config.LocalID = raft.ServerID(nodeID)

    // Storage for logs and stable state
    logStore, err := raftboltdb.NewBoltStore(filepath.Join("data", nodeID, "raft-log.db"))
    if err != nil {
        return nil, err
    }
    stableStore, err := raftboltdb.NewBoltStore(filepath.Join("data", nodeID, "raft-stable.db"))
    if err != nil {
        return nil, err
    }
    snapshotStore, err := raft.NewFileSnapshotStore(filepath.Join("data", nodeID), 3, os.Stderr)
    if err != nil {
        return nil, err
    }

    // TCP transport
    tcpAddr, err := net.ResolveTCPAddr("tcp", addr)
    if err != nil {
        return nil, err
    }
    transport, err := raft.NewTCPTransport(addr, tcpAddr, 3, 10*time.Second, os.Stderr)
    if err != nil {
        return nil, err
    }

    r, err := raft.NewRaft(config, fsm, logStore, stableStore, snapshotStore, transport)
    if err != nil {
        return nil, err
    }

    // Bootstrap the first node
    if bootstrap {
        cfg := raft.Configuration{
            Servers: []raft.Server{
                {
                    ID:      raft.ServerID(nodeID),
                    Address: raft.ServerAddress(addr),
                },
            },
        }
        r.BootstrapCluster(cfg)
    }

    return r, nil
}

To add a node to an existing cluster, the leader must add it:

func (s *Server) addPeer(nodeID, addr string) error {
    if s.raft.State() != raft.Leader {
        return errors.New("not the leader")
    }

    f := s.raft.AddVoter(raft.ServerID(nodeID), raft.ServerAddress(addr), 0, 0)
    return f.Error()
}

Handling Client Requests

Writes must go through the leader. If a client contacts a follower, we need to either forward the request or tell the client where the leader is.

type Server struct {
    raft *raft.Raft
    fsm  *KVStoreFSM
}

func (s *Server) Set(key, value string) error {
    if s.raft.State() != raft.Leader {
        return errors.New("not the leader")
    }

    cmd := Command{Type: CommandSet, Key: key, Value: value}
    data, err := json.Marshal(cmd)
    if err != nil {
        return err
    }

    f := s.raft.Apply(data, 10*time.Second)
    return f.Error()
}

func (s *Server) Delete(key string) error {
    if s.raft.State() != raft.Leader {
        return errors.New("not the leader")
    }

    cmd := Command{Type: CommandDelete, Key: key}
    data, err := json.Marshal(cmd)
    if err != nil {
        return err
    }

    f := s.raft.Apply(data, 10*time.Second)
    return f.Error()
}

Reads are trickier. Reading from the FSM directly is fast but might return stale data if there’s a network partition and the leader has been deposed. For strong consistency, use raft.VerifyLeader() before reading:

func (s *Server) Get(key string) (string, bool, error) {
    // Ensure we're still the leader
    if err := s.raft.VerifyLeader().Error(); err != nil {
        return "", false, err
    }

    val, ok := s.fsm.Get(key)
    return val, ok, nil
}

For better performance with slightly relaxed consistency, you can skip the verification and read directly from any node’s FSM. This is useful for workloads that can tolerate reading slightly stale data.

Demo: Surviving Node Failures

Let’s see the cluster in action. Start three nodes:

# Terminal 1 - Bootstrap the first node
./kvstore --id node1 --addr localhost:9001 --http :8001 --bootstrap

# Terminal 2
./kvstore --id node2 --addr localhost:9002 --http :8002 --join localhost:9001

# Terminal 3
./kvstore --id node3 --addr localhost:9003 --http :8003 --join localhost:9001

Write some data:

curl -X PUT -d "distributed" http://localhost:8001/kv/hello

Read it from any node:

curl http://localhost:8002/kv/hello
# distributed

Now kill the leader (node1). After a brief election timeout, one of the remaining nodes becomes the new leader:

curl http://localhost:8002/kv/hello
# distributed

curl -X PUT -d "still works" http://localhost:8002/kv/hello

The cluster keeps working. When node1 comes back, it rejoins as a follower and catches up on missed writes.

Tradeoffs

Distributed consensus isn’t free. Here’s what you give up:

Latency: Every write requires a round-trip to a majority of nodes. With 5 nodes spread across data centers, that’s potentially hundreds of milliseconds per write. Compare that to single-digit microseconds for an in-memory write.

Availability during partitions: With Raft’s strong consistency, if a majority of nodes can’t communicate, the cluster stops accepting writes. This is the CAP theorem in action - we chose consistency over availability.

Complexity: More moving parts means more ways to fail. Network issues, clock skew, and split-brain scenarios all need careful handling.

For many use cases, these tradeoffs are worth it. Losing a few hundred milliseconds per write is acceptable when the alternative is data loss or hours of downtime during failover.

Conclusion

We took our single-node KV store and distributed it across multiple nodes using Raft. The cluster now survives node failures while maintaining strong consistency - clients always see their writes after acknowledgment, even if the leader changes.

The hashicorp/raft library handles the hard parts: leader election, log replication, and safety guarantees. We just implemented the FSM interface to apply commands and manage snapshots.

Potential improvements:

Read replicas: Serve eventually-consistent reads from followers to scale read throughput
Linearizable reads: Use ReadIndex for consistent reads without leader verification overhead
Batching: Collect multiple client writes and replicate them in a single Raft entry
Membership changes: Dynamic cluster resizing without downtime

The interface stayed the same - clients don’t need to know they’re talking to a distributed system. That’s the power of good abstractions.

Building a Simple Key-Value Store in Go

2023-01-15T00:00:00+00:00

Key-value stores are everywhere - from Redis caching your sessions to etcd coordinating your Kubernetes cluster. I wanted to understand how they work under the hood, so I built one from scratch. This post covers the fundamentals: an in-memory store with thread safety, persistence via a write-ahead log, and a simple HTTP API.

This is part 1 of a series. In the next post, we’ll distribute this store across multiple nodes using Raft consensus.

The Interface

Let’s start with the basic operations every KV store needs:

type KVStore interface {
    Get(key string) (string, bool)
    Set(key, value string) error
    Delete(key string) error
    Close() error
}

Simple and familiar. Get returns the value and whether the key exists. Set and Delete can return errors because we’ll add persistence later. Close handles cleanup.

In-Memory Implementation

The simplest implementation uses a map with a mutex for thread safety:

type MemStore struct {
    mu   sync.RWMutex
    data map[string]string
}

func NewMemStore() *MemStore {
    return &MemStore{
        data: make(map[string]string),
    }
}

func (s *MemStore) Get(key string) (string, bool) {
    s.mu.RLock()
    defer s.mu.RUnlock()
    val, ok := s.data[key]
    return val, ok
}

func (s *MemStore) Set(key, value string) error {
    s.mu.Lock()
    defer s.mu.Unlock()
    s.data[key] = value
    return nil
}

func (s *MemStore) Delete(key string) error {
    s.mu.Lock()
    defer s.mu.Unlock()
    delete(s.data, key)
    return nil
}

func (s *MemStore) Close() error {
    return nil
}

We use RWMutex instead of Mutex so multiple readers can access the store concurrently. Writes still require exclusive access.

This works, but there’s a problem: all data is lost when the process exits. For a real database, we need persistence.

Adding Persistence with a Write-Ahead Log

A write-ahead log (WAL) is a simple but powerful technique: before modifying in-memory state, write the operation to disk. On restart, replay the log to reconstruct state.

First, let’s define our log entry format:

type LogEntry struct {
    Op    string `json:"op"`    // "set" or "delete"
    Key   string `json:"key"`
    Value string `json:"value,omitempty"`
}

Now the persistent store:

type PersistentStore struct {
    mu      sync.RWMutex
    data    map[string]string
    logFile *os.File
    encoder *json.Encoder
}

func NewPersistentStore(path string) (*PersistentStore, error) {
    s := &PersistentStore{
        data: make(map[string]string),
    }

    // Replay existing log if present
    if err := s.replay(path); err != nil {
        return nil, err
    }

    // Open log file for appending
    f, err := os.OpenFile(path, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644)
    if err != nil {
        return nil, err
    }
    s.logFile = f
    s.encoder = json.NewEncoder(f)

    return s, nil
}

The replay function reads the log and applies each operation:

func (s *PersistentStore) replay(path string) error {
    f, err := os.Open(path)
    if os.IsNotExist(err) {
        return nil // No log yet, start fresh
    }
    if err != nil {
        return err
    }
    defer f.Close()

    decoder := json.NewDecoder(f)
    for {
        var entry LogEntry
        if err := decoder.Decode(&entry); err == io.EOF {
            break
        } else if err != nil {
            return err
        }

        switch entry.Op {
        case "set":
            s.data[entry.Key] = entry.Value
        case "delete":
            delete(s.data, entry.Key)
        }
    }
    return nil
}

The write operations now log before modifying state:

func (s *PersistentStore) Set(key, value string) error {
    s.mu.Lock()
    defer s.mu.Unlock()

    // Write to log first
    entry := LogEntry{Op: "set", Key: key, Value: value}
    if err := s.encoder.Encode(entry); err != nil {
        return err
    }
    if err := s.logFile.Sync(); err != nil {
        return err
    }

    s.data[key] = value
    return nil
}

func (s *PersistentStore) Delete(key string) error {
    s.mu.Lock()
    defer s.mu.Unlock()

    entry := LogEntry{Op: "delete", Key: key}
    if err := s.encoder.Encode(entry); err != nil {
        return err
    }
    if err := s.logFile.Sync(); err != nil {
        return err
    }

    delete(s.data, key)
    return nil
}

The Sync() call is crucial - it forces the OS to flush the write to disk. Without it, data could be lost if the system crashes before the OS buffer is flushed.

Get and Close are straightforward:

func (s *PersistentStore) Get(key string) (string, bool) {
    s.mu.RLock()
    defer s.mu.RUnlock()
    val, ok := s.data[key]
    return val, ok
}

func (s *PersistentStore) Close() error {
    return s.logFile.Close()
}

HTTP API

Let’s expose the store via REST endpoints:

type Server struct {
    store KVStore
}

func NewServer(store KVStore) *Server {
    return &Server{store: store}
}

func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) {
    key := strings.TrimPrefix(r.URL.Path, "/kv/")
    if key == "" {
        http.Error(w, "key required", http.StatusBadRequest)
        return
    }

    switch r.Method {
    case http.MethodGet:
        s.handleGet(w, key)
    case http.MethodPut:
        s.handlePut(w, r, key)
    case http.MethodDelete:
        s.handleDelete(w, key)
    default:
        http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
    }
}

func (s *Server) handleGet(w http.ResponseWriter, key string) {
    val, ok := s.store.Get(key)
    if !ok {
        http.Error(w, "not found", http.StatusNotFound)
        return
    }
    w.Write([]byte(val))
}

func (s *Server) handlePut(w http.ResponseWriter, r *http.Request, key string) {
    body, err := io.ReadAll(r.Body)
    if err != nil {
        http.Error(w, err.Error(), http.StatusBadRequest)
        return
    }
    if err := s.store.Set(key, string(body)); err != nil {
        http.Error(w, err.Error(), http.StatusInternalServerError)
        return
    }
    w.WriteHeader(http.StatusOK)
}

func (s *Server) handleDelete(w http.ResponseWriter, key string) {
    if err := s.store.Delete(key); err != nil {
        http.Error(w, err.Error(), http.StatusInternalServerError)
        return
    }
    w.WriteHeader(http.StatusOK)
}

Wire it up in main:

func main() {
    store, err := NewPersistentStore("kv.log")
    if err != nil {
        log.Fatal(err)
    }
    defer store.Close()

    server := NewServer(store)
    http.Handle("/kv/", server)

    log.Println("listening on :8080")
    log.Fatal(http.ListenAndServe(":8080", nil))
}

Now you can interact with it:

# Set a value
curl -X PUT -d "world" http://localhost:8080/kv/hello

# Get it back
curl http://localhost:8080/kv/hello
# world

# Delete it
curl -X DELETE http://localhost:8080/kv/hello

Benchmarks

How much does persistence cost? I benchmarked both implementations with 10,000 sequential writes:

| Implementation | Writes/sec | Latency (p99) |
|----------------|------------|---------------|
| In-Memory      | 2,847,000  | 1.2µs         |
| With WAL       | 12,400     | 890µs         |

The WAL adds ~200x overhead. That’s the cost of fsync - waiting for the disk to confirm the write. In practice, you can batch writes or use a faster disk (NVMe helps significantly).

For reads, both implementations are identical since they just read from the in-memory map.

What’s Next

We have a working single-node KV store with durability. But what happens when that node fails? All your data becomes unavailable.

In the next post, we’ll distribute this store across multiple nodes using Raft consensus. Each write will be replicated to a majority of nodes before being acknowledged, so the system survives node failures. We’ll cover:

Leader election
Log replication
Consistency guarantees
The tradeoffs between availability and consistency

The interface will stay the same - clients won’t know they’re talking to a distributed system.

Raft Consensus in Go - Part 2: Log Replication and Commit

2021-05-06T00:00:00+00:00

In Part 1, we covered how Raft servers elect a leader through term-based voting. Once Raft has a leader, the hard part is done. The leader’s job is simple: replicate its log to followers and tell them when entries are safe to apply. Followers accept log entries from the leader and apply them in order.

This post covers log replication - how the leader keeps followers in sync, handles failures, and decides when a log entry is “committed” (safe to apply).

The full implementation is available at github.com/navgeet/raft.

The Log

The log is the core of Raft - it’s an ordered sequence of entries that will be applied to the state machine:

type LogEntry struct {
    // The term when this entry was added
    Term uint64

    // The index of this entry in the log (1-based)
    Index uint64

    // The actual command/data
    Command []byte
}

type Log interface {
    // Add an entry to the log
    Append(entry *LogEntry) error

    // Get entry at index
    Get(index uint64) (*LogEntry, error)

    // Remove entries from index onward
    Delete(index uint64) error

    // Last index in the log
    LastIndex() uint64

    // Last term in the log
    LastTerm() uint64
}

The leader appends new entries to its log. Followers receive these entries and append them too. Once a majority has an entry, it’s “committed.”

AppendEntries RPC

The leader sends AppendEntries RPCs to replicate log entries:

type AppendEntriesRequest struct {
    LeaderID string
    Term uint64

    // Previous entry info (for log consistency check)
    PrevLogIndex uint64
    PrevLogTerm uint64

    // Entries to replicate
    Entries []*LogEntry

    // Leader's commit index
    LeaderCommit uint64
}

type AppendEntriesResponse struct {
    Term uint64
    Success bool
    ConflictIndex uint64  // For faster conflict resolution
}

The key is PrevLogIndex and PrevLogTerm. Before appending new entries, the leader tells the follower what the previous entry was. If the follower doesn’t have an entry at that index with that term, it rejects the request. This ensures log consistency.

Replicating Entries

When a client sends a command to the leader, the leader appends it to its log and replicates it:

func (r *Raft) Apply(operation []byte) (*OperationResponse, error) {
    r.mu.Lock()

    // Only leader can accept operations
    if r.state != Leader {
        return nil, errNotLeader
    }

    // Append to our log
    entry := &LogEntry{
        Term: r.currentTerm,
        Index: r.log.LastIndex() + 1,
        Command: operation,
    }

    if err := r.log.Append(entry); err != nil {
        return nil, err
    }

    // Note the index so we can respond to client later
    index := entry.Index
    r.mu.Unlock()

    // Replicate to followers
    r.replicate()

    // Wait for entry to be committed
    return r.waitForCommit(index), nil
}

The leader waits for the entry to be replicated to a majority of servers before responding to the client. It does this by checking the commitIndex in a loop.

The Replicate Loop

On each heartbeat, the leader sends log entries to followers:

func (r *Raft) replicate() {
    for id, peer := range r.peers {
        if id == r.id {
            continue
        }

        go r.replicateToPeer(peer)
    }
}

func (r *Raft) replicateToPeer(peer Peer) {
    r.mu.Lock()

    id := peer.ID()
    nextIndex := r.nextIndex[id]

    // Get entries to send, starting at nextIndex
    prevLogIndex := nextIndex - 1
    var prevLogTerm uint64
    if prevLogIndex > 0 {
        prevEntry, _ := r.log.Get(prevLogIndex)
        prevLogTerm = prevEntry.Term
    }

    // Get log entries starting at nextIndex
    entries := r.log.GetRange(nextIndex, r.log.LastIndex())

    request := AppendEntriesRequest{
        LeaderID: r.id,
        Term: r.currentTerm,
        PrevLogIndex: prevLogIndex,
        PrevLogTerm: prevLogTerm,
        Entries: entries,
        LeaderCommit: r.commitIndex,
    }

    r.mu.Unlock()

    response, err := peer.AppendEntries(request)
    if err != nil {
        return
    }

    r.mu.Lock()
    defer r.mu.Unlock()

    if response.Term > r.currentTerm {
        r.currentTerm = response.Term
        r.state = Follower
        r.persist()
        return
    }

    if response.Success {
        // Replication succeeded, update our tracking
        newMatchIndex := prevLogIndex + uint64(len(request.Entries))
        r.matchIndex[id] = newMatchIndex
        r.nextIndex[id] = newMatchIndex + 1
    } else {
        // Replication failed, try with earlier entries
        // This is log recovery - the follower's log diverged
        r.nextIndex[id] = response.ConflictIndex
    }
}

The nextIndex tracks, for each follower, what we should send next. If replication succeeds, we advance it. If it fails, we backtrack and try again.

Handling AppendEntries on the Follower

A follower receives AppendEntries and decides whether to accept the entries:

func (r *Raft) handleAppendEntries(request AppendEntriesRequest) AppendEntriesResponse {
    r.mu.Lock()
    defer r.mu.Unlock()

    response := AppendEntriesResponse{Term: r.currentTerm}

    // Reset election timeout - we got a message from leader
    r.lastContact = time.Now()

    // If request term is lower, reject
    if request.Term < r.currentTerm {
        return response
    }

    // Update term and become follower if needed
    if request.Term > r.currentTerm {
        r.currentTerm = request.Term
        r.votedFor = ""
        r.state = Follower
        r.persist()
    }

    // Check log consistency: do we have the previous entry?
    if request.PrevLogIndex > 0 {
        prevEntry, err := r.log.Get(request.PrevLogIndex)
        if err != nil || prevEntry.Term != request.PrevLogTerm {
            // Log divergence - we don't have the expected previous entry
            // Return the conflicting index for faster recovery
            response.ConflictIndex = r.log.LastIndex()
            return response
        }
    }

    // Append new entries
    for i, entry := range request.Entries {
        index := request.PrevLogIndex + uint64(i) + 1

        // Check if entry already exists
        existing, err := r.log.Get(index)
        if err == nil && existing.Term == entry.Term {
            // Already have this entry, skip
            continue
        }

        // Entry doesn't exist or has different term - delete everything after
        r.log.Delete(index)
        r.log.Append(entry)
    }

    // Update commit index if leader has committed further
    if request.LeaderCommit > r.commitIndex {
        r.commitIndex = min(request.LeaderCommit, r.log.LastIndex())
        r.applyCond.Broadcast()  // Wake up apply loop
    }

    response.Success = true
    return response
}

The key moment is checking the previous entry. If the follower doesn’t have it, the logs have diverged and replication fails. The leader will back off and try with earlier entries until it finds one they both have.

Log Recovery

Log recovery handles the case where the leader and a follower’s logs diverge. This happens when:

A server was the leader but crashed
It had uncommitted entries on its log
It restarted and a new leader was elected
The new leader might not have those uncommitted entries

// When AppendEntries fails with ConflictIndex:

// Fast recovery: binary search for matching prefix
r.nextIndex[id] = response.ConflictIndex

// Next attempt will have earlier prevLogIndex
// Keep backing off until we find a matching entry

Instead of backing off one entry at a time (slow), the response includes ConflictIndex to speed up recovery. In practice, this makes recovery go from O(n) to O(log n).

Committing Entries

The tricky part: when is an entry “safe to apply”? An entry is committed when the leader knows it’s been replicated to a majority.

The leader tracks this using matchIndex - for each follower, the highest index it has replicated:

func (r *Raft) updateCommitIndex() {
    if r.state != Leader {
        return
    }

    // Count how many replicas have each log index
    for index := r.commitIndex + 1; index <= r.log.LastIndex(); index++ {
        count := 1  // Count ourselves

        for _, matchIdx := range r.matchIndex {
            if matchIdx >= index {
                count++
            }
        }

        // If majority has this index, it's committed
        if count > len(r.peers)/2 {
            entry, _ := r.log.Get(index)

            // Only commit entries from current term
            // (see Raft paper for why this is important)
            if entry.Term == r.currentTerm {
                r.commitIndex = index
            }
        }
    }
}

The leader runs this check periodically (usually on heartbeat). Once commitIndex advances, it tells followers in the next AppendEntries RPC, and they apply the entries.

The Apply Loop

Both leader and followers have an apply loop that applies committed entries to the state machine:

func (r *Raft) applyLoop() {
    for r.state != Shutdown {
        r.mu.Lock()

        // Wait until there are entries to apply
        for r.lastApplied >= r.commitIndex && r.state != Shutdown {
            r.applyCond.Wait()
        }

        // Apply all committed but unapplied entries
        for r.lastApplied < r.commitIndex {
            r.lastApplied++

            entry, _ := r.log.Get(r.lastApplied)

            r.mu.Unlock()

            // Apply to state machine (without holding lock)
            response := r.fsm.Apply(entry.Command)

            r.mu.Lock()

            // Send response to client
            r.operationResponseCh <- OperationResponse{
                Term: entry.Term,
                Index: entry.Index,
                Operation: entry.Command,
                Response: response,
            }
        }

        r.mu.Unlock()
    }
}

The loop is synchronized with commitIndex updates via applyCond (a condition variable). When commitIndex advances, the apply loop wakes up and applies all committed entries.

The important detail: release the lock before applying to the state machine. Holding the Raft lock while applying is a deadlock risk if the state machine tries to call back into Raft.

What Can Go Wrong

Uncommitted entries: If you respond to a client before commitIndex reaches their entry’s index, the client might see the change and then lose it if the leader crashes. The client will see conflicting behavior.

Out-of-order application: If you apply entries out of order, the state machine will be inconsistent. Always apply in index order.

Lost updates: If nextIndex never advances because replication keeps failing, the leader never knows the entry is replicated and never commits it. The client times out waiting. This is why the ConflictIndex optimization matters - it lets log recovery succeed quickly.

Split brain: If two servers think they’re leader, they could both be appending entries and committing different values. But the election logic from Part 1 prevents this. A server can only become leader with a majority vote, so only one “leader” can exist at a time.

The replication logic is where Raft shines. It handles log divergence, crashes mid-replication, and network partitions. The leader and followers stay in sync despite failures.

Raft Consensus in Go - Part 1: Elections and Terms

2021-04-02T00:00:00+00:00

Raft is a consensus algorithm designed to be simpler than Paxos. You have a cluster of servers that need to agree on a sequence of commands, even if some servers fail. One server becomes the leader, tells the others what to do, and if the leader dies, the rest elect a new one.

This post walks through building a Raft implementation in Go. We’ll start with the state management and election logic - the trickiest part of Raft is getting leaders elected correctly.

The full implementation is available at github.com/navgeet/raft.

The Core Struct

A Raft server needs to track a bunch of state:

type Raft struct {
    id string

    // Persistent state (must survive crashes)
    currentTerm uint64
    votedFor string
    log Log  // List of log entries

    // Volatile state
    state State  // Leader, Follower, or Candidate
    commitIndex uint64  // Highest log entry known to be committed
    lastApplied uint64  // Highest log entry applied to state machine

    // Leader state (only on leader)
    nextIndex map[string]uint64  // Next log index to send to each peer
    matchIndex map[string]uint64  // Highest log index replicated on each peer

    // Timing and networking
    lastContact time.Time
    peers map[string]Peer

    // Synchronization
    mu sync.Mutex
}

The key insight is splitting state into “persistent” and “volatile.” If the server crashes, currentTerm and votedFor must survive (written to disk). Everything else can be reconstructed.

Terms: Raft’s Clock

In Raft, time is divided into “terms.” Each term has at most one leader. Think of terms as logical clock ticks.

const defaultElectionTimeout = 150 * time.Millisecond
const defaultHeartbeatInterval = 50 * time.Millisecond

// A follower waits for a heartbeat from the leader.
// If none arrives within the election timeout, it becomes a candidate.
// The election timeout is randomized to avoid split votes:
electionTimeout := time.Duration(rand.Intn(150) + 150) * time.Millisecond

The timeout being randomized is crucial. If all servers had the same timeout, they’d all start elections at the same time and deadlock.

The Election Loop

Every server runs the election loop in a background goroutine:

func (r *Raft) electionLoop() {
    for r.state != Shutdown {
        r.mu.Lock()

        // If we're not getting heartbeats from the leader...
        timeSinceContact := time.Since(r.lastContact)

        // ...and we're a follower, start an election
        if r.state == Follower && timeSinceContact > electionTimeout {
            r.becomeCandidate()
        }

        r.mu.Unlock()

        time.Sleep(10 * time.Millisecond)
    }
}

func (r *Raft) becomeCandidate() {
    r.currentTerm++
    r.state = Candidate
    r.votedFor = r.id  // Vote for ourselves
    r.persist()  // Write to disk

    // Ask all peers for votes
    r.election()
}

When a follower’s election timeout fires, it increments its term and asks other servers for their votes. This is the moment of truth - will this server become the leader?

Requesting Votes

The candidate sends RequestVote RPCs to all peers:

type RequestVoteRequest struct {
    CandidateID string
    Term uint64
    LastLogIndex uint64
    LastLogTerm uint64
}

type RequestVoteResponse struct {
    Term uint64
    VoteGranted bool
}

Notice the LastLogIndex and LastLogTerm. A server won’t vote for a candidate unless that candidate’s log is at least as up-to-date as its own. This prevents electing a leader that doesn’t have all committed entries.

func (r *Raft) election() {
    request := RequestVoteRequest{
        CandidateID: r.id,
        Term: r.currentTerm,
        LastLogIndex: r.log.LastIndex(),
        LastLogTerm: r.log.LastTerm(),
    }

    votes := 1  // We vote for ourselves
    quorum := len(r.peers)/2 + 1

    for id, peer := range r.peers {
        if id == r.id {
            continue  // Don't vote ourselves
        }

        go func(peer Peer) {
            response, err := peer.RequestVote(request)
            if err != nil {
                return
            }

            r.mu.Lock()
            defer r.mu.Unlock()

            // Update term if peer has higher term
            if response.Term > r.currentTerm {
                r.currentTerm = response.Term
                r.state = Follower
                r.votedFor = ""
                r.persist()
                return
            }

            // Count the vote
            if response.VoteGranted && response.Term == r.currentTerm {
                votes++

                // If we have a quorum, we're the leader
                if votes >= quorum {
                    r.becomeLeader()
                }
            }
        }(peer)
    }
}

The trick is handling term updates. If a server gets a response with a higher term, it immediately steps down to follower. This prevents “split brain” where multiple servers think they’re the leader.

Handling Vote Requests

On the receiving end, a server needs to decide whether to grant a vote:

func (r *Raft) handleRequestVote(request RequestVoteRequest) RequestVoteResponse {
    r.mu.Lock()
    defer r.mu.Unlock()

    response := RequestVoteResponse{Term: r.currentTerm}

    // If request term is lower, reject
    if request.Term < r.currentTerm {
        return response
    }

    // If request term is higher, update ours and become follower
    if request.Term > r.currentTerm {
        r.currentTerm = request.Term
        r.votedFor = ""
        r.state = Follower
        r.persist()
    }

    // Check if we've already voted this term
    if r.votedFor != "" && r.votedFor != request.CandidateID {
        return response  // Already voted for someone else
    }

    // Check if candidate's log is up to date
    if request.LastLogTerm < r.log.LastTerm() {
        return response  // Candidate's log is behind
    }

    if request.LastLogTerm == r.log.LastTerm() &&
        request.LastLogIndex < r.log.LastIndex() {
        return response  // Same term but candidate's log is shorter
    }

    // Grant the vote
    r.votedFor = request.CandidateID
    r.persist()
    response.VoteGranted = true
    return response
}

The log comparison is critical. A server will only vote for a candidate if that candidate’s log is at least as up-to-date. This ensures that a leader has all previously committed entries.

Becoming Leader

Once a candidate gets a majority, it becomes leader:

func (r *Raft) becomeLeader() {
    r.state = Leader

    // Initialize nextIndex: for each peer, the index of the next log entry
    // to send to it. Start with our last log index + 1.
    for id := range r.peers {
        if id != r.id {
            r.nextIndex[id] = r.log.LastIndex() + 1
            r.matchIndex[id] = 0
        }
    }

    // Send initial empty AppendEntries as heartbeat
    // This announces the new leadership to followers
    r.heartbeat()
}

The nextIndex and matchIndex maps are crucial for log replication (next post). The leader tracks, for each follower, what log entries it has replicated.

The Heartbeat Loop

The leader needs to continuously send heartbeats to prevent other servers from timing out and starting elections:

func (r *Raft) heartbeatLoop() {
    ticker := time.NewTicker(50 * time.Millisecond)

    for range ticker.C {
        r.mu.Lock()

        // Only the leader sends heartbeats
        if r.state == Leader {
            r.heartbeat()
        }

        r.mu.Unlock()
    }
}

func (r *Raft) heartbeat() {
    for id, peer := range r.peers {
        if id == r.id {
            continue
        }

        go func(peer Peer) {
            // AppendEntries with no entries is a heartbeat
            request := AppendEntriesRequest{
                LeaderID: r.id,
                Term: r.currentTerm,
                LeaderCommit: r.commitIndex,
                // prevLogIndex, prevLogTerm, entries all empty for heartbeat
            }

            response, err := peer.AppendEntries(request)
            if err != nil {
                return
            }

            r.mu.Lock()
            defer r.mu.Unlock()

            // If peer has higher term, step down
            if response.Term > r.currentTerm {
                r.currentTerm = response.Term
                r.state = Follower
                r.persist()
            }
        }(peer)
    }
}

Heartbeats keep followers from timing out. If a leader crashes, heartbeats stop and followers start elections.

Resetting the Timeout

When a follower receives a heartbeat from the leader, it resets its election timeout:

func (r *Raft) handleAppendEntries(request AppendEntriesRequest) AppendEntriesResponse {
    r.mu.Lock()
    defer r.mu.Unlock()

    // Got a message from leader, reset timeout
    r.lastContact = time.Now()

    // ... rest of logic
}

This is simple but essential. The lastContact time is checked in the election loop to decide if it’s time to start an election.

Common Pitfalls

Race conditions on state transitions: Make sure all state reads/writes are protected by the mutex. It’s easy to read state without holding the lock and have it change mid-operation.

Term confusion: Always update your term if you see a higher one. If you don’t, you’ll have multiple “leaders” believing they’re in charge.

Split votes: If the election timeout isn’t randomized, all servers start elections simultaneously and deadlock.

Persistent state: If you forget to persist currentTerm and votedFor, a restarted server can vote twice in the same term, breaking the election guarantee.

The election logic handles the first challenge of distributed consensus: who gets to decide what happens. Once leaders are elected reliably, the next challenge is replicating decisions to all servers. That’s next.

Optimizing Merge Sort in Go: From Naive to Parallel

2019-09-15T00:00:00+00:00

I was practicing parallel merge sort for interview preparation and wanted to actually benchmark it to see the performance hit of a naive implementation. What started as a quick exercise turned into a rabbit hole of memory optimization, goroutine management, and some subtle bugs that took a while to figure out.

Along the way, I learned why the textbook implementation is deceptively inefficient, how to properly manage buffers to avoid unnecessary copies, and why my first two attempts at parallelization failed spectacularly (one was slow, the other deadlocked). Here’s what I found.

The Naive Implementation

Let’s start with the textbook version:

func MergeSortNaive(arr []int) []int {
    if len(arr) <= 1 {
        return arr
    }

    mid := len(arr) / 2
    left := MergeSortNaive(arr[:mid])
    right := MergeSortNaive(arr[mid:])

    return mergeNaive(left, right)
}

func mergeNaive(left, right []int) []int {
    result := make([]int, 0, len(left)+len(right))
    i, j := 0, 0

    for i < len(left) && j < len(right) {
        if left[i] <= right[j] {
            result = append(result, left[i])
            i++
        } else {
            result = append(result, right[j])
            j++
        }
    }

    result = append(result, left[i:]...)
    result = append(result, right[j:]...)
    return result
}

This implementation is clean and readable, but it has a problem: it allocates a new slice at every level of recursion. For an array of n elements, this means O(n) allocations, creating significant GC pressure.

Optimization 1: Pre-allocated Buffers

Instead of allocating at each level, we can allocate a single buffer upfront and reuse it throughout the sort:

func MergeSortSequential(arr []int) []int {
    if len(arr) <= 1 {
        return arr
    }

    buf := make([]int, len(arr))
    copy(buf, arr)

    return sequentialMergeSort(arr, buf)
}

But there’s a subtlety here. If we always merge from src into dst, we’d need to copy data back after each merge. The solution is to alternate buffers at each recursion level:

func sequentialMergeSort(src, dst []int) []int {
    if len(src) <= 1 {
        return src
    }

    mid := len(src) / 2

    // Recurse with swapped roles
    sequentialMergeSort(dst[:mid], src[:mid])
    sequentialMergeSort(dst[mid:], src[mid:])

    // Merge from dst back into src
    merge(dst[:mid], dst[mid:], src)
    return src
}

Why Alternating Buffers Work

At each level:

Children sort their portions and write results to their dst
Parent reads from those locations (now its src) and merges into its dst
No extra copies needed - data is always in the right place

Level 2: merge src → dst     (sorted data now in dst)
Level 1: merge dst → src     (reads from dst where sorted data is!)
Level 0: merge src → dst     (final result in dst)

Optimization 2: Parallelism

Merge sort is naturally parallelizable - the two recursive calls are independent. But naive parallelism has pitfalls.

Failed Attempt: Unlimited Goroutines

// DON'T DO THIS
go func() { left = MergeSort(arr[:mid]) }()
go func() { right = MergeSort(arr[mid:]) }()

This spawns 2^depth goroutines - exponential growth that overwhelms the scheduler.

Failed Attempt: Semaphore

// DON'T DO THIS EITHER
sem <- struct{}{}  // acquire
go func() {
    defer func() { <-sem }()  // release
    left = MergeSort(arr[:mid])
}()

This causes deadlock: parent goroutines hold semaphore slots while waiting for children that also need slots.

The Solution: Depth-Limited Parallelism

We only need log2(NumCPU) levels of parallelism to saturate all cores:

func MergeSort(arr []int) []int {
    buf := make([]int, len(arr))
    copy(buf, arr)

    // Calculate depth: log2(NumCPU) levels
    depth := 0
    for n := runtime.NumCPU(); n > 1; n /= 2 {
        depth++
    }

    return parallelMergeSort(arr, buf, depth)
}

func parallelMergeSort(src, dst []int, depth int) []int {
    if len(src) <= 1 {
        return src
    }

    mid := len(src) / 2

    if depth <= 0 {
        return sequentialMergeSort(src, dst)
    }

    var wg sync.WaitGroup
    wg.Add(1)

    // Spawn ONE goroutine, do other half in current goroutine
    go func() {
        defer wg.Done()
        parallelMergeSort(dst[:mid], src[:mid], depth-1)
    }()

    parallelMergeSort(dst[mid:], src[mid:], depth-1)
    wg.Wait()

    merge(dst[:mid], dst[mid:], src)
    return src
}

Key insights:

Only spawn one goroutine per level (not two) - current goroutine handles the other half
Depth limit prevents goroutine explosion
Still need to alternate buffers correctly

Benchmark Results

Testing on a 10th Gen Intel Core i5-10600K (12 threads):

Execution Time

| Size | Naive | Sequential | Parallel |
|------|-------|------------|----------|
| 1K | 56µs | 13µs | 33µs |
| 10K | 895µs | 143µs | 88µs |
| 100K | 12.7ms | 1.7ms | 536µs |
| 1M | 109ms | 15.2ms | 6.0ms |
| 10M | 1.07s | 238ms | 58ms |

Memory Allocations

| Size | Naive | Sequential | Parallel |
|------|-------|------------|----------|
| 1K | 999 | 1 | 31 |
| 10K | 9,999 | 1 | 31 |
| 100K | 100,011 | 1 | 31 |
| 1M | 1,000,025 | 1 | 32 |
| 10M | 10,000,031 | 2 | 32 |

Speedup

At 10M elements:

Parallel vs Sequential: 4.1x faster
Parallel vs Naive: 18.6x faster

Key Takeaways

Pre-allocation matters: Reducing allocations from O(n) to O(1) gives ~7x speedup even without parallelism.
Alternating buffers avoid copies: By swapping src/dst roles at each level, sorted data is always where the parent expects it.
Limit parallelism depth: You only need log2(NumCPU) levels of goroutines. More just adds overhead.
Spawn one goroutine, not two: The current goroutine can handle one half while a new goroutine handles the other.
Parallel overhead exists: For small inputs (1K), the sequential pre-allocated version is faster than parallel due to goroutine overhead.

Querying Git History with Datalog - Part 3: Advanced Queries and Analytics

2018-03-17T00:00:00+00:00

In Part 1, we covered the schema. In Part 2, we built the sync engine. Now let’s actually use this thing - what kind of questions can we answer with git data in Datomic?

The full code is at github.com/navgeet/muramasa.

File History

Git’s native commands for file history are awkward. git log -- filename shows commits that touched a file, but extracting structured data from the output is painful. With Datalog, it’s straightforward:

;; All commits that touched README.md
(d/q '[:find ?msg ?time
       :in $ ?filename
       :where [?c :git/type :git.types/commit]
              [?c :git.commit/msg ?msg]
              [?c :git.commit/time ?time]
              [?c :git.commit/tree ?tree]
              [?tree :git.tree/nodes ?node]
              [?node :git.node/filename ?file]
              [?file :file/name ?filename]
       :order-by [[?time :desc]]]
     db
     "README.md")

This walks the graph: commit → tree → node → file. Each step is a join. Datalog handles it.

Want to know when a file was added or deleted?

;; Find when README.md first appeared
(d/q '[:find (min ?time) .
       :in $ ?filename
       :where [?c :git/type :git.types/commit]
              [?c :git.commit/time ?time]
              [?c :git.commit/tree ?tree]
              [?tree :git.tree/nodes ?node]
              [?node :git.node/filename ?file]
              [?file :file/name ?filename]]
     db
     "README.md")
;; => #inst "2016-04-15T10:23:00.000-00:00"

;; Find when it was last modified
(d/q '[:find (max ?time) .
       :in $ ?filename
       :where [?c :git/type :git.types/commit]
              [?c :git.commit/time ?time]
              [?c :git.commit/tree ?tree]
              [?tree :git.tree/nodes ?node]
              [?node :git.node/filename ?file]
              [?file :file/name ?filename]]
     db
     "README.md")

Commit Patterns

Analyze commit activity over time. When do people commit most? What days of the week?

;; Commits per month
(d/q '[:find ?month (count ?c)
       :where [?c :git/type :git.types/commit]
              [?c :git.commit/time ?time]
              [(clojure.instant/format-date "yyyy-MM" ?time) ?month]]
     db)
;; => [["2024-01" 42] ["2024-02" 38] ...]

;; Commits by day of week
(d/q '[:find ?dow (count ?c)
       :where [?c :git/type :git.types/commit]
              [?c :git.commit/time ?time]
              [(clojure.instant/format-date "E" ?time) ?dow]]
     db)
;; => [["Mon" 98] ["Tue" 112] ["Fri" 45] ...]

The [(clojure.instant/format-date ...)] is a Datalog function call. You can call arbitrary Clojure functions in queries.

Repository Structure

What does the codebase look like? How many files, what types?

;; Total unique files that have ever existed
(d/q '[:find (count ?f) .
       :where [?f :file/name _]]
     db)
;; => 247

;; Files by extension
(d/q '[:find ?ext (count ?f)
       :where [?f :file/name ?name]
              [(re-find #"\\.([^.]+)$" ?name) [_ ?ext]]]
     db)
;; => [["clj" 42] ["md" 5] ["txt" 3] ...]

The re-find extracts file extensions using regex. Again, Datalog lets you call Clojure functions directly.

Finding Deleted Files

Git doesn’t make it easy to find files that used to exist but were deleted. With Muramasa:

;; Files that exist in any commit
(def all-files
  (d/q '[:find ?name
         :where [?f :file/name ?name]]
       db))

;; Files in the most recent commit
(def current-commit-sha
  (d/q '[:find ?sha .
         :where [?c :git/type :git.types/commit]
                [?c :git/sha ?sha]
                [?c :git.commit/time ?time]
         :order-by [[?time :desc]]
         :limit 1]
       db))

(def current-files
  (d/q '[:find ?name
         :in $ ?sha
         :where [?c :git/sha ?sha]
                [?c :git.commit/tree ?tree]
                [?tree :git.tree/nodes ?node]
                [?node :git.node/filename ?file]
                [?file :file/name ?name]]
       db
       current-commit-sha))

;; Difference = deleted files
(clojure.set/difference (set all-files) (set current-files))
;; => #{["old_config.yml"] ["deprecated.clj"] ...}

This is tedious with git commands. With Datalog, it’s a set operation.

Commit Message Analysis

Fulltext search is built into the schema:

;; Find all bug fixes
(d/q '[:find ?msg ?time
       :where [?c :git/type :git.types/commit]
              [?c :git.commit/msg ?msg]
              [?c :git.commit/time ?time]
              [(fulltext $ :git.commit/message "fix") [[?c]]]]
     db)

;; Commits mentioning specific issues
(d/q '[:find ?msg ?time
       :where [?c :git/type :git.types/commit]
              [?c :git.commit/msg ?msg]
              [?c :git.commit/time ?time]
              [(re-find #"#\d+" ?msg)]]
     db)
;; Finds commits with "#123" style issue references

You could build a dashboard showing which issues were addressed in which commits, when bugs were fixed, what features were added in each release.

Tree Similarity

Find commits with similar file sets:

(defn files-in-commit [db sha]
  (d/q '[:find ?name
         :in $ ?sha
         :where [?c :git/sha ?sha]
                [?c :git.commit/tree ?tree]
                [?tree :git.tree/nodes ?node]
                [?node :git.node/filename ?file]
                [?file :file/name ?name]]
       db
       sha))

(defn commit-similarity [db sha1 sha2]
  (let [files1 (set (files-in-commit db sha1))
        files2 (set (files-in-commit db sha2))
        intersection (clojure.set/intersection files1 files2)
        union (clojure.set/union files1 files2)]
    (/ (count intersection) (count union))))  ;; Jaccard similarity

;; Find commits similar to a given commit
(defn similar-commits [db target-sha threshold]
  (let [all-commits (d/q '[:find ?sha
                           :where [?c :git/type :git.types/commit]
                                  [?c :git/sha ?sha]]
                         db)]
    (filter
     (fn [[sha]]
       (> (commit-similarity db target-sha sha) threshold))
     all-commits)))

This could identify refactoring commits, merges that brought in similar changes, or releases with comparable scope.

Object Type Distribution

Understand repository composition:

;; Count objects by type
(d/q '[:find ?type (count ?e)
       :where [?e :git/type ?type]]
     db)
;; => [[:git.types/commit 142]
;;     [:git.types/tree 580]
;;     [:git.types/blob 1240]
;;     [:git.types/node 2450]]

;; Average tree size (number of nodes per tree)
(d/q '[:find (avg ?count) .
       :where [?t :git/type :git.types/tree]
              [(count ?t :git.tree/nodes) ?count]]
     db)
;; => 4.2

Time-Based Queries

Datomic is temporal - you can query the database “as of” a specific time. But for git, this is even more useful because commits have timestamps:

;; What files existed on 2024-01-01?
(d/q '[:find ?name
       :in $ ?date
       :where [?c :git/type :git.types/commit]
              [?c :git.commit/time ?time]
              [(<= ?time ?date)]
              [?c :git.commit/tree ?tree]
              [?tree :git.tree/nodes ?node]
              [?node :git.node/filename ?file]
              [?file :file/name ?name]
       ;; Get the most recent commit before date
       :order-by [[?time :desc]]
       :limit 1]
     db
     #inst "2024-01-01T00:00:00")

This reconstructs the repository state at a point in time. You could build a “time machine” viewer that shows what the repo looked like at any historical moment.

What This Enables

The killer feature isn’t any single query - it’s that you can answer questions you didn’t anticipate. With git log, you’re limited to what the commands expose. With Datalog, you have the full data model.

Want to find commits that added files matching a regex pattern, authored in a specific month, that touched fewer than 5 files total? That’s a Datalog query. Want to track how file count grew over time, segmented by directory? Datalog. Want to identify commits that might be “big refactorings” based on number of files touched and lack of new files? Datalog.

The limit is your SQL/Datalog skills, not git’s interface.

Performance

Muramasa isn’t optimized for huge repos. Syncing the Linux kernel (1M+ commits) would take a while. But for most projects (hundreds to low thousands of commits), sync is under a minute and queries are instant.

The bottleneck is usually parsing objects from JGit, not Datomic. Blob persistence to disk can be skipped for faster sync if you don’t need file contents.

For analytics, you sync once, query many times. That tradeoff works.

What’s Missing

Entity references for tree/parent relationships are currently disabled (see Part 2). Re-enabling them would allow queries like “find all files reachable from commit X” by walking tree → subtree → blob references.

Author/committer information isn’t captured. Adding :git.commit/author would enable “who touched this file” queries.

Branch/tag support would let you query “what’s on main vs what’s on feature-branch.”

The foundation is there. These are extensions, not fundamental rethinks.

Querying Git History with Datalog - Part 2: Parsing Git Objects with JGit

2018-02-20T00:00:00+00:00

In Part 1, we covered the Datomic schema for storing git objects. Now we need to actually parse those objects out of a git repository. This means dealing with JGit, Java’s git implementation, and walking the git object graph.

The challenge: git objects reference each other by SHA. A commit points to a tree SHA, trees contain blob SHAs. We need to resolve all these references and load everything into Datomic in the right order.

JGit Basics

JGit represents git objects with Java classes:

(import '[org.eclipse.jgit.revwalk RevCommit RevTree]
        '[org.eclipse.jgit.lib Repository ObjectId]
        '[org.eclipse.jgit.treewalk TreeWalk])

;; Load a repository
(def repo (load-repo "/path/to/repo"))

;; Get all commits
(def commits (rev-list repo))  ;; Returns seq of RevCommit objects

Each RevCommit is a handle to the commit object. To get its data, you call methods:

(.getId commit)          ;; ObjectId (SHA)
(.getShortMessage commit)  ;; First line of commit message
(.getFullMessage commit)   ;; Full commit message
(.getCommitTime commit)    ;; Unix timestamp
(.getParents commit)       ;; Array of parent commits
(.getTree commit)          ;; RevTree object

The Protocol Approach

Rather than dealing with JGit classes directly, we define a protocol for git objects:

(defprotocol IGitObject
  (parse [object repo]
    "Parse a JGit object into a Clojure record")

  (nodes [object]
    "Returns list of child object SHAs for graph traversal")

  (serialize [object]
    "Serializes object into Datomic entity map"))

This lets us treat all git objects uniformly - commits, trees, blobs all implement the same interface.

Records for Git Objects

We define Clojure records to hold parsed data:

(defrecord Commit [sha msg message time tree parents])
(defrecord Tree [sha nodes])
(defrecord Blob [sha uri])
(defrecord Node [type sha filename mode])

These are plain Clojure data - no JGit dependencies. Once parsed, we work with these records instead of JGit objects.

Parsing Commits

Implementing the protocol for commits:

(extend-protocol IGitObject
  RevCommit
  (parse [commit repo]
    (let [sha (.name (.getId commit))
          msg (.getShortMessage commit)
          message (.getFullMessage commit)
          time (java.util.Date. (* 1000 (.getCommitTime commit)))
          tree (.name (.getId (.getTree commit)))
          parents (map #(.name (.getId %)) (.getParents commit))]
      (->Commit sha msg message time tree (vec parents))))

  (nodes [commit]
    ;; Return the tree SHA - we need to parse the tree next
    [(.name (.getId (.getTree commit)))])

  (serialize [commit]
    {:git/sha (:sha commit)
     :git/type :git.types/commit
     :git.commit/msg (:msg commit)
     :git.commit/message (:message commit)
     :git.commit/time (:time commit)}))

Notice nodes returns the tree SHA. This is how we know what to parse next. When walking the object graph, we start with commits, then follow the SHAs in nodes to parse trees and blobs.

Parsing Trees

Trees are trickier. A tree contains multiple entries (nodes), each pointing to either another tree or a blob:

(extend-protocol IGitObject
  RevTree
  (parse [tree repo]
    (let [sha (.name (.getId tree))
          walk (TreeWalk/forPath (.newObjectReader repo) "" tree)]
      (when walk
        (.setRecursive walk false)  ;; Don't recurse into subtrees
        (let [nodes (loop [acc []]
                      (if (.next walk)
                        (recur
                         (conj acc
                               (->Node
                                (if (.isSubtree walk)
                                  :git.types/tree
                                  :git.types/blob)
                                (.name (.getObjectId walk 0))
                                (.getNameString walk)
                                (.getModeOctal walk))))
                        acc))]
          (->Tree sha nodes)))))

  (nodes [tree]
    ;; Return SHAs of all child nodes
    (map :sha (:nodes tree)))

  (serialize [tree]
    {:git/sha (:sha tree)
     :git/type :git.types/tree
     :git.tree/nodes (mapv serialize-node (:nodes tree))}))

TreeWalk is JGit’s API for iterating tree entries. We call .next repeatedly to walk through all entries, creating a Node record for each.

Parsing Blobs

Blobs are simple - they’re just content. In Muramasa, we store a URI instead of the raw bytes:

(extend-protocol IGitObject
  ObjectId  ;; Blobs are referenced by ObjectId, not RevBlob
  (parse [object-id repo]
    (let [sha (.name object-id)
          loader (.open (.newObjectReader repo) object-id)
          bytes (.getBytes loader)
          uri (write-blob-to-disk sha bytes *blob-dir*)]
      (->Blob sha uri)))

  (nodes [object-id]
    [])  ;; Blobs have no children

  (serialize [blob]
    {:git/sha (:sha blob)
     :git/type :git.types/blob
     :git.blob/uri (:uri blob)}))

write-blob-to-disk stores the blob content on disk with a directory structure:

(defn write-blob-to-disk [sha bytes blob-dir]
  (let [prefix1 (subs sha 0 2)
        prefix2 (subs sha 2 4)
        dir (io/file blob-dir prefix1 prefix2)
        file (io/file dir sha)]
    (.mkdirs dir)
    (with-open [out (io/output-stream file)]
      (.write out bytes))
    (str "file://" (.getAbsolutePath file))))

This creates paths like blobs/ab/cd/abcd1234..., similar to how git stores objects internally.

Walking the Object Graph

The sync process walks the entire object graph, starting from commits:

(defn walk-objects [repo commits db]
  (loop [to-visit (set (map #(.name (.getId %)) commits))
         visited #{}
         objects {}]

    (if (empty? to-visit)
      objects  ;; Done, return all parsed objects

      (let [sha (first to-visit)
            remaining (disj to-visit sha)]

        ;; Skip if already visited or in database
        (if (or (visited sha) (db-has-sha? db sha))
          (recur remaining (conj visited sha) objects)

          ;; Parse the object
          (let [obj-id (ObjectId/fromString sha)
                parsed (parse obj-id repo)
                child-shas (nodes parsed)]

            ;; Add parsed object and queue children
            (recur
             (into remaining child-shas)
             (conj visited sha)
             (assoc objects sha parsed))))))))

This is a breadth-first walk. We start with commit SHAs, parse each commit, extract its tree SHA, parse the tree, extract blob SHAs from the tree, parse blobs. The visited set prevents cycles and duplicate work.

Dependency Order

Git objects have dependencies:

Blobs have no dependencies
Trees depend on blobs and other trees
Commits depend on trees and parent commits

When transacting to Datomic, we need to ensure dependencies exist first. Otherwise, a commit referencing a tree that doesn’t exist yet will fail.

(defn prepare-transaction [objects]
  (let [blobs (filter #(instance? Blob (val %)) objects)
        trees (filter #(instance? Tree (val %)) objects)
        commits (filter #(instance? Commit (val %)) objects)]

    ;; Serialize in dependency order
    (vec (concat
          (map #(serialize (val %)) blobs)
          (map #(serialize (val %)) trees)
          (map #(serialize (val %)) commits)))))

By transacting blobs first, then trees, then commits, we ensure all references are valid.

Handling References

Entity references in Datomic use lookup refs - [:git/sha "abc123"] means “the entity whose :git/sha is abc123.”

When serializing a commit, we need to convert SHA strings to lookup refs:

(defn prepare-references [objects]
  (let [commits (filter #(instance? Commit (val %)) objects)]
    (mapcat
     (fn [[sha commit]]
       (concat
        ;; Add tree reference
        (when-let [tree-sha (:tree commit)]
          [{:db/id [:git/sha sha]
            :git.commit/tree [:git/sha tree-sha]}])

        ;; Add parent references
        (map (fn [parent-sha]
               {:db/id [:git/sha sha]
                :git.commit/parents [:git/sha parent-sha]})
             (:parents commit))))
     commits)))

This creates a second transaction that adds the references after entities exist. It’s a two-pass approach: first create entities, then wire them together.

The Complete Flow

Putting it all together:

(defn sync! [conn repo-path]
  (let [repo (load-repo repo-path)
        db (d/db conn)]

    ;; 1. Get all commits
    (let [commits (rev-list repo)

          ;; 2. Walk object graph, parse everything
          objects (walk-objects repo commits db)

          ;; 3. Prepare entities in dependency order
          entities (prepare-transaction objects)

          ;; 4. Transact entities
          _ @(d/transact conn entities)

          ;; 5. Add references
          refs (prepare-references objects)
          _ @(d/transact conn refs)]

      {:commits-synced (count commits)
       :objects-synced (count objects)})))

Next time: advanced queries and what you can actually do with git data in Datomic.

Querying Git History with Datalog - Part 1: Why and How

2018-01-13T00:00:00+00:00

Git history is a graph database. Commits point to trees, trees contain blobs, commits reference parent commits. But git’s native query interface is terrible - you’re stuck with shell commands and grep.

What if you could query git history with a real query language? Not git log | grep, but actual relational queries. “Show me all commits that touched file X and were authored by person Y in date range Z.” Or “find files that exist in commit A but not in commit B.”

That’s what Muramasa does. It syncs a git repository into Datomic, letting you query git history with Datalog.

Why Datomic?

Datomic is a database where everything is immutable. Perfect for git, where nothing ever changes - commits are eternal, trees are content-addressed, the SHA is the identity.

Datomic also has Datalog, a query language that’s basically SQL but for graphs. Git history is a graph. The match is natural.

The Schema

A git repository has four object types: commits, trees, blobs, and tags. Muramasa models these as Datomic entities:

;; Every git object has a type and SHA
{:db/ident :git/type
 :db/valueType :db.type/keyword
 :db/cardinality :db.cardinality/one}

{:db/ident :git/sha
 :db/valueType :db.type/string
 :db/cardinality :db.cardinality/one
 :db/unique :db.unique/identity}  ;; SHA is unique

The :db/unique :db.unique/identity is key. It means we can upsert by SHA - if we try to insert an object that already exists, Datomic merges it. This makes incremental sync trivial.

Commits

Commits have messages, timestamps, and references to parents and tree:

{:db/ident :git.commit/msg
 :db/valueType :db.type/string
 :db/cardinality :db.cardinality/one}

{:db/ident :git.commit/message  ;; Full message
 :db/valueType :db.type/string
 :db/cardinality :db.cardinality/one
 :db/fulltext true}  ;; Enable fulltext search

{:db/ident :git.commit/time
 :db/valueType :db.type/instant
 :db/cardinality :db.cardinality/one}

{:db/ident :git.commit/tree
 :db/valueType :db.type/ref  ;; Reference to tree entity
 :db/cardinality :db.cardinality/one}

{:db/ident :git.commit/parents
 :db/valueType :db.type/ref  ;; References to parent commits
 :db/cardinality :db.cardinality/many}

:db.type/ref is how you model relationships. :git.commit/tree points to another entity (the tree). This is the power - you can traverse the graph in queries.

Trees and Blobs

Trees contain nodes (file entries), blobs are file content:

{:db/ident :git.tree/nodes
 :db/valueType :db.type/ref
 :db/cardinality :db.cardinality/many
 :db/isComponent true}  ;; Nodes are owned by trees

{:db/ident :git.node/filename
 :db/valueType :db.type/ref  ;; Reference to file entity
 :db/cardinality :db.cardinality/one}

{:db/ident :git.blob/uri
 :db/valueType :db.type/string
 :db/cardinality :db.cardinality/one}

{:db/ident :file/name
 :db/valueType :db.type/string
 :db/cardinality :db.cardinality/one
 :db/unique :db.unique/identity
 :db/fulltext true}  ;; Search filenames

The sync! Function

The main API is a single function - sync!:

(defn sync! [conn repo-path]
  (let [repo (load-repo repo-path)
        db (d/db conn)]

    ;; Ensure schema exists
    (ensure-schema! conn)

    ;; Collect all commits
    (let [commits (rev-list repo)
          objects (parse-objects repo commits db)]

      ;; Transact in dependency order
      (transact-objects! conn objects)

      {:commits-synced (count commits)
       :objects-synced (count objects)})))

The process:

Load the git repo using JGit
Get all commits (git log --all)
Parse each commit and its referenced objects (tree, blobs)
Filter out objects already in the database (by SHA)
Transact to Datomic

Incremental Sync

Because SHAs are unique identities, sync is idempotent:

(defn db-has-sha? [db sha]
  (boolean
   (d/q '[:find ?e .
          :in $ ?sha
          :where [?e :git/sha ?sha]]
        db sha)))

Before parsing an object, we check if it exists. If it does, skip it. This means you can run sync! repeatedly and it only adds new commits.

Basic Queries

Once synced, you can query with Datalog:

;; Count total commits
(d/q '[:find (count ?c) .
       :where [?c :git/type :git.types/commit]]
     db)
;; => 142

;; Get recent commits with messages
(d/q '[:find ?msg ?time
       :where [?c :git/type :git.types/commit]
              [?c :git.commit/msg ?msg]
              [?c :git.commit/time ?time]
       :order-by [[?time :desc]]
       :limit 10]
     db)

;; Find all unique filenames
(d/q '[:find ?name
       :where [?f :file/name ?name]]
     db)

;; Fulltext search on commit messages
(d/q '[:find ?msg
       :where [?c :git.commit/message ?msg]
              [(fulltext $ :git.commit/message "bugfix") [[?c]]]]
     db)

The :where clause is a pattern. [?c :git/type :git.types/commit] means “find entities where the :git/type attribute equals :git.types/commit, bind the entity to ?c.”

Variables (prefixed with ?) unify across clauses. If ?c appears in multiple clauses, it must be the same entity in all of them.

Why This Matters

Git’s query capabilities are limited. Want to find commits in a date range that touched a specific file? You’re writing shell scripts. Want to analyze commit patterns over time? Export to CSV and load into pandas.

With Muramasa, it’s just Datalog:

;; Commits in date range touching specific file
(d/q '[:find ?msg ?time
       :in $ ?start ?end ?filename
       :where [?c :git/type :git.types/commit]
              [?c :git.commit/msg ?msg]
              [?c :git.commit/time ?time]
              [(>= ?time ?start)]
              [(<= ?time ?end)]
              [?c :git.commit/tree ?tree]
              [?tree :git.tree/nodes ?node]
              [?node :git.node/filename ?file]
              [?file :file/name ?filename]]
     db
     #inst "2024-01-01"
     #inst "2024-12-31"
     "README.md")

The query walks the graph - from commit to tree to node to file. Datalog handles the joins.

Next time: how we parse git objects with JGit and handle the object graph.

First blog post

2014-05-13T00:00:00+00:00

So I finally bit the bullet and published my first blog post, however little sense it may make.

The raison d’être of this blog is to document my journey into the software world, mostly remaining in the comforting clojure zone, and a recent foray in the fascinating field of artificial intelligence.

I’ve lately been researching Inductive Logic Programming (aka ILP), a fascinating technique to induce first-order hypothesis from first-order examples. Much like what Machine Learning achieves for propositional examples, albeit being much less powerful than ILP, and consequently much simpler. I hope to document some of my learnings in this blog over the next year.

Till next time.

Occam's razor

Vibe Coding to Production

How I Actually Vibe Code

What’s Usually Missing

The Fixes I Actually Apply

Tests, But Not Everywhere

Error Handling That Helps

Logging That’s Actually Useful

Security Stuff I Keep Forgetting

Config

Making It Deployable

The Actual Checklist

The Point

Building a Computer-Use Agent in Python - Part 2: Multi-Agent Architecture

Why Multi-Agent?

The Boss-Worker Pattern

BossAgent: The Orchestrator

Agent Creation and Reuse

Browser Automation with Playwright

Setup

XPath-Based Actions

System Prompt

Action Dispatch

Context Management

Context Passing Between Agents

Context Compaction

Exit Summaries

Full Orchestration

Real-Time Updates

Tradeoffs and Extensions

What Works

What’s Tricky

Extensions

Building a Computer-Use Agent in Python - Part 1: The Agent Loop

The Core Loop

System Control Tools

Mouse Control with Xlib

Keyboard with pyautogui

Screenshot Capture

Window Management

The Base Agent Class

The GUI Agent

System Prompt

Verification Loop

Context Compaction

Running It

Next Steps

Fine-Tuning Qwen3 0.6B for Chess Move Legality

Why Chess Move Legality?

The Dataset

Data Composition

Training Setup

Hyperparameters

Chat Template

Code

Training Observations

Inference

Evaluation

What This Enables

Limitations

Next Steps

The Digital Immune System: Building an AI-Powered Security Monitor in Go

Architecture

Data Structures

Collectors

Process Collector

Network Collector

Filesystem Collector

Auth Log Collector

Anomaly Detection with ML

AI Reasoning with Ollama

Remediation Engine

Main Loop

Running It

Tradeoffs

Platform Engineering for Small Teams: Building a Lightweight Internal Developer Portal

What We’re Building

The Service Catalog: Keep It Simple

Convention: The service.yaml File

Catalog Generation

Convention: The `service.yaml` File