Files
stack/mosaic-knowledge-SKILL-spec.md

26 KiB

name, description
name description
mosaic-knowledge Interact with the Mosaic knowledge layer API for persistent storage and semantic retrieval of findings, agent memory, and conversation archives. Use when: (1) saving important discoveries or insights as findings, (2) searching past findings with vector similarity, (3) storing/retrieving agent-specific key-value memory, (4) archiving conversation sessions for future reference, (5) semantically searching archived conversations. Triggers: "save finding", "remember this", "search findings", "what do we know about", "archive this conversation", "search archives", "my memory", "store in mosaic".

Mosaic Knowledge Layer Skill

Enable AI agents (Jarvis, Builder, etc.) to read and write to the Mosaic knowledge layer via OpenClaw's skill system. This provides persistent, semantically searchable storage for findings, agent memory, and conversation archives.


When to Use This Skill

Trigger conditions (check description for full list):

Scenario Action Example Trigger
Agent discovers something worth preserving save_finding "Save this as a finding", "Record this insight"
Agent needs to recall past discoveries search_findings "What do we know about X?", "Search findings for..."
Agent wants to persist key-value state save_memory "Remember that the API key is X", "Store my current task"
Agent needs to recall stored state get_memory "What was I working on?", "Get my memory keys"
Archiving a completed conversation archive_conversation "Archive this session", "Save this conversation"
Searching past conversations search_archives "When did we discuss X?", "Search conversation archives"

Authentication

Required Environment Variable

MOSAIC_API_TOKEN=<jwt-bearer-token>

The token must be a valid JWT obtained from Mosaic auth. Store securely and never hardcode in skill files.

Token Acquisition

Agents should obtain tokens through the Mosaic authentication flow (out of scope for this skill). Tokens typically have an expiration and may need refresh.

Using the Token

All API calls require the Authorization header:

Authorization: Bearer <MOSAIC_API_TOKEN>

API Configuration

Setting Value
Base URL https://mosaic-api.woltje.com
Auth Header Authorization: Bearer <token>
Content-Type application/json
Timeout 30 seconds (recommended)

Actions

1. save_finding

Store a finding with automatic embedding generation for semantic search.

When to use: Preserving discoveries, insights, decisions, or important information that should be retrievable later via semantic search.

API Endpoint: POST /findings

Request Body:

{
  "title": "string (required)",
  "content": "string (required)",
  "workspaceId": "string (required)",
  "tags": ["string"] // optional
}

cURL Example:

curl -X POST "https://mosaic-api.woltje.com/findings" \
  -H "Authorization: Bearer $MOSAIC_API_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "title": "Redis connection pool configuration",
    "content": "The production Redis instance requires a connection pool of at least 20 connections. Lower values cause timeout errors under load.",
    "workspaceId": "ws-prod-123",
    "tags": ["redis", "configuration", "production"]
  }'

Success Response (201):

{
  "id": "finding-abc123",
  "title": "Redis connection pool configuration",
  "content": "The production Redis instance requires a connection pool of at least 20 connections...",
  "workspaceId": "ws-prod-123",
  "tags": ["redis", "configuration", "production"],
  "embeddingGenerated": true,
  "createdAt": "2026-02-28T12:00:00Z"
}

Agent Usage Pattern:

# Jarvis saving an important discovery
def save_important_discovery(title: str, content: str, workspace_id: str, tags: list[str] = None):
    """Save a finding to the Mosaic knowledge layer."""
    payload = {
        "title": title,
        "content": content,
        "workspaceId": workspace_id
    }
    if tags:
        payload["tags"] = tags
    
    response = requests.post(
        "https://mosaic-api.woltje.com/findings",
        headers={"Authorization": f"Bearer {os.environ['MOSAIC_API_TOKEN']}"},
        json=payload,
        timeout=30
    )
    response.raise_for_status()
    return response.json()

2. search_findings

Perform vector similarity search over stored findings.

When to use: Recalling past discoveries, looking up related information, finding contextually relevant findings.

API Endpoint: GET /findings/search

