Skip to content

feat: scope analytics and config by agentId #7

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We鈥檒l occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
ttupper92618 wants to merge 1 commit into
base: feat/multi-tenant-registry
Choose a base branch
from
Open

feat: scope analytics and config by agentId #7

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
74 changes: 44 additions & 30 deletions src/analytics/db.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -87,6 +87,10 @@ export class FoxAnalyticsDB {
"ALTER TABLE write_events ADD COLUMN extracted_facts_json TEXT",
"ALTER TABLE write_events ADD COLUMN candidates_json TEXT",
"ALTER TABLE search_events ADD COLUMN graph_hit INTEGER DEFAULT 0",
"ALTER TABLE write_events ADD COLUMN agent_id TEXT",
"ALTER TABLE search_events ADD COLUMN agent_id TEXT",
"ALTER TABLE graph_events ADD COLUMN agent_id TEXT",
"ALTER TABLE memory_graph_links ADD COLUMN agent_id TEXT",
]) {
try { this.db.exec(sql); } catch { /* column already exists */ }
}
Expand Down Expand Up @@ -116,6 +120,7 @@ export class FoxAnalyticsDB {
latencyMs: number;
inferMode: boolean;
decisions?: { extractedFacts: string[]; candidates: { id: string; text: string }[]; actions: any[] };
agentId?: string;
}) {
if (!this.ready) return;
const ts = new Date().toISOString();
Expand All @@ -125,8 +130,8 @@ export class FoxAnalyticsDB {
const candidatesJson = opts.decisions?.candidates?.length
? JSON.stringify(opts.decisions.candidates) : null;
const stmt = this.db.prepare(
`INSERT INTO write_events (id, ts, event_type, memory_id, user_id, run_id, input_chars, output_text, llm_model, latency_ms, infer_mode, call_id, reason, extracted_facts_json, candidates_json)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`
`INSERT INTO write_events (id, ts, event_type, memory_id, user_id, run_id, input_chars, output_text, llm_model, latency_ms, infer_mode, call_id, reason, extracted_facts_json, candidates_json, agent_id)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`
);
const noneActions = opts.results.length === 0
? (opts.decisions?.actions || []).filter((a: any) => String(a?.event || "").toUpperCase() === "NONE")
Expand All @@ -144,7 +149,8 @@ export class FoxAnalyticsDB {
opts.user_id ?? null, opts.run_id ?? null,
opts.inputChars, memText, effectiveLlmModel,
opts.latencyMs, opts.inferMode ? 1 : 0, callId,
reason, extractedFactsJson, candidatesJson
reason, extractedFactsJson, candidatesJson,
opts.agentId ?? null
);
} catch { /* non-critical */ }
}
Expand All @@ -158,17 +164,19 @@ export class FoxAnalyticsDB {
topScore?: number;
latencyMs: number;
graphHit?: boolean;
agentId?: string;
}) {
if (!this.ready) return;
try {
this.db.prepare(
`INSERT INTO search_events (id, ts, user_id, run_id, query_chars, result_count, top_score, latency_ms, graph_hit)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`
`INSERT INTO search_events (id, ts, user_id, run_id, query_chars, result_count, top_score, latency_ms, graph_hit, agent_id)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`
).run(
crypto.randomUUID(), new Date().toISOString(),
opts.user_id ?? null, opts.run_id ?? null,
opts.queryChars, opts.resultCount, opts.topScore ?? null, opts.latencyMs,
opts.graphHit ? 1 : 0
opts.graphHit ? 1 : 0,
opts.agentId ?? null
);
} catch { /* non-critical */ }
}
Expand All @@ -179,26 +187,32 @@ export class FoxAnalyticsDB {
entitiesAdded?: number;
relationsAdded?: number;
latencyMs: number;
agentId?: string;
}) {
if (!this.ready) return;
try {
this.db.prepare(
`INSERT INTO graph_events (id, ts, user_id, run_id, entities_added, relations_added, latency_ms)
VALUES (?, ?, ?, ?, ?, ?, ?)`
`INSERT INTO graph_events (id, ts, user_id, run_id, entities_added, relations_added, latency_ms, agent_id)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)`
).run(
crypto.randomUUID(), new Date().toISOString(),
opts.user_id ?? null, opts.run_id ?? null,
opts.entitiesAdded ?? null, opts.relationsAdded ?? null, opts.latencyMs
opts.entitiesAdded ?? null, opts.relationsAdded ?? null, opts.latencyMs,
opts.agentId ?? null
);
} catch { /* non-critical */ }
}

