airkjw/claude-mem
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add comprehensive documentation for claude-mem codebase and create a test worker script

Browse Source 
- Introduced CODEMAP.md detailing project overview, architecture, directory structure, core components, commands, hooks system, SDK, services, shared components, utilities, and key workflows.
- Added a test-worker.sh script to automate testing of the SDK worker, including session creation, worker initiation, socket communication, and cleanup after finalization.
This commit is contained in:
Alex Newman
2025-10-16 13:56:18 -04:00
parent 18aa4f2538
commit 6e9be84a01
10 changed files with 2074 additions and 228 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/bin/cli.ts
+16
View File
@@ -174,6 +174,22 @@ program
summaryHook(JSON.parse(input));
});
program
.command('worker <sessionId>')
.description('Run SDK worker process (internal use)')
.action(async (sessionId: string) => {
try {
// Import and run the worker main function
const { main } = await import('../sdk/worker.js');
// Set process.argv so worker can parse sessionId
process.argv[2] = sessionId;
await main();
} catch (error: any) {
console.error(`[SDK Worker] Fatal error: ${error.message}`);
process.exit(1);
}
});
// Helper function to read stdin
async function readStdin(): Promise<string> {
return new Promise((resolve) => {
src/hooks/new.ts
+2 -20
View File
@@ -1,7 +1,6 @@
import { HooksDatabase } from '../services/sqlite/HooksDatabase.js';
import path from 'path';
import { spawn } from 'child_process';
import fs from 'fs';
export interface UserPromptSubmitInput {
session_id: string;
@@ -37,25 +36,8 @@ export function newHook(input: UserPromptSubmitInput): void {
db.close();
// Start SDK worker in background as detached process
// Try source first (development), then fall back to dist (production)
const srcWorkerPath = path.join(__dirname, '..', 'sdk', 'worker.ts');
const distWorkerPath = path.join(__dirname, '..', 'sdk', 'worker.js');
let workerPath: string;
if (fs.existsSync(srcWorkerPath)) {
workerPath = srcWorkerPath;
} else if (fs.existsSync(distWorkerPath)) {
workerPath = distWorkerPath;
} else {
// Fallback: assume we're in the bundled CLI
// In this case, we can't spawn the worker since it's bundled
// This is a limitation we'll need to address
console.error('[claude-mem] Worker not found, skipping background processing');
console.log('{"continue": true, "suppressOutput": true}');
process.exit(0);
}
const child = spawn('bun', [workerPath, sessionId.toString()], {
// Use 'claude-mem worker' CLI command which is always available
const child = spawn('claude-mem', ['worker', sessionId.toString()], {
detached: true,
stdio: 'ignore'
});
src/hooks/save.ts
+29 -20
View File
@@ -1,5 +1,7 @@
import net from 'net';
import { join } from 'path';
import { HooksDatabase } from '../services/sqlite/HooksDatabase.js';
import path from 'path';
import { PathDiscovery } from '../services/path-discovery.js';
export interface PostToolUseInput {
session_id: string;
@@ -18,11 +20,11 @@ const SKIP_TOOLS = new Set([
/**
* Save Hook - PostToolUse
* Queues tool observations for SDK processing
* Sends tool observations to worker via Unix socket
*/
export function saveHook(input: PostToolUseInput): void {
try {
const { session_id, cwd, tool_name, tool_input, tool_output } = input;
const { session_id, tool_name, tool_input, tool_output } = input;
// Skip certain tools
if (SKIP_TOOLS.has(tool_name)) {
@@ -30,38 +32,45 @@ export function saveHook(input: PostToolUseInput): void {
process.exit(0);
}
// Extract project from cwd
const project = path.basename(cwd);
// Find active SDK session
const db = new HooksDatabase();
const session = db.findActiveSDKSession(session_id);
db.close();
if (!session) {
// No active session yet - this can happen if UserPromptSubmit hasn't run
// Just exit silently
db.close();
console.log('{"continue": true, "suppressOutput": true}');
process.exit(0);
}
// Queue the observation
// SDK session ID might be null if init message hasn't arrived yet
// Use the internal ID as a fallback
const sdkSessionId = session.sdk_session_id || `pending-${session.id}`;
// Get socket path
const dataDir = PathDiscovery.getInstance().getDataDirectory();
const socketPath = join(dataDir, `worker-${session.id}.sock`);
db.queueObservation(
sdkSessionId,
// Send observation via Unix socket
const message = {
type: 'observation',
tool_name,
JSON.stringify(tool_input),
JSON.stringify(tool_output)
);
tool_input: JSON.stringify(tool_input),
tool_output: JSON.stringify(tool_output)
};
db.close();
const client = net.connect(socketPath, () => {
client.write(JSON.stringify(message) + '\n');
client.end();
});
// Output hook response
console.log('{"continue": true, "suppressOutput": true}');
process.exit(0);
client.on('error', (err) => {
// Socket not available - worker may have crashed or not started
console.error(`[claude-mem save] Socket error: ${err.message}`);
// Continue anyway, don't block Claude
});
client.on('close', () => {
console.log('{"continue": true, "suppressOutput": true}');
process.exit(0);
});
} catch (error: any) {
// On error, don't block Claude Code
src/hooks/summary.ts
+26 -14
View File
@@ -1,4 +1,7 @@
import net from 'net';
import { join } from 'path';
import { HooksDatabase } from '../services/sqlite/HooksDatabase.js';
import { PathDiscovery } from '../services/path-discovery.js';
export interface StopInput {
session_id: string;
@@ -8,7 +11,7 @@ export interface StopInput {
/**
* Summary Hook - Stop
* Signals SDK to finalize and generate summary
* Sends FINALIZE message to worker via Unix socket
*/
export function summaryHook(input: StopInput): void {
try {
@@ -17,29 +20,38 @@ export function summaryHook(input: StopInput): void {
// Find active SDK session
const db = new HooksDatabase();
const session = db.findActiveSDKSession(session_id);
db.close();
if (!session) {
// No active session - nothing to finalize
db.close();
console.log('{"continue": true, "suppressOutput": true}');
process.exit(0);
}
// Insert special FINALIZE message into observation queue
const sdkSessionId = session.sdk_session_id || `pending-${session.id}`;
// Get socket path
const dataDir = PathDiscovery.getInstance().getDataDirectory();
const socketPath = join(dataDir, `worker-${session.id}.sock`);
db.queueObservation(
sdkSessionId,
'FINALIZE',
'{}',
'{}'
);
// Send FINALIZE message via Unix socket
const message = {
type: 'finalize'
};
db.close();
const client = net.connect(socketPath, () => {
client.write(JSON.stringify(message) + '\n');
client.end();
});
// Output hook response
console.log('{"continue": true, "suppressOutput": true}');
process.exit(0);
client.on('error', (err) => {
// Socket not available - worker may have already finished or crashed
console.error(`[claude-mem summary] Socket error: ${err.message}`);
// Continue anyway, don't block Claude
});
client.on('close', () => {
console.log('{"continue": true, "suppressOutput": true}');
process.exit(0);
});
} catch (error: any) {
// On error, don't block Claude Code
src/sdk/worker.ts
+123 -29
View File
@@ -1,23 +1,39 @@
#!/usr/bin/env bun
/**
* SDK Worker Process
* Background agent that processes tool observations and generates session summaries
* Background server that processes tool observations via Unix socket
*/
import net from 'net';
import { unlinkSync, existsSync } from 'fs';
import { join } from 'path';
import { query } from '@anthropic-ai/claude-agent-sdk';
import { HooksDatabase } from '../services/sqlite/HooksDatabase.js';
import { PathDiscovery } from '../services/path-discovery.js';
import { buildInitPrompt, buildObservationPrompt, buildFinalizePrompt } from './prompts.js';
import { parseObservations, parseSummary } from './parser.js';
import type { Observation, SDKSession } from './prompts.js';
import type { SDKSession } from './prompts.js';
const POLL_INTERVAL_MS = 1000; // 1 second
const MODEL = 'claude-sonnet-4-5';
const DISALLOWED_TOOLS = ['Glob', 'Grep', 'ListMcpResourcesTool', 'WebSearch'];
interface ObservationMessage {
type: 'observation';
tool_name: string;
tool_input: string;
tool_output: string;
}
interface FinalizeMessage {
type: 'finalize';
}
type WorkerMessage = ObservationMessage | FinalizeMessage;
/**
* Main worker process entry point
*/
async function main() {
export async function main() {
const sessionDbId = parseInt(process.argv[2], 10);
if (!sessionDbId) {
@@ -30,21 +46,28 @@ async function main() {
}
/**
* SDK Worker class - handles the full lifecycle of observation processing
* SDK Worker - Unix socket server that processes observations
*/
class SDKWorker {
private sessionDbId: number;
private db: HooksDatabase;
private socketPath: string;
private server: net.Server | null = null;
private sdkSessionId: string | null = null;
private project: string = '';
private userPrompt: string = '';
private abortController: AbortController;
private isFinalized = false;
private pendingMessages: WorkerMessage[] = [];
constructor(sessionDbId: number) {
this.sessionDbId = sessionDbId;
this.db = new HooksDatabase();
this.abortController = new AbortController();
// Socket path: ~/.claude-mem/worker-{sessionId}.sock
const dataDir = PathDiscovery.getInstance().getDataDirectory();
this.socketPath = join(dataDir, `worker-${sessionDbId}.sock`);
}
/**
@@ -62,26 +85,92 @@ class SDKWorker {
this.project = session.project;
this.userPrompt = session.user_prompt;
// Start Unix socket server
await this.startSocketServer();
console.error(`[SDK Worker] Socket server listening: ${this.socketPath}`);
// Run SDK agent with streaming input
await this.runSDKAgent();
// Mark session as completed
this.db.markSessionCompleted(this.sessionDbId);
this.db.close();
this.cleanup();
} catch (error: any) {
console.error('[SDK Worker] Error:', error.message);
this.db.markSessionFailed(this.sessionDbId);
this.db.close();
this.cleanup();
process.exit(1);
}
}
/**
* Start Unix socket server to receive messages from hooks
*/
private async startSocketServer(): Promise<void> {
// Clean up old socket if it exists
if (existsSync(this.socketPath)) {
unlinkSync(this.socketPath);
}
return new Promise((resolve, reject) => {
this.server = net.createServer((socket) => {
let buffer = '';
socket.on('data', (chunk) => {
buffer += chunk.toString();
// Try to parse complete JSON messages (separated by newlines)
const lines = buffer.split('\n');
buffer = lines.pop() || ''; // Keep incomplete line in buffer
for (const line of lines) {
if (line.trim()) {
try {
const message: WorkerMessage = JSON.parse(line);
this.handleMessage(message);
} catch (err) {
console.error('[SDK Worker] Invalid message:', line);
}
}
}
});
socket.on('error', (err) => {
console.error('[SDK Worker] Socket connection error:', err.message);
});
});
this.server.on('error', (err: any) => {
if (err.code === 'EADDRINUSE') {
console.error(`[SDK Worker] Socket already in use: ${this.socketPath}`);
}
reject(err);
});
this.server.listen(this.socketPath, () => {
resolve();
});
});
}
/**
* Handle incoming message from hook
*/
private handleMessage(message: WorkerMessage): void {
this.pendingMessages.push(message);
if (message.type === 'finalize') {
this.isFinalized = true;
}
}
/**
* Load session from database
*/
private async loadSession(): Promise<SDKSession | null> {
// Query session by ID
const db = this.db as any;
const query = db.db.query(`
SELECT id, sdk_session_id, project, user_prompt
@@ -98,11 +187,9 @@ class SDKWorker {
* Run SDK agent with streaming input mode
*/
private async runSDKAgent(): Promise<void> {
const messageGenerator = this.createMessageGenerator();
await query({
model: MODEL,
messages: messageGenerator,
messages: () => this.createMessageGenerator(),
disallowedTools: DISALLOWED_TOOLS,
signal: this.abortController.signal,
onSystemInitMessage: (msg) => {
@@ -121,6 +208,7 @@ class SDKWorker {
/**
* Create async message generator for SDK streaming input
* Now pulls from socket messages instead of polling database
*/
private async* createMessageGenerator(): AsyncIterable<{ role: 'user'; content: string }> {
// Yield initial prompt
@@ -128,36 +216,37 @@ class SDKWorker {
const initPrompt = buildInitPrompt(this.project, claudeSessionId, this.userPrompt);
yield { role: 'user', content: initPrompt };
// Poll observation queue
// Process messages as they arrive via socket
while (!this.isFinalized) {
await this.sleep(POLL_INTERVAL_MS);
if (!this.sdkSessionId) {
continue; // Wait for SDK session ID to be captured
// Wait for messages to arrive
if (this.pendingMessages.length === 0) {
await this.sleep(100); // Short sleep, just to yield control
continue;
}
// Get pending observations
const observations = this.db.getPendingObservations(this.sdkSessionId, 10);
// Process all pending messages
while (this.pendingMessages.length > 0) {
const message = this.pendingMessages.shift()!;
for (const obs of observations) {
// Check for FINALIZE message
if (this.isFinalizationMessage(obs)) {
if (message.type === 'finalize') {
this.isFinalized = true;
const session = await this.loadSession();
if (session) {
const finalizePrompt = buildFinalizePrompt(session);
yield { role: 'user', content: finalizePrompt };
}
this.db.markObservationProcessed(obs.id);
break;
}
// Send observation to SDK
const observationPrompt = buildObservationPrompt(obs);
yield { role: 'user', content: observationPrompt };
// Mark as processed
this.db.markObservationProcessed(obs.id);
if (message.type === 'observation') {
// Build observation prompt
const observationPrompt = buildObservationPrompt({
tool_name: message.tool_name,
tool_input: message.tool_input,
tool_output: message.tool_output
});
yield { role: 'user', content: observationPrompt };
}
}
}
}
@@ -194,10 +283,15 @@ class SDKWorker {
}
/**
* Check if observation is a FINALIZE message
* Cleanup socket server and socket file
*/
private isFinalizationMessage(obs: Observation): boolean {
return obs.tool_name === 'FINALIZE';
private cleanup(): void {
if (this.server) {
this.server.close();
}
if (existsSync(this.socketPath)) {
unlinkSync(this.socketPath);
}
}
/**
src/services/sqlite/HooksDatabase.ts
-56
View File
@@ -103,62 +103,6 @@ export class HooksDatabase {
query.run(sdkSessionId, id);
}
/**
* Queue an observation for SDK processing
*/
queueObservation(
sdkSessionId: string,
toolName: string,
toolInput: string,
toolOutput: string
): void {
const nowEpoch = Date.now();
const query = this.db.query(`
INSERT INTO observation_queue
(sdk_session_id, tool_name, tool_input, tool_output, created_at_epoch)
VALUES (?, ?, ?, ?, ?)
`);
query.run(sdkSessionId, toolName, toolInput, toolOutput, nowEpoch);
}
/**
* Get pending observations for SDK processing
*/
getPendingObservations(sdkSessionId: string, limit: number = 10): Array<{
id: number;
tool_name: string;
tool_input: string;
tool_output: string;
created_at_epoch: number;
}> {
const query = this.db.query(`
SELECT id, tool_name, tool_input, tool_output, created_at_epoch
FROM observation_queue
WHERE sdk_session_id = ? AND processed_at_epoch IS NULL
ORDER BY created_at_epoch ASC
LIMIT ?
`);
return query.all(sdkSessionId, limit) as any[];
}
/**
* Mark observation as processed
*/
markObservationProcessed(id: number): void {
const nowEpoch = Date.now();
const query = this.db.query(`
UPDATE observation_queue
SET processed_at_epoch = ?
WHERE id = ?
`);
query.run(nowEpoch, id);
}
/**
* Store an observation (from SDK parsing)
*/
src/services/sqlite/migrations.ts
+59 -1
View File
@@ -305,6 +305,63 @@ export const migration004: Migration = {
}
};
/**
* Migration 005 - Remove orphaned tables
* Drops streaming_sessions (superseded by sdk_sessions)
* Drops observation_queue (superseded by Unix socket communication)
*/
export const migration005: Migration = {
version: 5,
up: (db: Database) => {
// Drop streaming_sessions - superseded by sdk_sessions in migration004
// This table was from v2 architecture and is no longer used
db.run(`DROP TABLE IF EXISTS streaming_sessions`);
// Drop observation_queue - superseded by Unix socket communication
// Worker now uses sockets instead of database polling for observations
db.run(`DROP TABLE IF EXISTS observation_queue`);
console.log('✅ Dropped orphaned tables: streaming_sessions, observation_queue');
},
down: (db: Database) => {
// Recreate tables if needed (though they should never be used)
db.run(`
CREATE TABLE IF NOT EXISTS streaming_sessions (
id INTEGER PRIMARY KEY AUTOINCREMENT,
claude_session_id TEXT UNIQUE NOT NULL,
sdk_session_id TEXT,
project TEXT NOT NULL,
title TEXT,
subtitle TEXT,
user_prompt TEXT,
started_at TEXT NOT NULL,
started_at_epoch INTEGER NOT NULL,
updated_at TEXT,
updated_at_epoch INTEGER,
completed_at TEXT,
completed_at_epoch INTEGER,
status TEXT NOT NULL DEFAULT 'active'
)
`);
db.run(`
CREATE TABLE IF NOT EXISTS observation_queue (
id INTEGER PRIMARY KEY AUTOINCREMENT,
sdk_session_id TEXT NOT NULL,
tool_name TEXT NOT NULL,
tool_input TEXT NOT NULL,
tool_output TEXT NOT NULL,
created_at_epoch INTEGER NOT NULL,
processed_at_epoch INTEGER,
FOREIGN KEY(sdk_session_id) REFERENCES sdk_sessions(sdk_session_id) ON DELETE CASCADE
)
`);
console.log('⚠️ Recreated streaming_sessions and observation_queue (for rollback only)');
}
};
/**
* All migrations in order
*/
@@ -312,5 +369,6 @@ export const migrations: Migration[] = [
migration001,
migration002,
migration003,
migration004
migration004,
migration005
];