Query Parameters:

Parameter Type Required Description
query string Yes Natural language search query
workspaceId string Yes Workspace to search within
limit integer No Max results (default: 10)
threshold float No Similarity threshold 0-1 (default: 0.7)

cURL Example:

curl -X GET "https://mosaic-api.woltje.com/findings/search?query=redis%20connection%20issues&workspaceId=ws-prod-123&limit=5" \
  -H "Authorization: Bearer $MOSAIC_API_TOKEN"

Success Response (200):

{
  "results": [
    {
      "id": "finding-abc123",
      "title": "Redis connection pool configuration",
      "content": "The production Redis instance requires a connection pool...",
      "tags": ["redis", "configuration", "production"],
      "similarity": 0.89,
      "createdAt": "2026-02-28T12:00:00Z"
    }
  ],
  "query": "redis connection issues",
  "totalResults": 1
}

Agent Usage Pattern:

# Jarvis searching for relevant past findings
def search_past_knowledge(query: str, workspace_id: str, limit: int = 10) -> list[dict]:
    """Search the knowledge layer for relevant findings."""
    params = {
        "query": query,
        "workspaceId": workspace_id,
        "limit": limit
    }
    
    response = requests.get(
        "https://mosaic-api.woltje.com/findings/search",
        headers={"Authorization": f"Bearer {os.environ['MOSAIC_API_TOKEN']}"},
        params=params,
        timeout=30
    )
    response.raise_for_status()
    return response.json().get("results", [])

3. save_memory

Upsert agent-specific key/value memory entries.

When to use: Persisting agent state, remembering context between sessions, storing configuration or preferences.

API Endpoint: POST /agent-memory

Request Body:

{
  "agentId": "string (required)",
  "workspaceId": "string (required)",
  "key": "string (required)",
  "value": "any (required)" // Can be string, number, object, array
}

cURL Example:

curl -X POST "https://mosaic-api.woltje.com/agent-memory" \
  -H "Authorization: Bearer $MOSAIC_API_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "agentId": "jarvis-main",
    "workspaceId": "ws-prod-123",
    "key": "current_task",
    "value": {
      "task": "Implementing RBAC migration",
      "started": "2026-02-28T10:00:00Z",
      "status": "in_progress"
    }
  }'

Success Response (200):

{
  "agentId": "jarvis-main",
  "workspaceId": "ws-prod-123",
  "key": "current_task",
  "value": {
    "task": "Implementing RBAC migration",
    "started": "2026-02-28T10:00:00Z",
    "status": "in_progress"
  },
  "updatedAt": "2026-02-28T12:30:00Z"
}

Agent Usage Pattern:

# Jarvis storing its current state
def remember(key: str, value: any, agent_id: str, workspace_id: str) -> dict:
    """Store a value in agent memory."""
    response = requests.post(
        "https://mosaic-api.woltje.com/agent-memory",
        headers={"Authorization": f"Bearer {os.environ['MOSAIC_API_TOKEN']}"},
        json={
            "agentId": agent_id,
            "workspaceId": workspace_id,
            "key": key,
            "value": value
        },
        timeout=30
    )
    response.raise_for_status()
    return response.json()

4. get_memory

Retrieve agent memory entries by agent and workspace.

When to use: Recalling stored state, checking memory keys, restoring context from previous sessions.

API Endpoint: GET /agent-memory

Query Parameters:

Parameter Type Required Description
agentId string Yes Agent identifier
workspaceId string Yes Workspace identifier
key string No Specific key to retrieve (omit for all keys)

cURL Example (all keys):

curl -X GET "https://mosaic-api.woltje.com/agent-memory?agentId=jarvis-main&workspaceId=ws-prod-123" \
  -H "Authorization: Bearer $MOSAIC_API_TOKEN"

cURL Example (specific key):

curl -X GET "https://mosaic-api.woltje.com/agent-memory?agentId=jarvis-main&workspaceId=ws-prod-123&key=current_task" \
  -H "Authorization: Bearer $MOSAIC_API_TOKEN"