getStats(days: number) {
getStats(days: number, agentId?: string) {
if (!this.ready) return null;
try {
const agentFilter = agentId ? " AND agent_id = ?" : "";
const agentFilterWhere = agentId ? " WHERE agent_id = ?" : "";
const agentParams = agentId ? [agentId] : [];

const eventRows = this.db.prepare(
`SELECT event_type, COUNT(*) as count FROM write_events GROUP BY event_type`
).all() as Array<{ event_type: string; count: number }>;
`SELECT event_type, COUNT(*) as count FROM write_events${agentFilterWhere} GROUP BY event_type`
).all(...agentParams) as Array<{ event_type: string; count: number }>;

const byEvent: Record<string, number> = { ADD: 0, UPDATE: 0, DELETE: 0, NONE: 0 };
for (const r of eventRows) {
Expand All @@ -209,26 +223,26 @@ export class FoxAnalyticsDB {
SELECT
COUNT(DISTINCT call_id) as totalCalls,
COUNT(DISTINCT CASE WHEN event_type = 'NONE' THEN call_id END) as noneCalls
FROM write_events WHERE call_id IS NOT NULL
`).get() as any;
FROM write_events WHERE call_id IS NOT NULL${agentFilter}
`).get(...agentParams) as any;
const noneRatePct = (callStats?.totalCalls ?? 0) > 0
? Math.round(((callStats.noneCalls ?? 0) / callStats.totalCalls) * 100)
: 0;
const totalCalls = callStats?.totalCalls ?? 0;

const latRow = this.db.prepare(
`SELECT AVG(latency_ms) as avg, MIN(latency_ms) as min, MAX(latency_ms) as max
FROM write_events WHERE latency_ms IS NOT NULL`
).get() as any;
FROM write_events WHERE latency_ms IS NOT NULL${agentFilter}`
).get(...agentParams) as any;

const byDayRaw = this.db.prepare(`
SELECT date(ts) as date, event_type, COUNT(*) as count,
CAST(AVG(latency_ms) AS INTEGER) as avg_latency_ms
FROM write_events
WHERE ts >= datetime('now', '-' || ? || ' days')
WHERE ts >= datetime('now', '-' || ? || ' days')${agentFilter}
GROUP BY date(ts), event_type
ORDER BY date(ts) ASC
`).all(days) as Array<{ date: string; event_type: string; count: number; avg_latency_ms: number | null }>;
`).all(days, ...agentParams) as Array<{ date: string; event_type: string; count: number; avg_latency_ms: number | null }>;

const byDayMap = new Map<string, { date: string; ADD: number; UPDATE: number; DELETE: number; NONE: number; avgLatencyMs: number | null }>();
for (const r of byDayRaw) {
Expand All @@ -240,8 +254,8 @@ export class FoxAnalyticsDB {

const recent = this.db.prepare(`
SELECT ts, event_type, memory_id, user_id, run_id, output_text, reason, latency_ms, infer_mode
FROM write_events ORDER BY ts DESC LIMIT 20
`).all() as any[];
FROM write_events${agentFilterWhere} ORDER BY ts DESC LIMIT 20
`).all(...agentParams) as any[];

const searchRow = this.db.prepare(`
SELECT COUNT(*) as total,
Expand All @@ -250,8 +264,8 @@ export class FoxAnalyticsDB {
CAST(AVG(result_count) * 10 AS INTEGER) / 10.0 as avgResults,
CAST(AVG(top_score) * 1000 AS INTEGER) / 1000.0 as avgTopScore,
CAST(AVG(latency_ms) AS INTEGER) as avgLatencyMs
FROM search_events
`).get() as any;
FROM search_events${agentFilterWhere}
`).get(...agentParams) as any;

const searchTotal = searchRow?.total ?? 0;
const zeroResultRatePct = searchTotal > 0
Expand All @@ -267,18 +281,18 @@ export class FoxAnalyticsDB {
SUM(CASE WHEN result_count = 0 THEN 1 ELSE 0 END) as zeroResults,
CAST(AVG(latency_ms) AS INTEGER) as avgLatencyMs
FROM search_events
WHERE ts >= datetime('now', '-' || ? || ' days')
WHERE ts >= datetime('now', '-' || ? || ' days')${agentFilter}
GROUP BY date(ts)
ORDER BY date(ts) ASC
`).all(days) as Array<{ date: string; count: number; zeroResults: number; avgLatencyMs: number | null }>;
`).all(days, ...agentParams) as Array<{ date: string; count: number; zeroResults: number; avgLatencyMs: number | null }>;

const graphRow = this.db.prepare(`
SELECT COUNT(*) as totalWrites,
SUM(relations_added) as totalRelations,
SUM(entities_added) as totalEntities,
CAST(AVG(latency_ms) AS INTEGER) as avgLatencyMs
FROM graph_events
`).get() as any;
FROM graph_events${agentFilterWhere}
`).get(...agentParams) as any;

return {
summary: {
Expand Down Expand Up @@ -370,12 +384,12 @@ export class FoxAnalyticsDB {
return (result.changes as number) > 0;
}

insertGraphLinks(vectorMemoryId: string, nodeIds: string[], edgeIds: string[], userId?: string) {
insertGraphLinks(vectorMemoryId: string, nodeIds: string[], edgeIds: string[], userId?: string, agentId?: string) {
const stmt = this.db.prepare(
"INSERT INTO memory_graph_links (vector_memory_id, graph_node_id, graph_edge_id, user_id) VALUES (?, ?, ?, ?)"
"INSERT INTO memory_graph_links (vector_memory_id, graph_node_id, graph_edge_id, user_id, agent_id) VALUES (?, ?, ?, ?, ?)"
);
for (const nodeId of nodeIds) stmt.run(vectorMemoryId, nodeId, null, userId ?? null);
for (const edgeId of edgeIds) stmt.run(vectorMemoryId, null, edgeId, userId ?? null);
for (const nodeId of nodeIds) stmt.run(vectorMemoryId, nodeId, null, userId ?? null, agentId ?? null);
for (const edgeId of edgeIds) stmt.run(vectorMemoryId, null, edgeId, userId ?? null, agentId ?? null);
}

getLinkedNodeIds(vectorMemoryId: string): string[] {
Expand Down
13 changes: 9 additions & 4 deletions src/pipeline/write.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,7 +34,7 @@ export const trackAddResult = (mode: "infer" | "raw", result: any) => {
}
};

export const captureGraphLinks = (result: any, userId?: string) => {
export const captureGraphLinks = (result: any, userId?: string, agentId?: string) => {
if (!analyticsDb?.ready || !GRAPH_ENABLED) return;
const nodeIds: string[] = result?.added_node_ids ?? [];
const edgeIds: string[] = result?.added_edge_ids ?? [];
Expand All @@ -44,7 +44,7 @@ export const captureGraphLinks = (result: any, userId?: string) => {
const ev = String(r?.metadata?.event || r?.event || "").toUpperCase();
if (ev === "ADD" || ev === "UPDATE") {
const memId = r?.id ?? r?.memory_id;
if (memId) analyticsDb.insertGraphLinks(memId, nodeIds, edgeIds, userId);
if (memId) analyticsDb.insertGraphLinks(memId, nodeIds, edgeIds, userId, agentId);

@chatgpt-codex-connector chatgpt-codex-connector Bot Mar 14, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Pass agent ID when capturing graph links

This call now writes agent_id into memory_graph_links, but v2Write still invokes captureGraphLinks with only (result, userId), so agentId is always undefined and new rows get NULL for agent_id. That means the agent-scoping column added in this commit is never populated for graph links, which breaks per-agent attribution/filtering for this dataset.

Useful? React with 馃憤聽/ 馃憥.

}
}
};
Expand Down Expand Up @@ -129,6 +129,7 @@ export const executeWriteAndRecord = async (
parsed: z.infer<typeof v2WriteSchema>,
idem: ReturnType<typeof idempotencyPrecheck>,
memoryOverride?: Memory,
agentId?: string,
): Promise<{ status: number; body: any }> => {
const t0 = Date.now();
const out = await v2Write(parsed, memoryOverride);
Expand All @@ -139,6 +140,7 @@ export const executeWriteAndRecord = async (
latencyMs,
inferMode: parsed.infer_preferred !== false,
decisions: (out as any).decisions ?? undefined,
agentId,
});
if (GRAPH_ENABLED) {
const graphRelations: any[] = (out as any).relations || [];
Expand All @@ -149,6 +151,7 @@ export const executeWriteAndRecord = async (
entitiesAdded: addedEntities.length,
relationsAdded: graphRelations.length,
latencyMs,
agentId,
});
}
const body = { ok: true, data: out };
Expand All @@ -173,6 +176,8 @@ export const handleV2Write = async (

runtimeStats.requests.add += 1;

const reqAgentId = (req as any).agent?.id as string | undefined;

if (parsed.data.async) {
const inFlightCount = [...asyncJobs.values()].filter(j => j.status === "pending" || j.status === "running").length;
if (inFlightCount >= ASYNC_JOB_MAX) {
Expand Down Expand Up @@ -201,7 +206,7 @@ export const handleV2Write = async (
job.status = "running";
try {
const noopIdem = { type: "none" as const };
const { body } = await executeWriteAndRecord(parsed.data, noopIdem, memoryOverride);
const { body } = await executeWriteAndRecord(parsed.data, noopIdem, memoryOverride, reqAgentId);
job.status = "completed";
job.completed_at = new Date().toISOString();
job.result = body.data;
Expand All @@ -216,7 +221,7 @@ export const handleV2Write = async (
return res.status(202).json(acceptedBody);
}

const { status, body } = await executeWriteAndRecord(parsed.data, idem, memoryOverride);
const { status, body } = await executeWriteAndRecord(parsed.data, idem, memoryOverride, reqAgentId);
return res.status(status).json(body);
} catch (err: any) {
return v2Err(res, 500, "INTERNAL_ERROR", String(err?.message || err));
Expand Down
Loading
Loading