Agent Internals

This document covers the internals of the nekoni agent: how it processes messages, how to extend it with tools and skills, and how the security layer works end-to-end.

Table of Contents

1. Request Lifecycle
2. ReAct Loop
3. Tool System
4. Skill System
5. RAG Pipeline
6. Memory
7. WebRTC + DataChannel Auth
8. DataChannel Message Protocol
9. Trace Events
10. Adding a Tool
11. Code Style

Request Lifecycle

A message travels through these layers before the user gets a response:

Mobile or Web App (DataChannel)
  │
  ▼
DataChannelHandler._handle_message()    # webrtc/channel.py
  │  auth check (_auth_complete == True)
  │  route by msg_type prefix
  ├─► rag_*   → _handle_rag()
  ├─► skill_* / cron_* → _handle_skill()
  └─► plain message → on_message callback
  ▼
main._on_datachannel_message()          # main.py
  │  keyed by clientId — supports multiple concurrent peers (mobile + web app simultaneously)
  │  look up or create SessionContext
  ▼
AgentLoop.run()                         # agent/loop.py
  │  1. RAG retrieval (top-k chunks, filtered by min score)
  │  2. Build system prompt (tools JSON + RAG context)
  │  3. LLM call (Ollama), streaming chunks back to mobile
  │  4. Parse response — tool call? → execute → loop
  │     text response? → done
  ▼
trace_manager.emit()                    # api/ws.py
  │  broadcast each step to dashboard WebSocket
  ▼
DataChannel.send(response)              # back to mobile

ReAct Loop

File: apps/agent/src/nekoni_agent/agent/loop.py

The loop implements a simplified ReAct (Reason + Act) pattern:

user message
  │
  ├─► RAG query (ChromaDB, top_k=5, min_score=0.3)
  │     → inject matching chunks into system prompt
  │
  └─► for iteration in range(max_react_iterations=8):
        │
        ├─► Ollama chat(messages)  ← streaming, chunks sent to mobile live
        │
        ├─► parse response for tool_call JSON
        │     {"tool_call": {"name": "...", "arguments": {...}}}
        │
        ├─► if tool call found:
        │     execute tool
        │     append (assistant: tool_call_json) to context
        │     append (tool: result) to context
        │     continue loop
        │
        └─► if plain text: return as final response

Tool call format the LLM must output:

{"tool_call": {"name": "get_time", "arguments": {"timezone": "UTC"}}}

The parser first tries json.loads(response), then falls back to a regex search for embedded JSON — accommodating models that add prose around the call.

Session context (agent/context.py) keeps a rolling window of the last 20 messages. System messages are pinned; oldest non-system messages are dropped when the limit is exceeded.

Tool System

Files: tools/base.py, tools/registry.py, tools/builtin/

Architecture

Tool (ABC)
  ├── name: str
  ├── description: str
  ├── parameters_schema: dict  ← JSON Schema, sent to LLM
  └── execute(**kwargs) → Any  ← called by AgentLoop

ToolRegistry
  ├── register(tool)
  ├── execute(name, **kwargs)
  └── to_json_schema() → list[dict]  ← injected into system prompt

Built-in Tools

How Tools Are Exposed to the LLM

At startup, ToolRegistry.to_json_schema() serializes all tools:

[
  {
    "name": "get_time",
    "description": "Get the current date and time.",
    "parameters": {
      "type": "object",
      "properties": {
        "timezone": {"type": "string", "description": "..."}
      },
      "required": []
    }
  }
]

This JSON is embedded directly into the system prompt so the LLM knows what it can call and how to format calls.

Skill System

Files: skills/models.py, skills/runner.py, skills/scheduler.py

Skills are named prompt templates stored in SQLite. They can be run on demand or scheduled via cron expressions. Manageable from both the web dashboard and the mobile app over WebRTC.

Architecture

Skill (SQLite row)
  ├── id: UUID
  ├── name: str
  ├── description: str
  ├── prompt: str          ← markdown prompt sent to the agent loop
  └── createdAt: int

CronJob (SQLite row)
  ├── id: UUID
  ├── skillId: UUID        ← foreign key to skill
  ├── cronExpression: str  ← standard 5-field cron (APScheduler)
  ├── enabled: bool
  ├── lastRun: int | None
  └── createdAt: int

Running a Skill

run_skill(skill_id, prompt, agent_loop, sessions, trace_cb)
  │
  ├─► load or create SessionContext (session_id = "skill:{skill_id}")
  ├─► emit skill_event trace (event="skill_run_start")
  ├─► AgentLoop.run(prompt)  ← same path as user messages
  ├─► save messages to episodic memory
  └─► emit skill_event trace (event="skill_run_complete")