Success Response (200) - All Keys:

{
  "agentId": "jarvis-main",
  "workspaceId": "ws-prod-123",
  "memory": [
    {
      "key": "current_task",
      "value": {"task": "Implementing RBAC migration", "status": "in_progress"},
      "updatedAt": "2026-02-28T12:30:00Z"
    },
    {
      "key": "preferred_model",
      "value": "claude-sonnet",
      "updatedAt": "2026-02-27T09:00:00Z"
    }
  ]
}

Success Response (200) - Specific Key:

{
  "agentId": "jarvis-main",
  "workspaceId": "ws-prod-123",
  "key": "current_task",
  "value": {
    "task": "Implementing RBAC migration",
    "started": "2026-02-28T10:00:00Z",
    "status": "in_progress"
  },
  "updatedAt": "2026-02-28T12:30:00Z"
}

Agent Usage Pattern:

# Jarvis recalling its stored memory
def recall(key: str = None, agent_id: str, workspace_id: str) -> dict | list:
    """Retrieve agent memory (specific key or all)."""
    params = {
        "agentId": agent_id,
        "workspaceId": workspace_id
    }
    if key:
        params["key"] = key
    
    response = requests.get(
        "https://mosaic-api.woltje.com/agent-memory",
        headers={"Authorization": f"Bearer {os.environ['MOSAIC_API_TOKEN']}"},
        params=params,
        timeout=30
    )
    response.raise_for_status()
    return response.json()

5. archive_conversation

Archive a conversation session for future semantic search.

When to use: End of important sessions, before context compaction, preserving decision-making context.

API Endpoint: POST /conversation-archive/ingest

Request Body:

{
  "sessionId": "string (required)",
  "workspaceId": "string (required)",
  "agentId": "string (required)",
  "messages": [
    {
      "role": "user" | "assistant" | "system",
      "content": "string",
      "timestamp": "ISO8601 datetime"
    }
  ],
  "metadata": {  // optional
    "title": "string",
    "summary": "string",
    "tags": ["string"],
    "custom": {}
  }
}

cURL Example:

curl -X POST "https://mosaic-api.woltje.com/conversation-archive/ingest" \
  -H "Authorization: Bearer $MOSAIC_API_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "sessionId": "session-xyz789",
    "workspaceId": "ws-prod-123",
    "agentId": "jarvis-main",
    "messages": [
      {
        "role": "user",
        "content": "How should we handle the Redis connection pooling?",
        "timestamp": "2026-02-28T11:00:00Z"
      },
      {
        "role": "assistant",
        "content": "Based on load testing, I recommend a minimum pool of 20 connections...",
        "timestamp": "2026-02-28T11:01:00Z"
      }
    ],
    "metadata": {
      "title": "Redis Configuration Discussion",
      "tags": ["redis", "infrastructure", "production"]
    }
  }'

Success Response (201):

{
  "id": "archive-def456",
  "sessionId": "session-xyz789",
  "workspaceId": "ws-prod-123",
  "agentId": "jarvis-main",
  "messageCount": 2,
  "embeddingGenerated": true,
  "archivedAt": "2026-02-28T12:00:00Z"
}

Agent Usage Pattern:

# Jarvis archiving a completed session
def archive_session(
    session_id: str,
    workspace_id: str,
    agent_id: str,
    messages: list[dict],
    metadata: dict = None
) -> dict:
    """Archive a conversation session to the knowledge layer."""
    payload = {
        "sessionId": session_id,
        "workspaceId": workspace_id,
        "agentId": agent_id,
        "messages": messages
    }
    if metadata:
        payload["metadata"] = metadata
    
    response = requests.post(
        "https://mosaic-api.woltje.com/conversation-archive/ingest",
        headers={"Authorization": f"Bearer {os.environ['MOSAIC_API_TOKEN']}"},
        json=payload,
        timeout=60  # Longer timeout for large archives
    )
    response.raise_for_status()
    return response.json()

6. search_archives

Perform semantic search over archived conversations.

