Memory Worker Skill

Version: 2.0 (2026-03-01 Update)

Purpose

Extract important information from daily notes and store it in structured SQLite database.

CRITICAL: Schema Correction (2026.2.26+ Security)

The generated script MUST have correct SQLite schema. Check for these issues:

FIXED Schema for memory_cells

7 columns including updated_at:

cursor.execute('''
    CREATE TABLE IF NOT EXISTS memory_cells (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        scene TEXT,
        cell_type TEXT,
        salience INTEGER,
        content TEXT,
        source_file TEXT,
        created_at TEXT,
        updated_at TEXT DEFAULT CURRENT_TIMESTAMP
    )
''')

INSERT must include all 7 columns:

cursor.execute('''
    INSERT INTO memory_cells (scene, cell_type, salience, content, source_file, created_at, updated_at)
    VALUES (?, ?, ?, ?, ?, ?, ?)
''', (scene, cell_type, salience, content, source_file, timestamp, timestamp))

FIXED Schema for scenes

6 columns including high_salience_count:

cursor.execute('''
    CREATE TABLE IF NOT EXISTS scenes (
        scene TEXT PRIMARY KEY,
        summary TEXT,
        item_count INTEGER DEFAULT 0,
        high_salience_count INTEGER DEFAULT 0,
        created_at TEXT,
        updated_at TEXT DEFAULT CURRENT_TIMESTAMP
    )
''')

INSERT/UPDATE must include high_salience_count:

cursor.execute('''
    INSERT OR REPLACE INTO scenes (scene, summary, item_count, high_salience_count, created_at, updated_at)
    VALUES (?, ?, ?, ?, ?, ?)
''', (scene, summary, count, high_salience, timestamp, timestamp))

FIXED Icon Characters (No Unicode Emoji)

Use simple characters, not emojis:

Task: [x] not ✅
Maintenance: [w] not 🔧
Heartbeat: [h] not 💓
Note: [n] not 📝

Updated Template

#!/usr/bin/env python3
import sqlite3
import os
from datetime import datetime

# Configuration
DB_PATH = os.path.expanduser("~/.openclaw/memory.db")
SOURCE_FILE = "YYYY-MM-DD.md"
SCENE = "YYYY-MM-DD"

def init_db():
    conn = sqlite3.connect(DB_PATH)
    cursor = conn.cursor()
    
    # FIXED: Include updated_at column
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS memory_cells (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            scene TEXT,
            cell_type TEXT,
            salience INTEGER,
            content TEXT,
            source_file TEXT,
            created_at TEXT,
            updated_at TEXT DEFAULT CURRENT_TIMESTAMP
        )
    ''')
    
    # FIXED: Include high_salience_count column
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS scenes (
            scene TEXT PRIMARY KEY,
            summary TEXT,
            item_count INTEGER DEFAULT 0,
            high_salience_count INTEGER DEFAULT 0,
            created_at TEXT,
            updated_at TEXT DEFAULT CURRENT_TIMESTAMP
        )
    ''')
    
    conn.commit()
    return conn

def insert_cells(conn):
    cursor = conn.cursor()
    timestamp = datetime.now().isoformat()
    
    # Extract cells from daily note
    cells = [
        (SCENE, 'task', 5, 'High priority task description...'),
        (SCENE, 'maintenance', 4, 'Maintenance activity description...'),
        # ... more cells
    ]
    
    count = 0
    for scene, cell_type, salience, content in cells:
        # FIXED: Include updated_at column
        cursor.execute('''
            INSERT INTO memory_cells (scene, cell_type, salience, content, source_file, created_at, updated_at)
            VALUES (?, ?, ?, ?, ?, ?, ?)
        ''', (scene, cell_type, salience, content, SOURCE_FILE, timestamp, timestamp))
        count += 1
    
    high_salience = sum(1 for _, _, salience, _ in cells if salience >= 4)
    summary = 'Generated scene summary'
    
    # FIXED: Include high_salience_count
    cursor.execute('''
        INSERT OR REPLACE INTO scenes (scene, summary, item_count, high_salience_count, created_at, updated_at)
        VALUES (?, ?, ?, ?, ?, ?)
    ''', (SCENE, summary, count, high_salience, timestamp, timestamp))
    
    conn.commit()
    return count

def generate_report(conn):
    cursor = conn.cursor()
    
    cursor.execute('SELECT cell_type, COUNT(*) FROM memory_cells WHERE source_file = ? GROUP BY cell_type', (SOURCE_FILE,))
    type_counts = cursor.fetchall()
    
    cursor.execute('SELECT COUNT(*) FROM memory_cells')
    total = cursor.fetchone()[0]
    
    cursor.execute('SELECT summary FROM scenes WHERE scene = ?', (SCENE,))
    summary = cursor.fetchone()
    
    # FIXED: No Unicode emojis
    icons = {'task': '[x]', 'maintenance': '[w]', 'heartbeat': '[h]', 'note': '[n]'}
    
    report = f"""[Memory Worker Daily] - {SCENE}\n\n**Files Processed:** {SOURCE_FILE}\n"""
    for cell_type, cnt in type_counts:
        icon = icons.get(cell_type, '[*]')
        report += f"{icon} {cell_type}: {cnt}\n"
    
    report += f"\n**Summary:**\n{summary[0] if summary else 'N/A'}\n\n**Total:** {total} cells"
    return report

def main():
    conn = init_db()
    count = insert_cells(conn)
    report = generate_report(conn)
    conn.close()
    print(report)
    print(f"Successfully inserted {count} cells")

if __name__ == "__main__":
    main()

Testing

Before generating:

Verify schema columns match table creation
Verify INSERT contains all columns
Verify no Unicode emojis
Test run locally before committing

History

2026-03-01: Fixed schema mismatch (added updated_at and high_salience_count columns)
2026-02-20: Initial version

5.4 KiB Raw Permalink Blame History