Scheduler

SkillScheduler wraps APScheduler's AsyncIOScheduler. On startup it loads all enabled cron jobs from SQLite and schedules them via CronTrigger.from_crontab(). When a cron job fires, it calls run_skill() and then updates lastRun.

skill_scheduler.configure(agent_loop, sessions, trace_cb)
await skill_scheduler.start()   # loads DB, starts APScheduler

skill_scheduler.schedule_job(job)    # add/update a job
skill_scheduler.unschedule_job(id)   # remove a job

Management via REST

Management via WebRTC DataChannel

The mobile app manages skills over the authenticated DataChannel using these message types:

RAG Pipeline

Files: rag/pipeline.py, rag/embedder.py, rag/store.py

Ingest

document (file or text)
  │
  ├─► extractor: plain text, PDF via pypdf (required for PDFs)
  │
  ├─► chunker: split into ~512-token windows, 50-token overlap
  │     (approximated as chars/4)
  │
  ├─► embedder: sentence-transformers all-MiniLM-L6-v2
  │     runs in thread pool executor (non-blocking)
  │
  └─► ChromaDB upsert
        collection: "nekoni_docs"
        similarity: cosine
        persistence: data/chroma/
        metadata: {doc_id, source, chunk_index}

Document ID is sha256(text)[:16] — re-ingesting the same content is idempotent.

Supported file types: plain text, PDF (pypdf must be installed), any text-decodable format.

Query

user query string
  │
  ├─► embed with all-MiniLM-L6-v2
  │
  └─► ChromaDB query (cosine similarity, top_k=5)
        filter: score >= rag_min_score (default 0.3)
        returns: [{content (truncated to 600 chars), source, score}, ...]

The top chunks are injected into the system prompt before the LLM call:

Relevant context from your knowledge base:
[document.pdf]: chunk text here
---
[notes.txt]: another chunk

Document Management

Client Upload Protocol (WebRTC)

Used by both the mobile app and the web app. Large files are chunked into 15 KB base64 segments sent as sequential DataChannel messages:

mobile → {type: "rag_upload_start", uploadId, filename, totalChunks}
mobile → {type: "rag_upload_chunk", uploadId, index, content}  × N
agent  → reassemble → ingest_file() → {type: "rag_upload_response", doc_id, chunks}

mobile → {type: "rag_list"}   → agent → {type: "rag_list_response", documents: [...]}
mobile → {type: "rag_delete", doc_id} → agent → {type: "rag_delete_response"}

Memory

Short-term: Working Memory (memory/short.py)

An in-process bounded deque (maxlen=50). Stores typed entries for the current session.

mem = WorkingMemory(maxlen=50)
mem.add("observation", "User mentioned they like jazz")
mem.get_recent(n=10)  # last 10 entries

Long-term: Episodic Memory (memory/long.py)

SQLite-backed via aiosqlite. Stores the full message history across sessions.

CREATE TABLE episodes (
    id         INTEGER PRIMARY KEY AUTOINCREMENT,
    session_id TEXT NOT NULL,
    role       TEXT NOT NULL,       -- user / assistant / tool
    content    TEXT NOT NULL,
    timestamp  REAL NOT NULL
);

WebRTC + DataChannel Auth

Multi-peer Support

The agent maintains a separate RTCPeerConnection per connected client, keyed by clientId. Multiple devices (e.g. mobile app and web app simultaneously) can be connected at the same time. Each peer goes through its own auth handshake and maintains its own session context.

WebRTC Peer (webrtc/peer.py)

On startup the agent:

1. Loads or generates its Ed25519 identity key
2. Opens a WebSocket to the signal server and joins its fixed room
3. Waits for peer_joined events (one per connecting client)
4. On receiving an offer, creates a new RTCPeerConnection (aiortc) for that client, generates an answer, sends it back
5. On icecandidate, forwards to the correct peer via signaling
6. Once DataChannel is established, hands off to DataChannelHandler (one instance per peer)

HTTPS / TLS

The web app connects to the agent over HTTPS (:8443) to satisfy browser security requirements. A scripts/gen_cert.py generates a self-signed certificate on first run; scripts/run_agent.py starts an asyncio SSL termination proxy that accepts TLS on :8443 and pipes raw bytes to the plain HTTP uvicorn server on :8000. A single uvicorn/FastAPI instance handles all traffic — there is no second app initialisation.

All outgoing signaling messages are signed:

sig = sign_payload({k: v for k, v in msg.items()}, self._private_key)

DataChannel Handshake (webrtc/channel.py)

After the DTLS-encrypted DataChannel opens, a 4-step application-layer mutual auth runs (same for mobile and web app):