When to use: Finding past discussions, recalling decision context, tracing the evolution of ideas.

API Endpoint: GET /conversation-archive/search

Query Parameters:

Parameter Type Required Description
query string Yes Natural language search query
workspaceId string No Filter by workspace
agentId string No Filter by agent
limit integer No Max results (default: 10)
threshold float No Similarity threshold 0-1 (default: 0.7)

cURL Example:

curl -X GET "https://mosaic-api.woltje.com/conversation-archive/search?query=redis%20pool%20configuration&workspaceId=ws-prod-123&limit=5" \
  -H "Authorization: Bearer $MOSAIC_API_TOKEN"

Success Response (200):

{
  "results": [
    {
      "id": "archive-def456",
      "sessionId": "session-xyz789",
      "title": "Redis Configuration Discussion",
      "relevantExcerpt": "Based on load testing, I recommend a minimum pool of 20 connections...",
      "similarity": 0.91,
      "archivedAt": "2026-02-28T12:00:00Z",
      "tags": ["redis", "infrastructure", "production"]
    }
  ],
  "query": "redis pool configuration",
  "totalResults": 1
}

Agent Usage Pattern:

# Jarvis searching past conversations
def search_past_conversations(
    query: str,
    workspace_id: str = None,
    agent_id: str = None,
    limit: int = 10
) -> list[dict]:
    """Search archived conversations for relevant context."""
    params = {"query": query, "limit": limit}
    if workspace_id:
        params["workspaceId"] = workspace_id
    if agent_id:
        params["agentId"] = agent_id
    
    response = requests.get(
        "https://mosaic-api.woltje.com/conversation-archive/search",
        headers={"Authorization": f"Bearer {os.environ['MOSAIC_API_TOKEN']}"},
        params=params,
        timeout=30
    )
    response.raise_for_status()
    return response.json().get("results", [])

Example Usage Patterns

Jarvis Agent Pattern (Orchestrator)

# Jarvis: Main orchestrator agent usage

class MosaicKnowledgeClient:
    """Client for Jarvis to interact with Mosaic knowledge layer."""
    
    def __init__(self, agent_id: str, workspace_id: str):
        self.agent_id = agent_id
        self.workspace_id = workspace_id
        self.base_url = "https://mosaic-api.woltje.com"
        self.token = os.environ.get("MOSAIC_API_TOKEN")
    
    def _headers(self):
        return {
            "Authorization": f"Bearer {self.token}",
            "Content-Type": "application/json"
        }
    
    # === Findings ===
    
    def save_insight(self, title: str, content: str, tags: list = None):
        """Save an important discovery or insight."""
        # Implementation uses save_finding endpoint
        pass
    
    def recall_insights(self, query: str, limit: int = 5) -> list:
        """Search for relevant past insights."""
        # Implementation uses search_findings endpoint
        pass
    
    # === Memory ===
    
    def remember(self, key: str, value: any):
        """Store a value in persistent memory."""
        # Implementation uses save_memory endpoint
        pass
    
    def recall(self, key: str = None):
        """Retrieve value(s) from memory."""
        # Implementation uses get_memory endpoint
        pass
    
    # === Archives ===
    
    def archive_session(self, messages: list, metadata: dict = None):
        """Archive the current conversation session."""
        # Implementation uses archive_conversation endpoint
        pass
    
    def search_history(self, query: str) -> list:
        """Search past archived conversations."""
        # Implementation uses search_archives endpoint
        pass


# Usage in Jarvis workflow:

# 1. On session start: recall current state
jarvis_memory = MosaicKnowledgeClient("jarvis-main", "ws-prod-123")
current_task = jarvis_memory.recall("current_task")

# 2. During work: save important discoveries
jarvis_memory.save_insight(
    title="API rate limit discovered",
    content="The payment API has a hard limit of 100 req/min",
    tags=["api", "rate-limiting", "payments"]
)

# 3. Update working state
jarvis_memory.remember("current_task", {
    "task": "RBAC migration",
    "phase": "testing",
    "blocked": False
})

