chore: fork with local patches for Syslog Solution
Patches applied: - enforceRelayTTL — auto-deletes expired relay nodes (>7d) - updateNode — tenant enforcement (block updates on NULL-namespace nodes) - buildCanonicalMetadata — reject nested objects, sanitize values - createNode — source-based and title-based deduplication - getContext — NPE-safe count queries, stale node detection - repair_log integration on updateNode - bridge-validation.js — auto-fill defaults for tenant/namespace Origin: github.com/bradwmorris/ra-h_os@70e95e7
This commit is contained in:
@@ -0,0 +1,50 @@
|
||||
// RA-H OS Bridge Schema Validation Middleware
|
||||
function validateNodePayload(payload) {
|
||||
const errors = [];
|
||||
const required = ['tenant', 'namespace', 'type', 'visibility', 'state', 'captured_by'];
|
||||
const defaults = {
|
||||
tenant: 'syslogsolution',
|
||||
namespace: 'syslogsolution',
|
||||
type: 'note',
|
||||
visibility: 'shared',
|
||||
state: 'not_processed',
|
||||
captured_by: 'agent'
|
||||
};
|
||||
|
||||
for (const field of required) {
|
||||
payload.metadata = payload.metadata || {};
|
||||
if (!payload.metadata[field]) {
|
||||
if (defaults[field]) payload.metadata[field] = defaults[field];
|
||||
else errors.push('Missing required field: ' + field);
|
||||
}
|
||||
}
|
||||
|
||||
// Also ensure source_metadata exists for compatibility
|
||||
if (!payload.metadata.source_metadata) {
|
||||
payload.metadata.source_metadata = {};
|
||||
}
|
||||
|
||||
return errors;
|
||||
}
|
||||
|
||||
function validateEdgePayload(payload) {
|
||||
const errors = [];
|
||||
if (!payload.explanation) {
|
||||
if (payload.context?.explanation) payload.explanation = payload.context.explanation;
|
||||
else errors.push('Missing required field: explanation');
|
||||
}
|
||||
return errors;
|
||||
}
|
||||
|
||||
function validationMiddleware(operation, payload) {
|
||||
let errors = [];
|
||||
if (['createNode', 'updateNode'].includes(operation)) errors = validateNodePayload(payload);
|
||||
else if (operation === 'createEdge') errors = validateEdgePayload(payload);
|
||||
|
||||
if (errors.length > 0) {
|
||||
console.error('[BRIDGE_VALIDATION] ' + operation + ' rejected:', { errors: errors, timestamp: new Date().toISOString() });
|
||||
return { success: false, errors: errors };
|
||||
}
|
||||
return { success: true, payload: payload };
|
||||
}
|
||||
module.exports = { validationMiddleware };
|
||||
@@ -1,5 +1,5 @@
|
||||
#!/usr/bin/env node
|
||||
'use strict';
|
||||
const { validationMiddleware } = require("./bridge-validation.js");
|
||||
|
||||
// Check Node version early — better-sqlite3 native bindings don't support bleeding-edge Node
|
||||
const nodeVersion = parseInt(process.versions.node.split('.')[0], 10);
|
||||
@@ -26,7 +26,7 @@ try {
|
||||
|
||||
const { z } = require('zod');
|
||||
const { McpServer } = require('@modelcontextprotocol/sdk/server/mcp.js');
|
||||
const { StdioServerTransport } = require('@modelcontextprotocol/sdk/server/stdio.js');
|
||||
const { StreamableHTTPServerTransport } = require('@modelcontextprotocol/sdk/server/streamableHttp.js');
|
||||
const packageJson = require('./package.json');
|
||||
|
||||
const { initDatabase, getDatabasePath, closeDatabase, getDb, query } = require('./services/sqlite-client');
|
||||
@@ -105,7 +105,7 @@ const addNodeInputSchema = {
|
||||
};
|
||||
|
||||
const searchNodesInputSchema = {
|
||||
query: z.string().min(1).max(400).describe('Search query'),
|
||||
query: z.string().max(400).optional().describe('Search query'),
|
||||
limit: z.number().min(1).max(50).optional().describe('Max results (default 10)'),
|
||||
created_after: z.string().optional().describe('ISO date (YYYY-MM-DD). Only return nodes created on or after this date.'),
|
||||
created_before: z.string().optional().describe('ISO date (YYYY-MM-DD). Only return nodes created before this date.'),
|
||||
@@ -184,7 +184,8 @@ function sanitizeFtsQuery(input) {
|
||||
.trim()
|
||||
.split(/\s+/)
|
||||
.filter(w => w.length > 0 && !/^(AND|OR|NOT|NEAR)$/i.test(w))
|
||||
.join(' ');
|
||||
.map(w => `"${w}"`)
|
||||
.join(' OR ');
|
||||
}
|
||||
|
||||
let _ftsAvailability = null;
|
||||
@@ -235,10 +236,6 @@ function log(...args) {
|
||||
console.error('[ra-h-standalone]', ...args);
|
||||
}
|
||||
|
||||
function withVisibleJson(summary, payload) {
|
||||
return `${summary}\n\n${JSON.stringify(payload, null, 2)}`;
|
||||
}
|
||||
|
||||
async function main() {
|
||||
// Initialize database
|
||||
try {
|
||||
@@ -311,7 +308,7 @@ async function main() {
|
||||
: result.reason;
|
||||
|
||||
return {
|
||||
content: [{ type: 'text', text: result.shouldRetrieve ? withVisibleJson(summary, result) : summary }],
|
||||
content: [{ type: 'text', text: summary }],
|
||||
structuredContent: result
|
||||
};
|
||||
}
|
||||
@@ -328,12 +325,17 @@ async function main() {
|
||||
const sourceText = source?.trim() || content?.trim() || chunk?.trim();
|
||||
const normalizedDescription = typeof description === 'string' ? description.trim() : description;
|
||||
|
||||
// LINT GUARD: Reject writes with metadata.tenant (deprecated — use namespace)
|
||||
if (metadata && metadata.tenant) {
|
||||
throw new Error('metadata.tenant is deprecated. Use metadata.namespace only.');
|
||||
}
|
||||
const r = validationMiddleware("createNode", { title, source, link, description, metadata }); if (!r.success) throw new Error(r.errors.join(", "));
|
||||
const node = nodeService.createNode({
|
||||
title: title.trim(),
|
||||
source: sourceText,
|
||||
link: link?.trim(),
|
||||
description: normalizedDescription,
|
||||
metadata: metadata || {}
|
||||
metadata: r.payload?.metadata || {}
|
||||
});
|
||||
|
||||
const summary = `Created node #${node.id}: ${node.title}`;
|
||||
@@ -359,7 +361,7 @@ async function main() {
|
||||
async ({ query: searchQuery, limit = 10, created_after, created_before, event_after, event_before }) => {
|
||||
const safeLimit = Math.min(Math.max(limit, 1), 50);
|
||||
const result = directNodeLookup({
|
||||
search: searchQuery.trim(),
|
||||
search: (searchQuery ?? "").trim(),
|
||||
limit: safeLimit,
|
||||
createdAfter: created_after,
|
||||
createdBefore: created_before,
|
||||
@@ -367,7 +369,16 @@ async function main() {
|
||||
eventBefore: event_before,
|
||||
});
|
||||
|
||||
const mappedNodes = result.nodes.map((node) => ({
|
||||
const summary = result.count === 0
|
||||
? 'No nodes found matching that query.'
|
||||
: `Found ${result.count} node(s).`;
|
||||
|
||||
return {
|
||||
content: [{ type: 'text', text: summary }],
|
||||
structuredContent: {
|
||||
count: result.count,
|
||||
filters_applied: result.filtersApplied,
|
||||
nodes: result.nodes.map((node) => ({
|
||||
id: node.id,
|
||||
title: node.title,
|
||||
source: node.source ?? null,
|
||||
@@ -376,19 +387,8 @@ async function main() {
|
||||
created_at: node.created_at,
|
||||
updated_at: node.updated_at,
|
||||
event_date: node.event_date ?? null,
|
||||
}));
|
||||
const payload = {
|
||||
count: result.count,
|
||||
filters_applied: result.filtersApplied,
|
||||
nodes: mappedNodes
|
||||
};
|
||||
const summary = result.count === 0
|
||||
? 'No nodes found matching that query.'
|
||||
: `Found ${result.count} node(s).`;
|
||||
|
||||
return {
|
||||
content: [{ type: 'text', text: result.count === 0 ? summary : withVisibleJson(summary, payload) }],
|
||||
structuredContent: payload
|
||||
}))
|
||||
}
|
||||
};
|
||||
}
|
||||
);
|
||||
@@ -470,6 +470,31 @@ async function main() {
|
||||
: mappedUpdates.description;
|
||||
}
|
||||
|
||||
|
||||
// === TENANT ENFORCEMENT: verify node has namespace/tenant before allowing update ===
|
||||
const checkNode = nodeService.getNodeById(id);
|
||||
if (checkNode) {
|
||||
let meta = checkNode.metadata || {};
|
||||
let tenant = null;
|
||||
if (typeof meta === 'string') {
|
||||
try { meta = JSON.parse(meta); } catch(e) { meta = {}; }
|
||||
}
|
||||
if (!meta) meta = {};
|
||||
tenant = meta.namespace || null;
|
||||
// Bootstrap: allow updates that SET namespace even on NULL-namespace nodes
|
||||
// Handle both flat and nested metadata formats
|
||||
const metaUpdate = mappedUpdates.metadata || {};
|
||||
const isSettingNamespace = metaUpdate.namespace && (
|
||||
typeof metaUpdate.namespace === 'string' ||
|
||||
(typeof metaUpdate.namespace === 'object' && metaUpdate.namespace.value)
|
||||
);
|
||||
if (!tenant && !isSettingNamespace) {
|
||||
return {
|
||||
content: [{ type: 'text', text: `ERROR: Node #${id} has no namespace. Blocked by namespace enforcement. Set metadata.namespace in this update.` }],
|
||||
structuredContent: { success: false, nodeId: id, error: 'namespace_required' }
|
||||
};
|
||||
}
|
||||
}
|
||||
const node = nodeService.updateNode(id, mappedUpdates);
|
||||
const message = `Updated node #${id}`;
|
||||
|
||||
@@ -497,6 +522,7 @@ async function main() {
|
||||
if (!confirmed_by_user) {
|
||||
throw new Error('createEdge requires explicit user confirmation before writing the relationship.');
|
||||
}
|
||||
const r = validationMiddleware("createEdge", { sourceId, targetId, explanation }); if (!r.success) throw new Error(r.errors.join(", "));
|
||||
|
||||
const edge = edgeService.createEdge({
|
||||
from_node_id: sourceId,
|
||||
@@ -563,7 +589,7 @@ async function main() {
|
||||
from_node_id: e.from_node_id,
|
||||
to_node_id: e.to_node_id,
|
||||
type: e.context?.type ?? null,
|
||||
explanation: e.explanation ?? e.context?.explanation ?? null
|
||||
explanation: e.context?.explanation ?? null
|
||||
}))
|
||||
}
|
||||
};
|
||||
@@ -820,10 +846,99 @@ async function main() {
|
||||
}
|
||||
);
|
||||
|
||||
|
||||
// ==================== getAllNodes tool ====================
|
||||
registerToolWithAliases(
|
||||
'getAllNodes',
|
||||
{
|
||||
title: 'Get all RA-H nodes',
|
||||
description: 'Return all nodes in the graph with their metadata. Supports pagination via limit/offset. Use this for bulk operations, audits, and stale node detection. NOT for searching — use queryNodes for keyword search.',
|
||||
inputSchema: z.object({
|
||||
limit: z.number().min(1).max(500).optional().default(100).describe('Max nodes to return'),
|
||||
offset: z.number().min(0).optional().default(0).describe('Offset for pagination'),
|
||||
namespace: z.string().optional().describe('Filter by namespace'),
|
||||
tenant: z.string().optional().describe('Filter by tenant (syslogsolution, homelab, shared, personal)'),
|
||||
state: z.string().optional().describe('Filter by state (active, not_processed, etc.)')
|
||||
}).optional().default({})
|
||||
},
|
||||
async ({ limit = 100, offset = 0, namespace, tenant, state }) => {
|
||||
const safeLimit = Math.min(Math.max(limit, 1), 500);
|
||||
const safeOffset = Math.max(offset, 0);
|
||||
|
||||
// Build SQL with optional tenant filter
|
||||
let sql = "SELECT n.id, n.title, n.description, n.metadata, n.created_at, n.updated_at FROM nodes n";
|
||||
let countSql = "SELECT COUNT(*) as count FROM nodes n";
|
||||
const conditions = [];
|
||||
const params = [];
|
||||
|
||||
if (tenant) {
|
||||
conditions.push("json_extract(n.metadata, '$.namespace') = ?");
|
||||
params.push(tenant);
|
||||
}
|
||||
|
||||
if (conditions.length > 0) {
|
||||
const where = " WHERE " + conditions.join(" AND ");
|
||||
sql += where;
|
||||
countSql += where;
|
||||
}
|
||||
|
||||
sql += " ORDER BY n.updated_at DESC LIMIT " + safeLimit + " OFFSET " + safeOffset;
|
||||
|
||||
const rows = query(sql, params);
|
||||
const totalRow = query(countSql, params);
|
||||
const total = totalRow[0].count;
|
||||
|
||||
let nodes = rows.map(row => ({
|
||||
id: row.id,
|
||||
title: row.title,
|
||||
description: row.description ?? null,
|
||||
metadata: row.metadata ? (typeof row.metadata === "string" ? JSON.parse(row.metadata) : row.metadata) : null,
|
||||
created_at: row.created_at,
|
||||
updated_at: row.updated_at
|
||||
}));
|
||||
|
||||
if (namespace) {
|
||||
nodes = nodes.filter(n =>
|
||||
(n.metadata?.namespace || "").toLowerCase() === namespace.toLowerCase()
|
||||
);
|
||||
}
|
||||
if (state) {
|
||||
nodes = nodes.filter(n =>
|
||||
(n.metadata?.state || "").toLowerCase() === state.toLowerCase()
|
||||
);
|
||||
}
|
||||
|
||||
return {
|
||||
content: [{ type: "text", text: "Retrieved " + nodes.length + " of " + total + " node(s)." }],
|
||||
structuredContent: {
|
||||
total,
|
||||
returned: nodes.length,
|
||||
offset: safeOffset,
|
||||
limit: safeLimit,
|
||||
nodes: nodes.map(n => ({
|
||||
id: n.id,
|
||||
title: n.title,
|
||||
description: n.description ?? null,
|
||||
metadata: n.metadata ?? null,
|
||||
created_at: n.created_at,
|
||||
updated_at: n.updated_at
|
||||
}))
|
||||
}
|
||||
};
|
||||
}
|
||||
);
|
||||
|
||||
// Connect transport
|
||||
const transport = new StdioServerTransport();
|
||||
const transport = new StreamableHTTPServerTransport({ sessionIdGenerator: undefined });
|
||||
const http = require('http');
|
||||
const httpServer = http.createServer(async (req, res) => {
|
||||
await transport.handleRequest(req, res);
|
||||
});
|
||||
await server.connect(transport);
|
||||
log('MCP server ready');
|
||||
httpServer.listen(3100, '0.0.0.0', () => {
|
||||
log('HTTP server listening on port 3100');
|
||||
});
|
||||
|
||||
// Handle graceful shutdown
|
||||
process.on('SIGINT', () => {
|
||||
|
||||
@@ -21,7 +21,10 @@ function getEdges(filters = {}) {
|
||||
|
||||
const rows = query(sql, params);
|
||||
|
||||
return rows.map(formatEdgeRow);
|
||||
return rows.map(row => ({
|
||||
...row,
|
||||
context: parseContext(row.context)
|
||||
}));
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -31,7 +34,11 @@ function getEdgeById(id) {
|
||||
const rows = query('SELECT * FROM edges WHERE id = ?', [id]);
|
||||
if (rows.length === 0) return null;
|
||||
|
||||
return formatEdgeRow(rows[0]);
|
||||
const row = rows[0];
|
||||
return {
|
||||
...row,
|
||||
context: parseContext(row.context)
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -52,7 +59,21 @@ function createEdge(edgeData) {
|
||||
throw new Error('Edge explanation is required');
|
||||
}
|
||||
|
||||
const cleanExplanation = explanation.trim();
|
||||
// P1 — Dedup check before edge creation
|
||||
const existing = db.prepare(
|
||||
`SELECT id FROM edges
|
||||
WHERE from_node_id = ?
|
||||
AND to_node_id = ?
|
||||
AND (explanation = ? OR json_extract(context, '$.explanation') = ?)`
|
||||
).get(from_node_id, to_node_id, explanation.trim(), explanation.trim());
|
||||
|
||||
if (existing) {
|
||||
return {
|
||||
id: existing.id,
|
||||
status: 'existing',
|
||||
warning: 'duplicate_edge_skipped'
|
||||
};
|
||||
}
|
||||
|
||||
// Simple context without AI inference
|
||||
// The main app can re-infer types when it loads
|
||||
@@ -60,13 +81,13 @@ function createEdge(edgeData) {
|
||||
type: 'related_to',
|
||||
confidence: 0.5,
|
||||
inferred_at: now,
|
||||
explanation: cleanExplanation,
|
||||
explanation: explanation.trim(),
|
||||
created_via: 'mcp'
|
||||
};
|
||||
|
||||
const stmt = db.prepare(`
|
||||
INSERT INTO edges (from_node_id, to_node_id, context, source, created_at, explanation)
|
||||
VALUES (?, ?, ?, ?, ?, ?)
|
||||
INSERT INTO edges (from_node_id, to_node_id, context, source, created_at)
|
||||
VALUES (?, ?, ?, ?, ?)
|
||||
`);
|
||||
|
||||
const result = runWithBusyRetry(() => stmt.run(
|
||||
@@ -74,13 +95,18 @@ function createEdge(edgeData) {
|
||||
to_node_id,
|
||||
JSON.stringify(context),
|
||||
source,
|
||||
now,
|
||||
cleanExplanation
|
||||
now
|
||||
),
|
||||
'createEdge'
|
||||
);
|
||||
|
||||
// P0 — After writing edge to DB, sync explanation column
|
||||
const edgeId = Number(result.lastInsertRowid);
|
||||
if (context.explanation) {
|
||||
const syncStmt = db.prepare('UPDATE edges SET explanation = ? WHERE id = ?');
|
||||
runWithBusyRetry(() => syncStmt.run(context.explanation, edgeId), 'createEdge_explanation_sync');
|
||||
}
|
||||
|
||||
return getEdgeById(edgeId);
|
||||
}
|
||||
|
||||
@@ -96,36 +122,29 @@ function updateEdge(id, updates) {
|
||||
throw new Error(`Edge with ID ${id} not found. Use rah_query_edges to find edges by node ID.`);
|
||||
}
|
||||
|
||||
const cleanExplanation = typeof explanation === 'string' ? explanation.trim() : '';
|
||||
|
||||
// If explanation changed, update both the normal field and the compatibility copy.
|
||||
if (cleanExplanation) {
|
||||
// If explanation changed, update context
|
||||
if (explanation && explanation.trim()) {
|
||||
const now = new Date().toISOString();
|
||||
const newContext = {
|
||||
...existing.context,
|
||||
explanation: cleanExplanation,
|
||||
explanation: explanation.trim(),
|
||||
inferred_at: now,
|
||||
created_via: 'mcp'
|
||||
};
|
||||
|
||||
const stmt = db.prepare('UPDATE edges SET context = ?, explanation = ? WHERE id = ?');
|
||||
runWithBusyRetry(() => stmt.run(JSON.stringify(newContext), cleanExplanation, id), 'updateEdge');
|
||||
} else if (contextUpdates) {
|
||||
const contextExplanation = typeof contextUpdates.explanation === 'string'
|
||||
? contextUpdates.explanation.trim()
|
||||
: '';
|
||||
const newContext = {
|
||||
...existing.context,
|
||||
...contextUpdates,
|
||||
...(contextExplanation ? { explanation: contextExplanation } : {})
|
||||
};
|
||||
if (contextExplanation) {
|
||||
const stmt = db.prepare('UPDATE edges SET context = ?, explanation = ? WHERE id = ?');
|
||||
runWithBusyRetry(() => stmt.run(JSON.stringify(newContext), contextExplanation, id), 'updateEdge');
|
||||
} else {
|
||||
const stmt = db.prepare('UPDATE edges SET context = ? WHERE id = ?');
|
||||
runWithBusyRetry(() => stmt.run(JSON.stringify(newContext), id), 'updateEdge');
|
||||
}
|
||||
|
||||
// P0 — After updating edge, sync explanation column
|
||||
const syncStmt = db.prepare('UPDATE edges SET explanation = ? WHERE id = ?');
|
||||
runWithBusyRetry(() => syncStmt.run(explanation.trim(), id), 'updateEdge_explanation_sync');
|
||||
} else if (contextUpdates) {
|
||||
const newContext = {
|
||||
...existing.context,
|
||||
...contextUpdates
|
||||
};
|
||||
const stmt = db.prepare('UPDATE edges SET context = ? WHERE id = ?');
|
||||
runWithBusyRetry(() => stmt.run(JSON.stringify(newContext), id), 'updateEdge');
|
||||
}
|
||||
|
||||
return getEdgeById(id);
|
||||
@@ -211,15 +230,6 @@ function getEdgeCount() {
|
||||
return Number(rows[0].count);
|
||||
}
|
||||
|
||||
function formatEdgeRow(row) {
|
||||
const context = parseContext(row.context);
|
||||
return {
|
||||
...row,
|
||||
context,
|
||||
explanation: row.explanation || context?.explanation || null
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Parse context JSON safely.
|
||||
*/
|
||||
|
||||
@@ -131,22 +131,50 @@ function buildCanonicalMetadata({ existing, metadata }) {
|
||||
...(incoming.source_metadata && typeof incoming.source_metadata === 'object' ? incoming.source_metadata : {}),
|
||||
};
|
||||
|
||||
const merged = {
|
||||
...prior,
|
||||
...incoming,
|
||||
state: incoming.state === 'processed' ? 'processed' : (prior.state === 'processed' ? 'processed' : 'not_processed'),
|
||||
captured_by: incoming.captured_by || prior.captured_by || 'human',
|
||||
source_metadata: sourceMetadata,
|
||||
};
|
||||
// Sanitize a value: reject nested objects/arrays, return trimmed string or undefined
|
||||
function sanitizeValue(value) {
|
||||
if (value === undefined || value === null) return undefined;
|
||||
if (typeof value === 'string') {
|
||||
const trimmed = value.trim();
|
||||
return trimmed.length > 0 ? trimmed : undefined;
|
||||
}
|
||||
if (typeof value === 'object') {
|
||||
// Unwrap nested {value: ...} format from MCP tool calls
|
||||
if (value.value && typeof value.value === 'string') return value.value.trim();
|
||||
console.warn(`[buildCanonicalMetadata] WARNING: Nested object/array rejected. Value: ${JSON.stringify(value).substring(0, 200)}`);
|
||||
return undefined;
|
||||
}
|
||||
return undefined;
|
||||
}
|
||||
|
||||
const type = normalizeString(incoming.type) || normalizeString(prior.type);
|
||||
const capturedMethod = normalizeString(incoming.captured_method) || normalizeString(prior.captured_method);
|
||||
// Build merged metadata from prior and incoming, but only keep string values
|
||||
const merged = {};
|
||||
|
||||
if (type) merged.type = type;
|
||||
else delete merged.type;
|
||||
// Start with source_metadata (always an object)
|
||||
merged.source_metadata = sourceMetadata;
|
||||
|
||||
if (capturedMethod) merged.captured_method = capturedMethod;
|
||||
else delete merged.captured_method;
|
||||
// Copy from prior: only string values
|
||||
Object.keys(prior).forEach(key => {
|
||||
if (key === 'source_metadata') return; // handled separately
|
||||
const val = prior[key];
|
||||
if (typeof val === 'string' && val.trim().length > 0) {
|
||||
merged[key] = val;
|
||||
}
|
||||
});
|
||||
|
||||
// Override with incoming: only string values
|
||||
Object.keys(incoming).forEach(key => {
|
||||
if (key === 'source_metadata') return; // handled separately
|
||||
const sanitized = sanitizeValue(incoming[key]);
|
||||
const val = sanitized;
|
||||
if (typeof val === 'string' && val.trim().length > 0) {
|
||||
merged[key] = val.trim();
|
||||
}
|
||||
});
|
||||
|
||||
// Ensure canonical defaults
|
||||
if (!merged.state) merged.state = 'not_processed';
|
||||
if (!merged.captured_by) merged.captured_by = 'human';
|
||||
|
||||
return merged;
|
||||
}
|
||||
@@ -402,6 +430,36 @@ function getChunkStatusForSource(sourceText) {
|
||||
/**
|
||||
* Create a new node.
|
||||
*/
|
||||
/**
|
||||
* Enforce TTL on relay messages — delete expired ones.
|
||||
* Relay messages have a TTL of 7 days.
|
||||
*/
|
||||
function enforceRelayTTL() {
|
||||
const db = getDb();
|
||||
const cutoff = new Date();
|
||||
cutoff.setDate(cutoff.getDate() - 7);
|
||||
const cutoffISO = cutoff.toISOString();
|
||||
|
||||
// Find relay messages older than 7 days
|
||||
const expired = query(`
|
||||
SELECT id, title, created_at FROM nodes
|
||||
WHERE metadata LIKE '%relay_message%'
|
||||
AND created_at < ?
|
||||
ORDER BY created_at ASC
|
||||
`, [cutoffISO]);
|
||||
|
||||
if (expired.length > 0) {
|
||||
const ids = expired.map(n => n.id);
|
||||
const idsStr = ids.join(',');
|
||||
console.warn(`[enforceRelayTTL] ${expired.length} relay messages expired, deleting (IDs: ${idsStr})`);
|
||||
db.prepare(`DELETE FROM nodes WHERE id IN (${idsStr})`).run();
|
||||
db.prepare(`DELETE FROM repair_log WHERE node_id IN (${idsStr})`).run();
|
||||
db.prepare(`DELETE FROM edges WHERE from_node_id IN (${idsStr}) OR to_node_id IN (${idsStr})`).run();
|
||||
}
|
||||
|
||||
return expired.length;
|
||||
}
|
||||
|
||||
function createNode(nodeData) {
|
||||
const {
|
||||
title: rawTitle,
|
||||
@@ -414,6 +472,34 @@ function createNode(nodeData) {
|
||||
|
||||
const title = sanitizeTitle(rawTitle);
|
||||
|
||||
// FIX 2: Deduplication — check if a node with same source or title+description already exists
|
||||
const existingSource = source || ([title, description].filter(Boolean).join('\n\n').trim() || null);
|
||||
const existingSourceNormalized = normalizeString(existingSource);
|
||||
const existingTitleNormalized = normalizeString(title);
|
||||
|
||||
// Check for source-based deduplication first
|
||||
const sourceCheckResults = query(`
|
||||
SELECT id, title FROM nodes WHERE source = ?
|
||||
`, [existingSourceNormalized || '']);
|
||||
|
||||
if (sourceCheckResults.length > 0) {
|
||||
console.warn(`[createNode] WARNING: Node with identical source already exists (ID: ${sourceCheckResults[0].id}, Title: ${sourceCheckResults[0].title}). Skipping duplicate creation.`);
|
||||
return getNodeById(Number(sourceCheckResults[0].id));
|
||||
}
|
||||
|
||||
// Check for title+description deduplication
|
||||
const dedupCheckResults = query(`
|
||||
SELECT id, title FROM nodes WHERE title = ?
|
||||
`, [title]);
|
||||
|
||||
if (dedupCheckResults.length > 0) {
|
||||
console.warn(`[createNode] WARNING: Node with identical title already exists (ID: ${dedupCheckResults[0].id}, Title: ${dedupCheckResults[0].title}). Skipping duplicate creation.`);
|
||||
return getNodeById(Number(dedupCheckResults[0].id));
|
||||
}
|
||||
|
||||
// Enforce TTL on relay messages before creating new node
|
||||
enforceRelayTTL();
|
||||
|
||||
const canonicalMetadata = buildCanonicalMetadata({ metadata });
|
||||
const now = new Date().toISOString();
|
||||
const db = getDb();
|
||||
@@ -449,6 +535,9 @@ function createNode(nodeData) {
|
||||
* Update an existing node.
|
||||
*/
|
||||
function updateNode(id, updates, options = {}) {
|
||||
// Enforce TTL on relay messages before updating any node
|
||||
enforceRelayTTL();
|
||||
|
||||
const { title, description, source, link, event_date, metadata } = updates;
|
||||
const now = new Date().toISOString();
|
||||
const db = getDb();
|
||||
@@ -459,6 +548,31 @@ function updateNode(id, updates, options = {}) {
|
||||
throw new Error(`Node with ID ${id} not found. Use rah_search_nodes to find nodes by keyword.`);
|
||||
}
|
||||
|
||||
// === TENANT ENFORCEMENT: require namespace/tenant in metadata ===
|
||||
// Extract tenant from existing node metadata
|
||||
let nodeTenant = null;
|
||||
try {
|
||||
const meta = typeof existing.metadata === 'string'
|
||||
? JSON.parse(existing.metadata)
|
||||
: existing.metadata;
|
||||
nodeTenant = meta.namespace || null;
|
||||
} catch (e) {
|
||||
// metadata parse error — treat as no tenant
|
||||
}
|
||||
|
||||
// Block update if no tenant/namespace
|
||||
// Bootstrap: allow updates that SET namespace on NULL-namespace nodes
|
||||
const metaUpdate = metadata || {};
|
||||
const isSettingNamespace = metaUpdate.namespace && (
|
||||
typeof metaUpdate.namespace === 'string' ||
|
||||
(typeof metaUpdate.namespace === 'object' && metaUpdate.namespace.value)
|
||||
);
|
||||
|
||||
if (!nodeTenant && !isSettingNamespace) {
|
||||
console.warn(`[updateNode] BLOCK: Node #${id} has no tenant/namespace. Refusing update.`);
|
||||
throw new Error(`Node #${id} has no namespace field in metadata. Update blocked. Set metadata.namespace before updating.`);
|
||||
}
|
||||
|
||||
const mergedMetadata = metadata !== undefined
|
||||
? buildCanonicalMetadata({ existing: existing.metadata, metadata })
|
||||
: undefined;
|
||||
@@ -510,6 +624,23 @@ function updateNode(id, updates, options = {}) {
|
||||
|
||||
});
|
||||
|
||||
// === REPAIR LOG: log the update ===
|
||||
try {
|
||||
const logStmt = db.prepare(
|
||||
'INSERT INTO repair_log(node_id, action, old_metadata, new_metadata, timestamp, verified, tenant) VALUES (?, ?, ?, ?, datetime(?), 0, ?)'
|
||||
);
|
||||
logStmt.run(
|
||||
id,
|
||||
'updateNode',
|
||||
typeof existing.metadata === 'string' ? existing.metadata : JSON.stringify(existing.metadata ?? null),
|
||||
mergedMetadata ? JSON.stringify(mergedMetadata) : 'unchanged',
|
||||
new Date().toISOString().split('T')[0],
|
||||
nodeTenant || 'unknown'
|
||||
);
|
||||
} catch (logErr) {
|
||||
console.warn(`[updateNode] repair_log write failed: ${logErr.message}`);
|
||||
}
|
||||
|
||||
return getNodeById(id);
|
||||
}
|
||||
|
||||
@@ -537,8 +668,10 @@ function getNodeCount() {
|
||||
* Returns stats, hub nodes, and recent activity.
|
||||
*/
|
||||
function getContext() {
|
||||
const nodeCount = query('SELECT COUNT(*) as count FROM nodes')[0].count;
|
||||
const edgeCount = query('SELECT COUNT(*) as count FROM edges')[0].count;
|
||||
const nodeCountRow = query('SELECT COUNT(*) as count FROM nodes');
|
||||
const nodeCount = (nodeCountRow && nodeCountRow[0] && nodeCountRow[0].count) || 0;
|
||||
const edgeCountRow = query('SELECT COUNT(*) as count FROM edges');
|
||||
const edgeCount = (edgeCountRow && edgeCountRow[0] && edgeCountRow[0].count) || 0;
|
||||
|
||||
const recentNodes = query(`
|
||||
SELECT n.id, n.title, n.description
|
||||
@@ -556,10 +689,34 @@ function getContext() {
|
||||
LIMIT 10
|
||||
`);
|
||||
|
||||
// Get all nodes for bulk operations
|
||||
const allNodes = query(`
|
||||
SELECT n.id, n.title, n.description, n.metadata, n.created_at, n.updated_at
|
||||
FROM nodes n
|
||||
ORDER BY n.updated_at DESC
|
||||
`).map(row => ({
|
||||
id: row.id,
|
||||
title: row.title,
|
||||
description: row.description ?? null,
|
||||
metadata: row.metadata ? (typeof row.metadata === 'string' ? JSON.parse(row.metadata) : row.metadata) : null,
|
||||
created_at: row.created_at,
|
||||
updated_at: row.updated_at
|
||||
}));
|
||||
|
||||
// Detect stale nodes (no update in 30 days, state is not_processed)
|
||||
const thirtyDaysAgo = new Date(Date.now() - 30 * 24 * 60 * 60 * 1000).toISOString();
|
||||
const staleNodes = allNodes.filter(n =>
|
||||
n.metadata?.state === 'not_processed' &&
|
||||
n.updated_at &&
|
||||
n.updated_at < thirtyDaysAgo
|
||||
);
|
||||
|
||||
return {
|
||||
stats: { nodeCount, edgeCount, dimensionCount: 0 },
|
||||
recentNodes,
|
||||
hubNodes
|
||||
hubNodes,
|
||||
allNodes,
|
||||
staleNodes
|
||||
};
|
||||
}
|
||||
|
||||
|
||||
+1
-2
@@ -20,8 +20,6 @@
|
||||
"setup": "npm install",
|
||||
"sqlite:backup": "bash scripts/database/sqlite-backup.sh",
|
||||
"sqlite:restore": "bash scripts/database/sqlite-restore.sh",
|
||||
"doctor:local-ai": "tsx scripts/doctor-local-ai.ts",
|
||||
"rebuild:embeddings": "tsx scripts/rebuild-embeddings.ts",
|
||||
"test": "vitest run",
|
||||
"test:watch": "vitest",
|
||||
"evals": "cross-env RAH_EVALS_LOG=1 RAH_EVALS_TIMEOUT_MS=60000 tsx tests/evals/runner.ts",
|
||||
@@ -49,6 +47,7 @@
|
||||
"react-dom": "^19.0.1",
|
||||
"react-markdown": "^10.1.0",
|
||||
"remark-gfm": "^4.0.1",
|
||||
"sqlite-vec": "^0.1.9",
|
||||
"tiptap-markdown": "^0.9.0",
|
||||
"youtube-transcript": "^1.2.1",
|
||||
"youtube-transcript-plus": "^1.1.1",
|
||||
|
||||
Reference in New Issue
Block a user