Step 1  Client → Agent   {type:"hello", pubKey, nonce_m}
Step 2  Agent  → Client  {type:"challenge", nonce_a, sig=sign(nonce_m)}
Step 3  Client → Agent   {type:"response", sig=sign(nonce_a)}
Step 4  Agent  → Client  {type:"ready"}

• Step 1: Agent checks pubKey is in approved_devices.json — if not, closes immediately
• Step 2: Agent signs client's nonce with its identity key so the client can verify it's talking to the real agent (matches QR-scanned pubKey)
• Step 3: Client signs agent's nonce — proves possession of the registered private key
• Step 4: Both sides are authenticated; normal message exchange begins

Any failure at any step sends {type:"auth_failed", reason} and closes the channel.

Signature Format

All signatures (signaling + DataChannel):

sig = base64url(Ed25519Sign(UTF8(JSON.stringify(payload_without_sig, sorted_keys)), senderPrivKey))

• Payload is JSON-serialized with sorted top-level keys only, no extra whitespace
• Signature is Ed25519 over the raw UTF-8 encoded JSON bytes (no pre-hashing)
• Encoded as base64url without padding
• Python: cryptography library, Ed25519PrivateKey.sign(payload_bytes)
• TypeScript: @noble/ed25519 v3, ed.sign(payloadBytes, privKeyBytes)
• React Native: tweetnacl, nacl.sign.detached(payloadBytes, secretKeyBytes)

DataChannel Message Protocol

All messages are JSON. After auth, the channel handles:

Chat messages

{"type": "message", "content": "...", "sessionId": "..."}

Response: streaming {"type": "chunk", "content": "..."} tokens, then {"type": "message_end"} (or {"type": "message", "content": "..."} if no streaming occurred).

Ping/pong

{"type": "ping"}  →  {"type": "pong"}

Session sync

{"type": "sync", "sessionId": "...", "afterTimestamp": 0}
→ {"type": "sync_response", "messages": [...]}

RAG operations

See Mobile Upload Protocol above.

Skill/cron operations

See Management via WebRTC DataChannel above.

Trace Events

Every step in the agent loop emits a TraceEvent broadcast to all connected dashboard WebSocket clients (/ws/traces).

interface TraceEvent {
  id: string
  sessionId: string
  timestamp: number           // Unix ms
  type: "llm_call" | "tool_call" | "rag_query" | "skill_event" | "error" | "message"
  data: Record<string, unknown>
  parentId?: string
}

Event types emitted by AgentLoop.run():

Trace sessions: The /api/traces/sessions endpoint returns distinct sessionId values with event counts and last timestamps. The dashboard Traces page can filter by session.

Adding a Tool

1. Create apps/agent/src/nekoni_agent/tools/builtin/my_tool.py:

from typing import Any
from ..base import Tool

class WeatherTool(Tool):
    @property
    def name(self) -> str:
        return "get_weather"

    @property
    def description(self) -> str:
        return "Get current weather for a city."

    @property
    def parameters_schema(self) -> dict:
        return {
            "type": "object",
            "properties": {
                "city": {
                    "type": "string",
                    "description": "City name, e.g. 'London'",
                },
            },
            "required": ["city"],
        }

    async def execute(self, city: str) -> dict:
        return {"city": city, "temp": "22°C", "condition": "sunny"}

2. Register in apps/agent/src/nekoni_agent/main.py:

from .tools.builtin.my_tool import WeatherTool
tool_registry.register(WeatherTool())

Guidelines:

• Keep execute() async — use asyncio.get_event_loop().run_in_executor() for blocking work
• Return JSON-serializable types
• Keep description concise — it goes into the LLM's context window
• Raise exceptions on failure; the loop catches them and reports an error trace event

Code Style

TypeScript / JavaScript

• Fat arrow functions: const foo = () => {} — not function foo() {}
• Named exports: export const Foo = ... — default exports only for Expo Router page files (framework requirement)
• Hooks: export const useXxx = (...) => { ... }
• React components: export const MyComponent = (props: Props) => { ... }

Prettier (.prettierrc)

{
  "printWidth": 60,
  "tabWidth": 2,
  "singleQuote": true,
  "trailingComma": "all",
  "semi": false,
  "arrowParens": "always"
}

Run: npx prettier --write "apps/**/*.{ts,tsx}"

Python (Ruff)

Configured in apps/agent/pyproject.toml. Line length 88, double quotes, Black-compatible.

# from apps/agent/
.venv/bin/ruff format src/        # format
.venv/bin/ruff check src/         # lint
.venv/bin/ruff check --fix src/   # lint + auto-fix