# 4. Before compaction: archive important context
jarvis_memory.archive_session(messages, {
    "title": "RBAC Migration Planning",
    "summary": "Completed design phase, moving to implementation"
})

Builder Agent Pattern (Worker)

# Builder: Task-specific worker agent usage

class BuilderKnowledgeHelper:
    """Simplified client for Builder agents (task-focused)."""
    
    def __init__(self, task_id: str, workspace_id: str):
        self.task_id = task_id
        self.workspace_id = workspace_id
        self.agent_id = f"builder-{task_id}"
        self.base_url = "https://mosaic-api.woltje.com"
        self.token = os.environ.get("MOSAIC_API_TOKEN")
    
    def log_discovery(self, title: str, content: str):
        """Log a discovery relevant to the current task."""
        # Uses save_finding with task_id as tag
        pass
    
    def check_related_work(self, query: str) -> list:
        """Check if similar work has been done before."""
        # Uses search_findings
        pass
    
    def save_progress(self, state: dict):
        """Save current task progress."""
        # Uses save_memory with key "task_progress"
        pass
    
    def get_progress(self) -> dict:
        """Resume from saved progress."""
        # Uses get_memory with key "task_progress"
        pass


# Usage in Builder workflow:

builder = BuilderKnowledgeHelper("task-rbac-001", "ws-prod-123")

# Check for prior related work
related = builder.check_related_work("RBAC permission checks")
if related:
    print(f"Found {len(related)} related findings to review")

# Log discoveries as you work
builder.log_discovery(
    "Permission model structure",
    "Permissions are stored in a hierarchical tree with inheritance"
)

# Save progress for resumption
builder.save_progress({
    "completed_steps": ["design", "migration-script"],
    "current_step": "testing",
    "files_modified": ["src/auth/permissions.ts"]
})

Error Handling

HTTP Status Codes

Code Meaning Action
200 Success (GET/PUT) Proceed normally
201 Created (POST) Resource created successfully
400 Bad Request Check request body format, validate required fields
401 Unauthorized Token missing, invalid, or expired. Refresh token.
403 Forbidden Token valid but lacks permissions for this workspace/resource
404 Not Found Resource doesn't exist (for GET by ID)
409 Conflict Duplicate key (memory), unique constraint violation
422 Unprocessable Validation error - check field constraints
429 Rate Limited Back off with exponential retry
500 Server Error Retry with backoff, log for investigation
503 Service Unavailable API temporarily down, retry after delay

Error Response Format

{
  "error": {
    "code": "VALIDATION_ERROR",
    "message": "Invalid request body",
    "details": [
      {"field": "workspaceId", "message": "workspaceId is required"}
    ]
  }
}

Retry Strategy

import time
import random

def mosaic_request_with_retry(
    method: str,
    endpoint: str,
    max_retries: int = 3,
    **kwargs
) -> dict:
    """Make API request with exponential backoff retry."""
    
    for attempt in range(max_retries + 1):
        try:
            response = requests.request(
                method,
                f"https://mosaic-api.woltje.com{endpoint}",
                headers={
                    "Authorization": f"Bearer {os.environ['MOSAIC_API_TOKEN']}",
                    "Content-Type": "application/json"
                },
                timeout=30,
                **kwargs
            )
            
            # Success
            if response.status_code in (200, 201):
                return response.json()
            
            # Client error - don't retry
            if 400 <= response.status_code < 500 and response.status_code != 429:
                error_data = response.json().get("error", {})
                raise MosaicClientError(
                    code=error_data.get("code", "UNKNOWN"),
                    message=error_data.get("message", response.text),
                    details=error_data.get("details", [])
                )
            
            # Rate limit or server error - retry with backoff
            if response.status_code in (429, 500, 502, 503, 504):
                if attempt < max_retries:
                    backoff = (2 ** attempt) + random.uniform(0, 1)
                    time.sleep(backoff)
                    continue
            
            response.raise_for_status()
            
        except requests.Timeout:
            if attempt < max_retries:
                time.sleep(2 ** attempt)
                continue
            raise MosaicTimeoutError("Request timed out after retries")
        
        except requests.ConnectionError as e:
            if attempt < max_retries:
                time.sleep(2 ** attempt)
                continue
            raise MosaicConnectionError(f"Connection failed: {e}")
    
    raise MosaicMaxRetriesError("Max retries exceeded")


class MosaicClientError(Exception):
    def __init__(self, code: str, message: str, details: list = None):
        self.code = code
        self.message = message
        self.details = details or []
        super().__init__(f"[{code}] {message}")

class MosaicTimeoutError(Exception): pass
class MosaicConnectionError(Exception): pass
class MosaicMaxRetriesError(Exception): pass

Common Error Scenarios

Error Cause Resolution
AUTH_TOKEN_EXPIRED JWT has expired Obtain new token from Mosaic auth
WORKSPACE_NOT_FOUND Invalid workspaceId Verify workspace exists and user has access
VALIDATION_ERROR Missing/invalid fields Check required fields and data types
EMBEDDING_FAILED Vector embedding failed Retry; if persistent, check content length
RATE_LIMIT_EXCEEDED Too many requests Implement backoff, reduce request frequency
DUPLICATE_KEY Memory key already exists (on create) Use upsert behavior (POST is idempotent)

Best Practices

For Findings

  1. Write descriptive titles - "Redis connection pool configuration" > "Redis note"
  2. Include actionable content - What should future agents know/do?
  3. Use consistent tags - Establish tag vocabulary per workspace
  4. Link related findings - Reference finding IDs in content when applicable

For Memory

  1. Use namespaced keys - task.current, config.preferred_model, state.last_check
  2. Keep values serializable - JSON-compatible types only
  3. Set TTL expectations - Memory is persistent; consider cleanup strategies
  4. Don't store secrets - Use environment variables or secret managers

For Archives

  1. Add meaningful metadata - Title, summary, tags improve searchability
  2. Archive at natural boundaries - Task completion, context compaction, session end
  3. Include decision context - Why decisions were made, not just what
  4. Respect size limits - Very large sessions may need chunking

General

  1. Always handle auth errors - Token refresh flow is critical
  2. Implement retry logic - Network issues are transient
  3. Log failures for debugging - Include request ID if available
  4. Batch operations when possible - Reduce API calls for bulk operations
  5. Cache frequently accessed data - Reduce redundant searches

Skill Integration Notes

For OpenClaw Integration

When implementing this as an OpenClaw skill:

  1. Create helper functions that wrap the raw API calls with retry/auth logic
  2. Store MOSAIC_API_TOKEN in environment variables (never in skill files)
  3. Provide agent_id and workspace_id through skill configuration or context
  4. Consider caching for frequently accessed memory keys

Suggested Tool Functions

If exposing as MCP tools or OpenClaw functions:

# Suggested function signatures for skill tools

def mosaic_save_finding(
    title: str,
    content: str,
    tags: list[str] = None
) -> dict:
    """Save an important finding to the Mosaic knowledge layer."""
    pass

def mosaic_search_findings(
    query: str,
    limit: int = 10
) -> list[dict]:
    """Search for relevant findings using semantic similarity."""
    pass

def mosaic_remember(
    key: str,
    value: any
) -> dict:
    """Store a value in persistent agent memory."""
    pass

def mosaic_recall(
    key: str = None
) -> dict | list[dict]:
    """Retrieve value(s) from agent memory."""
    pass

def mosaic_archive_conversation(
    messages: list[dict],
    title: str = None,
    tags: list[str] = None
) -> dict:
    """Archive a conversation for future reference."""
    pass

def mosaic_search_archives(
    query: str,
    limit: int = 10
) -> list[dict]:
    """Search archived conversations semantically."""
    pass

Version History

Version Date Changes
1.0.0 2026-02-28 Initial spec for MS22-SKILL-001

Spec Author: Jarvis (OpenClaw orchestrator)
Implementation Status: Design complete, awaiting Codex worker implementation