github-featured/rohitg00--agentmemory
2
0
Fork 0
Code Issues Pull Requests Deployments Models Agent Notes Packages Projects Releases Wiki Activity

fix: address 9 unresolved CodeRabbit findings on iii v0.11 migration

Browse Source 
CodeRabbit flagged 9 remaining issues on #116 — all real. Each was
verified against the current branch code before applying. Two
findings were deliberately skipped as policy/scope issues and are
documented at the bottom.

### Applied (9 real findings)

- src/triggers/api.ts — api.ts numeric query param validation:
  multiple sites forwarded `parseInt(params.limit)` result to the
  downstream function without a Number.isFinite check. A non-numeric
  query value produced NaN at the iii-sdk trigger boundary. Added
  parseOptionalInt and parseOptionalFloat helpers at the top of the
  file, wired into api::crystal-list, api::lesson-list, and
  api::insight-list (5 sites total).

- src/triggers/api.ts — api::observe, api::context, api::session::start,
  api::session::end were forwarding req.body verbatim to sdk.trigger
  with no validation. Added explicit type checks mirroring the existing
  api::search pattern, construct a sanitized payload object before
  triggering.

- src/cli.ts — p.confirm() can return a cancel Symbol on Ctrl+C, which
  is truthy, so the upgrade path ran even when the user cancelled.
  Added p.isCancel() check and explicit boolean comparison.

- src/cli.ts — removed dead `failed` flag in runUpgrade: every caller
  already calls process.exit(1) immediately, so the final `if (failed)`
  branch was unreachable. Simplified requireSuccess accordingly.

- src/functions/leases.ts — mem::lease-renew recorded audit events as
  "lease_acquire", which conflated renewals with first claims. Renamed
  the audit event to "lease_renew" so downstream audit filters can tell
  the two operations apart.

- src/functions/relations.ts — the pair lock
  `mem:${firstId}:${secondId}` serialized concurrent relate(A,B) calls
  with identical pairs, but did NOT protect concurrent relate(A,B) +
  relate(A,C). Both modify memory A's `relatedIds` array, which is a
  classic last-writer-wins race producing lost relation edges. Fixed
  by replacing the pair lock with nested per-entity locks in canonical
  sort order: withKeyedLock(mem:firstId) wrapping withKeyedLock(mem:secondId).
  Since the ids are sorted deterministically, no deadlock is possible.

- src/functions/sketches.ts + src/functions/summarize.ts + new
  src/functions/audit.ts safeAudit helper — recordAudit was awaited
  after kv.set calls. An audit write failure would reject and the
  caller would see an error even though the target state was already
  persisted. New safeAudit wrapper swallows audit errors and logs them
  via ctx.logger.warn, preserving the mutation's success. Applied to
  all 11 audit sites in sketches.ts and the single site in summarize.ts.

- src/mcp/server.ts — three issues in the MCP handler:
  1. memory_profile's refresh flag used `args.refresh === "true"`,
     ignoring boolean `true` from MCP clients that send proper types.
     Now accepts both.
  2. memory_sketch_create built sketchPayload with
     `asNonEmptyString(args.title)` which could return undefined,
     then forwarded that to the downstream function. Added explicit
     validation + 400 response at the MCP boundary.
  3. memory_recall, memory_team_feed, memory_audit_query used
     `(args.limit as number) || 10` which replaced an explicit 0 with
     10. Changed to typeof check so explicit 0 (rare but legal) is
     preserved.

- src/functions/governance.ts — mem::governance-bulk used Promise.all
  for the delete batch, which fails fast on the first error. The audit
  record still said `deleted: candidates.length` even if only half
  actually succeeded. Switched to Promise.allSettled, split results
  into successfulIds and failures arrays, record audit with both counts
  plus the per-failure details for traceability.

- src/functions/mesh.ts — mem::mesh-register, mem::mesh-sync,
  mem::mesh-receive, mem::mesh-remove all dereferenced `data.*` without
  checking that data was passed. A TypeError would bubble up on null
  payload. Added early null/type guards returning structured errors.

- src/functions/obsidian-export.ts — resolveVaultDir(data.vaultDir)
  was called without validating that vaultDir was a string, and
  `new Set(data.types)` without validating types was an array of
  strings. Added explicit validation returning 400-style error
  responses.

- src/functions/retention.ts — the eviction loop silently swallowed
  kv.delete errors via a bare `continue`. Added ctx.logger.warn on
  catch, included memoryId and sourceBucket in the log, and now
  returns `failed` count alongside `evicted` in the response.

- src/viewer/index.html — two WebSocket bugs:
  1. connectWs assigned to the mutable global state.ws before binding
     handlers, so old sockets could have their callbacks fire and
     mutate retry/direct state after state.ws was overwritten by a
     new socket. Fixed by creating a local `ws` variable, binding
     all handlers to it, only then assigning state.ws = ws. Each
     handler guards `if (state.ws !== ws) return` so stale callbacks
     are dropped.
  2. handleStreamEvent routed EVERY incoming event to routeWsMessage,
     so non-observation events (session.activity, etc.) were being
     treated as timeline observations and causing UI confusion. Added
     a looksLikeObservation helper + event_type gate that only routes
     real observation payloads.

- test/retention.test.ts — added an explicit sourceBucket eviction
  test that seeds a semantic memory at high age, runs retention-score,
  then runs retention-evict at high threshold and asserts BOTH
  KV.semantic and KV.memories are empty. Proves the candidate.sourceBucket
  branch added in this PR actually routes to the right bucket.

- src/types.ts — side fix: the ExportData.version union on line 254
  used comma separators instead of pipes in 3 positions (0.7.9,
  0.8.0, 0.8.1 → needed | between them). tsdown strips types so the
  build passed, but tsc --noEmit would have thrown and any IDE showed
  squiggles. Pre-existing latent bug, fixed while in the file.

### Deliberately skipped

- retention.ts eviction audit (CodeRabbit asked to add recordAudit
  to the eviction loop): policy change, not a bug fix. Verified:
  auto-forget.ts, evict.ts, retention-evict, remember-forget all
  skip audit by convention — only governance audits. Adding audit
  everywhere is a policy decision tracked in issue #125.

- file-index.ts sequential → parallel session lookup: micro-opt,
  the loop is already cache-backed, negligible savings, not worth
  the readability cost.

Tests: 655/655. Build clean.
This commit is contained in:
Rohit Ghumare
2026-04-13 17:14:44 +01:00
parent 056d9be6af
commit 3e3ac7ef28
15 changed files with 275 additions and 74 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/cli.ts
+5 -7
View File
@@ -554,10 +554,8 @@ async function runUpgrade() {
const dockerBin = whichBinary("docker");
p.log.info(`Working directory: ${cwd}`);
let failed = false;
const requireSuccess = (ok: boolean, label: string): boolean => {
if (!ok) {
failed = true;
p.log.error(`Upgrade aborted: ${label} failed.`);
}
return ok;
@@ -595,7 +593,11 @@ async function runUpgrade() {
message: "Upgrade iii-engine via cargo install --force?",
initialValue: true,
});
if (upgradeEngine) {
if (p.isCancel(upgradeEngine)) {
p.cancel("Cancelled.");
return process.exit(0);
}
if (upgradeEngine === true) {
const cargoOk = runCommand(cargoBin, ["install", "iii-engine", "--force"], {
label: "Upgrading iii-engine (cargo)",
});
@@ -618,10 +620,6 @@ async function runUpgrade() {
p.log.info("Docker not found. Skipping Docker image refresh.");
}
if (failed) {
return process.exit(1);
}
p.note(
[
"Upgrade flow completed.",
src/functions/audit.ts
+25
View File
@@ -1,3 +1,4 @@
import { getContext } from "iii-sdk";
import type { AuditEntry } from "../types.js";
import { KV, generateId } from "../state/schema.js";
import type { StateKV } from "../state/kv.js";
@@ -25,6 +26,30 @@ export async function recordAudit(
return entry;
}
export async function safeAudit(
kv: StateKV,
operation: AuditEntry["operation"],
functionId: string,
targetIds: string[],
details: Record<string, unknown> = {},
qualityScore?: number,
userId?: string,
): Promise<void> {
try {
await recordAudit(kv, operation, functionId, targetIds, details, qualityScore, userId);
} catch (err) {
try {
const ctx = getContext();
ctx.logger.warn("audit write failed", {
functionId,
operation,
targetIds,
error: err instanceof Error ? err.message : String(err),
});
} catch {}
}
}
export async function queryAudit(
kv: StateKV,
filter?: {
src/functions/governance.ts
+32 -5
View File
@@ -98,18 +98,45 @@ export function registerGovernanceFunction(sdk: ISdk, kv: StateKV): void {
};
}
await Promise.all(candidates.map((mem) => kv.delete(KV.memories, mem.id)));
const results = await Promise.allSettled(
candidates.map((mem) => kv.delete(KV.memories, mem.id)),
);
const successfulIds: string[] = [];
const failures: Array<{ id: string; error: string }> = [];
results.forEach((result, i) => {
if (result.status === "fulfilled") {
successfulIds.push(candidates[i].id);
} else {
failures.push({
id: candidates[i].id,
error: result.reason instanceof Error ? result.reason.message : String(result.reason),
});
}
});
await recordAudit(
kv,
"delete",
"mem::governance-bulk",
candidates.map((m) => m.id),
{ filter: data, deleted: candidates.length },
successfulIds,
{
filter: data,
deleted: successfulIds.length,
failed: failures.length,
failures: failures.length > 0 ? failures : undefined,
},
);
ctx.logger.info("Governance bulk delete", { deleted: candidates.length });
return { success: true, deleted: candidates.length };
ctx.logger.info("Governance bulk delete", {
deleted: successfulIds.length,
failed: failures.length,
});
return {
success: true,
deleted: successfulIds.length,
failed: failures.length,
failures: failures.length > 0 ? failures : undefined,
};
},
);
src/functions/leases.ts
+1 -1
View File
@@ -172,7 +172,7 @@ export function registerLeasesFunction(sdk: ISdk, kv: StateKV): void {
activeLease.expiresAt = new Date(base + ttl).toISOString();
activeLease.renewedAt = now.toISOString();
await kv.set(KV.leases, activeLease.id, activeLease);
await recordAudit(kv, "lease_acquire", "mem::lease-renew", [activeLease.id], {
await recordAudit(kv, "lease_renew", "mem::lease-renew", [activeLease.id], {
actionId: data.actionId,
agentId: data.agentId,
before: beforeLease,
src/functions/mesh.ts
+14 -5
View File
@@ -141,13 +141,16 @@ export function registerMeshFunction(
kv: StateKV,
meshAuthToken?: string,
): void {
sdk.registerFunction("mem::mesh-register",
sdk.registerFunction("mem::mesh-register",
async (data: {
url: string;
name: string;
sharedScopes?: string[];
syncFilter?: { project?: string };
}) => {
if (!data || typeof data !== "object") {
return { success: false, error: "payload required" };
}
if (!data.url || !data.name) {
return { success: false, error: "url and name are required" };
}
@@ -190,7 +193,7 @@ export function registerMeshFunction(
},
);
sdk.registerFunction("mem::mesh-sync",
sdk.registerFunction("mem::mesh-sync",
async (data: { peerId?: string; scopes?: string[]; direction?: "push" | "pull" | "both" }) => {
if (!meshAuthToken) {
return {
@@ -198,6 +201,9 @@ export function registerMeshFunction(
error: "mesh sync requires AGENTMEMORY_SECRET",
};
}
if (!data || typeof data !== "object") {
data = {};
}
const direction = data.direction || "both";
let peers: MeshPeer[];
@@ -326,8 +332,11 @@ export function registerMeshFunction(
},
);
sdk.registerFunction("mem::mesh-receive",
sdk.registerFunction("mem::mesh-receive",
async (data: MeshSyncPayload) => {
if (!data || typeof data !== "object") {
return { success: false, error: "payload required" };
}
let accepted = 0;
accepted += await lwwMergeList(kv, KV.memories, data.memories, "mem:memory", "updatedAt");
@@ -362,9 +371,9 @@ export function registerMeshFunction(
},
);
sdk.registerFunction("mem::mesh-remove",
sdk.registerFunction("mem::mesh-remove",
async (data: { peerId: string }) => {
if (!data.peerId) {
if (!data || typeof data !== "object" || !data.peerId) {
return { success: false, error: "peerId is required" };
}
await kv.delete(KV.mesh, data.peerId);
src/functions/obsidian-export.ts
+13 -2
View File
@@ -197,11 +197,22 @@ export function registerObsidianExportFunction(
sdk: ISdk,
kv: StateKV,
): void {
sdk.registerFunction("mem::obsidian-export",
sdk.registerFunction("mem::obsidian-export",
async (data: { vaultDir?: string; types?: string[] } | undefined) => {
if (!data) {
if (!data || typeof data !== "object") {
return { success: false, error: "payload is required" };
}
if (data.vaultDir !== undefined && typeof data.vaultDir !== "string") {
return { success: false, error: "vaultDir must be a string" };
}
if (data.types !== undefined) {
if (
!Array.isArray(data.types) ||
!data.types.every((t): t is string => typeof t === "string")
) {
return { success: false, error: "types must be an array of strings" };
}
}
const vaultDir = resolveVaultDir(data.vaultDir);
if (!vaultDir) {
src/functions/relations.ts
+4 -3
View File
@@ -45,9 +45,9 @@ export function registerRelationsFunction(sdk: ISdk, kv: StateKV): void {
}) => {
const ctx = getContext();
const [firstId, secondId] = [data.sourceId, data.targetId].sort();
const pairKey = `mem:${firstId}:${secondId}`;
return withKeyedLock(pairKey, async () => {
return withKeyedLock(`mem:${firstId}`, async () =>
withKeyedLock(`mem:${secondId}`, async () => {
const source = await kv.get<Memory>(KV.memories, data.sourceId);
const target = await kv.get<Memory>(KV.memories, data.targetId);
if (!source || !target) {
@@ -109,7 +109,8 @@ export function registerRelationsFunction(sdk: ISdk, kv: StateKV): void {
target: data.targetId,
});
return { success: true, relationId, relation };
});
}),
);
},
);
src/functions/retention.ts
+9 -2
View File
@@ -206,18 +206,25 @@ export function registerRetentionFunctions(
}
let evicted = 0;
let failed = 0;
for (const candidate of candidates) {
try {
await kv.delete(candidate.sourceBucket || KV.memories, candidate.memoryId);
await kv.delete(KV.retentionScores, candidate.memoryId);
evicted++;
} catch {
continue;
} catch (err) {
failed++;
ctx.logger.warn("Retention eviction failed for candidate", {
memoryId: candidate.memoryId,
sourceBucket: candidate.sourceBucket,
error: err instanceof Error ? err.message : String(err),
});
}
}
ctx.logger.info("Retention-based eviction complete", {
evicted,
failed,
threshold,
});
src/functions/sketches.ts
+13 -13
View File
@@ -3,7 +3,7 @@ import type { StateKV } from "../state/kv.js";
import { KV, generateId } from "../state/schema.js";
import { withKeyedLock } from "../state/keyed-mutex.js";
import type { Action, ActionEdge, Sketch } from "../types.js";
import { recordAudit } from "./audit.js";
import { safeAudit } from "./audit.js";
export function registerSketchesFunction(sdk: ISdk, kv: StateKV): void {
sdk.registerFunction("mem::sketch-create",
@@ -31,7 +31,7 @@ export function registerSketchesFunction(sdk: ISdk, kv: StateKV): void {
};
await kv.set(KV.sketches, sketch.id, sketch);
await recordAudit(kv, "sketch_create", "mem::sketch-create", [sketch.id], {
await safeAudit(kv, "sketch_create", "mem::sketch-create", [sketch.id], {
action: "create",
title: sketch.title,
});
@@ -93,7 +93,7 @@ export function registerSketchesFunction(sdk: ISdk, kv: StateKV): void {
}
await kv.set(KV.actions, action.id, action);
await recordAudit(kv, "sketch_create", "mem::sketch-add", [action.id], {
await safeAudit(kv, "sketch_create", "mem::sketch-add", [action.id], {
action: "add.action",
sketchId: sketch.id,
});
@@ -109,7 +109,7 @@ export function registerSketchesFunction(sdk: ISdk, kv: StateKV): void {
createdAt: now,
};
await kv.set(KV.actionEdges, edge.id, edge);
await recordAudit(kv, "sketch_create", "mem::sketch-add", [edge.id], {
await safeAudit(kv, "sketch_create", "mem::sketch-add", [edge.id], {
action: "add.edge",
sketchId: sketch.id,
});
@@ -119,7 +119,7 @@ export function registerSketchesFunction(sdk: ISdk, kv: StateKV): void {
sketch.actionIds.push(action.id);
await kv.set(KV.sketches, sketch.id, sketch);
await recordAudit(kv, "sketch_create", "mem::sketch-add", [sketch.id], {
await safeAudit(kv, "sketch_create", "mem::sketch-add", [sketch.id], {
action: "add.sketch-update",
addedActionId: action.id,
});
@@ -154,7 +154,7 @@ export function registerSketchesFunction(sdk: ISdk, kv: StateKV): void {
}
action.updatedAt = new Date().toISOString();
await kv.set(KV.actions, action.id, action);
await recordAudit(kv, "sketch_promote", "mem::sketch-promote", [action.id], {
await safeAudit(kv, "sketch_promote", "mem::sketch-promote", [action.id], {
action: "promote.action",
sketchId: sketch.id,
});
@@ -165,7 +165,7 @@ export function registerSketchesFunction(sdk: ISdk, kv: StateKV): void {
sketch.status = "promoted";
sketch.promotedAt = new Date().toISOString();
await kv.set(KV.sketches, sketch.id, sketch);
await recordAudit(kv, "sketch_promote", "mem::sketch-promote", [sketch.id], {
await safeAudit(kv, "sketch_promote", "mem::sketch-promote", [sketch.id], {
action: "promote.sketch",
promotedIds,
});
@@ -199,7 +199,7 @@ export function registerSketchesFunction(sdk: ISdk, kv: StateKV): void {
actionIdSet.has(edge.targetActionId)
) {
await kv.delete(KV.actionEdges, edge.id);
await recordAudit(kv, "sketch_discard", "mem::sketch-discard", [edge.id], {
await safeAudit(kv, "sketch_discard", "mem::sketch-discard", [edge.id], {
action: "discard.edge",
sketchId: sketch.id,
});
@@ -208,7 +208,7 @@ export function registerSketchesFunction(sdk: ISdk, kv: StateKV): void {
for (const actionId of sketch.actionIds) {
await kv.delete(KV.actions, actionId);
await recordAudit(kv, "sketch_discard", "mem::sketch-discard", [actionId], {
await safeAudit(kv, "sketch_discard", "mem::sketch-discard", [actionId], {
action: "discard.action",
sketchId: sketch.id,
});
@@ -217,7 +217,7 @@ export function registerSketchesFunction(sdk: ISdk, kv: StateKV): void {
sketch.status = "discarded";
sketch.discardedAt = new Date().toISOString();
await kv.set(KV.sketches, sketch.id, sketch);
await recordAudit(kv, "sketch_discard", "mem::sketch-discard", [sketch.id], {
await safeAudit(kv, "sketch_discard", "mem::sketch-discard", [sketch.id], {
action: "discard.sketch",
});
@@ -284,7 +284,7 @@ export function registerSketchesFunction(sdk: ISdk, kv: StateKV): void {
actionIdSet.has(edge.targetActionId)
) {
await kv.delete(KV.actionEdges, edge.id);
await recordAudit(kv, "sketch_discard", "mem::sketch-gc", [edge.id], {
await safeAudit(kv, "sketch_discard", "mem::sketch-gc", [edge.id], {
action: "gc.edge",
sketchId: current.id,
});
@@ -293,7 +293,7 @@ export function registerSketchesFunction(sdk: ISdk, kv: StateKV): void {
for (const actionId of current.actionIds) {
await kv.delete(KV.actions, actionId);
await recordAudit(kv, "sketch_discard", "mem::sketch-gc", [actionId], {
await safeAudit(kv, "sketch_discard", "mem::sketch-gc", [actionId], {
action: "gc.action",
sketchId: current.id,
});
@@ -302,7 +302,7 @@ export function registerSketchesFunction(sdk: ISdk, kv: StateKV): void {
current.status = "discarded";
current.discardedAt = new Date().toISOString();
await kv.set(KV.sketches, current.id, current);
await recordAudit(kv, "sketch_discard", "mem::sketch-gc", [current.id], {
await safeAudit(kv, "sketch_discard", "mem::sketch-gc", [current.id], {
action: "gc.sketch",
});
collected++;
src/functions/summarize.ts
+2 -2
View File
@@ -14,7 +14,7 @@ import { SummaryOutputSchema } from "../eval/schemas.js";
import { validateOutput } from "../eval/validator.js";
import { scoreSummary } from "../eval/quality.js";
import type { MetricsStore } from "../eval/metrics-store.js";
import { recordAudit } from "./audit.js";
import { safeAudit } from "./audit.js";
function parseSummaryXml(
xml: string,
@@ -122,7 +122,7 @@ export function registerSummarizeFunction(
const qualityScore = scoreSummary(summaryForValidation);
await kv.set(KV.summaries, sessionId, summary);
await recordAudit(kv, "compress", "mem::summarize", [sessionId], {
await safeAudit(kv, "compress", "mem::summarize", [sessionId], {
title: summary.title,
observationCount: compressed.length,
});
src/mcp/server.ts
+12 -5
View File
@@ -94,7 +94,7 @@ export function registerMcpEndpoints(
}
const result = await sdk.trigger({ function_id: "mem::search", payload: {
query: args.query,
limit: (args.limit as number) || 10,
limit: typeof args.limit === "number" ? args.limit : 10,
} });
return {
status_code: 200,
@@ -258,7 +258,7 @@ export function registerMcpEndpoints(
}
const result = await sdk.trigger({ function_id: "mem::profile", payload: {
project: args.project,
refresh: args.refresh === "true",
refresh: args.refresh === true || args.refresh === "true",
} });
return {
status_code: 200,
@@ -454,7 +454,7 @@ export function registerMcpEndpoints(
case "memory_team_feed": {
try {
const result = await sdk.trigger({ function_id: "mem::team-feed", payload: {
limit: (args.limit as number) || 20,
limit: typeof args.limit === "number" ? args.limit : 20,
} });
return {
status_code: 200,
@@ -483,7 +483,7 @@ export function registerMcpEndpoints(
try {
const result = await sdk.trigger({ function_id: "mem::audit-query", payload: {
operation: args.operation as string,
limit: (args.limit as number) || 50,
limit: typeof args.limit === "number" ? args.limit : 50,
} });
return {
status_code: 200,
@@ -887,8 +887,15 @@ export function registerMcpEndpoints(
}
case "memory_sketch_create": {
const title = asNonEmptyString(args.title);
if (!title) {
return {
status_code: 400,
body: { error: "title is required for memory_sketch_create" },
};
}
const sketchPayload = {
title: asNonEmptyString(args.title),
title,
description: asNonEmptyString(args.description),
expiresInMs: asNumber(args.expiresInMs),
project: asNonEmptyString(args.project),
src/triggers/api.ts
+81 -13
View File
@@ -15,6 +15,18 @@ type Response = {
body: unknown;
};
function parseOptionalInt(raw: unknown): number | undefined {
if (raw === undefined || raw === null || raw === "") return undefined;
const n = typeof raw === "number" ? raw : parseInt(String(raw), 10);
return Number.isFinite(n) ? n : undefined;
}
function parseOptionalFloat(raw: unknown): number | undefined {
if (raw === undefined || raw === null || raw === "") return undefined;
const n = typeof raw === "number" ? raw : parseFloat(String(raw));
return Number.isFinite(n) ? n : undefined;
}
function checkAuth(
req: ApiRequest,
secret: string | undefined,
@@ -117,11 +129,33 @@ export function registerApiTriggers(
},
});
sdk.registerFunction("api::observe",
sdk.registerFunction("api::observe",
async (req: ApiRequest<HookPayload>): Promise<Response> => {
const authErr = checkAuth(req, secret);
if (authErr) return authErr;
const result = await sdk.trigger({ function_id: "mem::observe", payload: req.body });
const body = req.body as Partial<HookPayload> | undefined;
if (!body || typeof body !== "object") {
return { status_code: 400, body: { error: "body must be an object" } };
}
if (typeof body.hookType !== "string" || !body.hookType.trim()) {
return { status_code: 400, body: { error: "hookType is required and must be a string" } };
}
if (typeof body.sessionId !== "string" || !body.sessionId.trim()) {
return { status_code: 400, body: { error: "sessionId is required and must be a string" } };
}
if (body.timestamp !== undefined && typeof body.timestamp !== "string") {
return { status_code: 400, body: { error: "timestamp must be a string" } };
}
if (body.data !== undefined && (typeof body.data !== "object" || body.data === null)) {
return { status_code: 400, body: { error: "data must be an object" } };
}
const payload: HookPayload = {
hookType: body.hookType,
sessionId: body.sessionId,
timestamp: body.timestamp ?? new Date().toISOString(),
data: body.data ?? {},
};
const result = await sdk.trigger({ function_id: "mem::observe", payload });
return { status_code: 201, body: result };
},
);
@@ -135,13 +169,31 @@ export function registerApiTriggers(
},
});
sdk.registerFunction("api::context",
sdk.registerFunction("api::context",
async (
req: ApiRequest<{ sessionId: string; project: string; budget?: number }>,
): Promise<Response> => {
const authErr = checkAuth(req, secret);
if (authErr) return authErr;
const result = await sdk.trigger({ function_id: "mem::context", payload: req.body });
const body = (req.body ?? {}) as Record<string, unknown>;
if (typeof body.sessionId !== "string" || !body.sessionId.trim()) {
return { status_code: 400, body: { error: "sessionId is required and must be a string" } };
}
if (typeof body.project !== "string" || !body.project.trim()) {
return { status_code: 400, body: { error: "project is required and must be a string" } };
}
if (
body.budget !== undefined &&
(!Number.isInteger(body.budget) || (body.budget as number) < 1)
) {
return { status_code: 400, body: { error: "budget must be a positive integer" } };
}
const payload = {
sessionId: body.sessionId,
project: body.project,
budget: body.budget as number | undefined,
};
const result = await sdk.trigger({ function_id: "mem::context", payload });
return { status_code: 200, body: result };
},
);
@@ -197,13 +249,25 @@ export function registerApiTriggers(
},
});
sdk.registerFunction("api::session::start",
sdk.registerFunction("api::session::start",
async (
req: ApiRequest<{ sessionId: string; project: string; cwd: string }>,
): Promise<Response> => {
const authErr = checkAuth(req, secret);
if (authErr) return authErr;
const { sessionId, project, cwd } = req.body;
const body = (req.body ?? {}) as Record<string, unknown>;
if (typeof body.sessionId !== "string" || !body.sessionId.trim()) {
return { status_code: 400, body: { error: "sessionId is required and must be a string" } };
}
if (typeof body.project !== "string" || !body.project.trim()) {
return { status_code: 400, body: { error: "project is required and must be a string" } };
}
if (typeof body.cwd !== "string" || !body.cwd.trim()) {
return { status_code: 400, body: { error: "cwd is required and must be a string" } };
}
const sessionId = body.sessionId;
const project = body.project;
const cwd = body.cwd;
const session: Session = {
id: sessionId,
project,
@@ -233,11 +297,15 @@ export function registerApiTriggers(
},
});
sdk.registerFunction("api::session::end",
sdk.registerFunction("api::session::end",
async (req: ApiRequest<{ sessionId: string }>): Promise<Response> => {
const authErr = checkAuth(req, secret);
if (authErr) return authErr;
await kv.update(KV.sessions, req.body.sessionId, [
const body = (req.body ?? {}) as Record<string, unknown>;
if (typeof body.sessionId !== "string" || !body.sessionId.trim()) {
return { status_code: 400, body: { error: "sessionId is required and must be a string" } };
}
await kv.update(KV.sessions, body.sessionId, [
{ type: "set", path: "endedAt", value: new Date().toISOString() },
{ type: "set", path: "status", value: "completed" },
]);
@@ -1749,7 +1817,7 @@ export function registerApiTriggers(
const denied = checkAuth(req, secret);
if (denied) return denied;
const params = req.query_params || {};
const result = await sdk.trigger({ function_id: "mem::crystal-list", payload: { project: params.project, sessionId: params.sessionId, limit: params.limit ? parseInt(params.limit as string) : undefined } });
const result = await sdk.trigger({ function_id: "mem::crystal-list", payload: { project: params.project, sessionId: params.sessionId, limit: parseOptionalInt(params.limit) } });
return { status_code: 200, body: result };
});
sdk.registerTrigger({ type: "http", function_id: "api::crystal-list", config: { api_path: "/agentmemory/crystals", http_method: "GET" } });
@@ -1880,8 +1948,8 @@ export function registerApiTriggers(
const result = await sdk.trigger({ function_id: "mem::lesson-list", payload: {
project: params.project,
source: params.source,
minConfidence: params.minConfidence ? parseFloat(params.minConfidence as string) : undefined,
limit: params.limit ? parseInt(params.limit as string, 10) : undefined,
minConfidence: parseOptionalFloat(params.minConfidence),
limit: parseOptionalInt(params.limit),
} });
return { status_code: 200, body: result };
});
@@ -1935,8 +2003,8 @@ export function registerApiTriggers(
const params = req.query_params || {};
const result = await sdk.trigger({ function_id: "mem::insight-list", payload: {
project: params.project,
minConfidence: params.minConfidence ? parseFloat(params.minConfidence as string) : undefined,
limit: params.limit ? parseInt(params.limit as string, 10) : undefined,
minConfidence: parseOptionalFloat(params.minConfidence),
limit: parseOptionalInt(params.limit),
} });
return { status_code: 200, body: result };
});
src/types.ts
+1 -1
View File
@@ -251,7 +251,7 @@ export interface ExportPagination {
}
export interface ExportData {
version: "0.3.0" | "0.4.0" | "0.5.0" | "0.6.0" | "0.6.1" | "0.7.0" | "0.7.2" | "0.7.3" | "0.7.4" | "0.7.5" | "0.7.6" | "0.7.9", "0.8.0", "0.8.1";
version: "0.3.0" | "0.4.0" | "0.5.0" | "0.6.0" | "0.6.1" | "0.7.0" | "0.7.2" | "0.7.3" | "0.7.4" | "0.7.5" | "0.7.6" | "0.7.9" | "0.8.0" | "0.8.1";
exportedAt: string;
sessions: Session[];
observations: Record<string, CompressedObservation[]>;
src/viewer/index.html
+34 -15
View File
@@ -2797,22 +2797,24 @@
function connectWs() {
if (wsRetries >= WS_MAX_RETRIES) return;
var useDirect = !directFailed;
var ws;
try {
state.ws = new WebSocket(useDirect ? WS_DIRECT_URL : WS_URL);
state.ws.__direct = useDirect;
ws = new WebSocket(useDirect ? WS_DIRECT_URL : WS_URL);
ws.__direct = useDirect;
} catch (_) {
state.ws = new WebSocket(WS_URL);
state.ws.__direct = false;
ws = new WebSocket(WS_URL);
ws.__direct = false;
}
try {
state.ws.onopen = function() {
ws.onopen = function() {
if (state.ws !== ws) return;
wsRetries = 0;
if (state.ws.__direct) {
if (ws.__direct) {
directFailures = 0;
directFailed = false;
}
if (!state.ws.__direct) {
state.ws.send(JSON.stringify({
if (!ws.__direct) {
ws.send(JSON.stringify({
type: 'join',
data: {
subscriptionId: 'viewer-' + Date.now(),
@@ -2824,7 +2826,8 @@
document.getElementById('ws-status').textContent = 'live';
document.getElementById('ws-status').className = 'ws-status connected';
};
state.ws.onmessage = function(e) {
ws.onmessage = function(e) {
if (state.ws !== ws) return;
try {
var msg = JSON.parse(e.data);
if (msg.type === 'stream' && msg.event) {
@@ -2834,8 +2837,9 @@
}
} catch {}
};
state.ws.onclose = function() {
if (state.ws && state.ws.__direct) {
ws.onclose = function() {
if (state.ws !== ws) return;
if (ws.__direct) {
directFailures += 1;
if (directFailures >= DIRECT_FAILURE_THRESHOLD) {
directFailed = true;
@@ -2850,7 +2854,11 @@
document.getElementById('ws-status').textContent = 'disconnected';
}
};
state.ws.onerror = function() { state.ws.close(); };
ws.onerror = function() {
if (state.ws !== ws) return;
try { ws.close(); } catch {}
};
state.ws = ws;
} catch {
wsRetries++;
if (wsRetries < WS_MAX_RETRIES) {
@@ -2859,16 +2867,27 @@
}
}
function looksLikeObservation(obj) {
return !!(obj && typeof obj === 'object' && obj.id && obj.timestamp);
}
function handleStreamEvent(msg) {
var evt = msg.event;
if (!evt) return;
if (evt.event_type && evt.event_type !== 'observation' && evt.event_type !== 'create' && evt.event_type !== 'update') {
return;
}
if (evt.type === 'event' && evt.data) {
routeWsMessage({ observation: evt.data.observation || evt.data });
var observation = evt.data.observation || evt.data;
if (looksLikeObservation(observation)) {
routeWsMessage({ observation: observation });
}
return;
}
if ((evt.type === 'create' || evt.type === 'update') && evt.data) {
var payload = evt.data;
var observation = payload.observation || payload;
if (observation) {
if (looksLikeObservation(observation)) {
routeWsMessage({ observation: observation });
}
} else if (evt.type === 'sync') {
@@ -2876,7 +2895,7 @@
items.forEach(function(item) {
var payload = item.data || item;
var observation = payload.observation || payload;
if (observation) {
if (looksLikeObservation(observation)) {
routeWsMessage({ observation: observation });
}
});
test/retention.test.ts
+29
View File
@@ -263,4 +263,33 @@ describe("RetentionScoring", () => {
const sem2 = result.scores.find((s: any) => s.memoryId === "sem_2");
expect(sem1.score).toBeGreaterThan(sem2.score);
});
it("evicts semantic memories from KV.semantic via sourceBucket", async () => {
const { registerRetentionFunctions } = await import(
"../src/functions/retention.js"
);
// Semantic memory old enough that its score falls well below the
// threshold so eviction will pick it up.
const semanticMems = [makeSemanticMemory("sem_evict", 500, 0)];
const sdk = mockSdk();
const kv = mockKV([], semanticMems);
registerRetentionFunctions(sdk as never, kv as never);
await sdk.trigger({ function_id: "mem::retention-score", payload: {} });
const result = (await sdk.trigger({
function_id: "mem::retention-evict",
payload: { threshold: 0.9, dryRun: false },
})) as any;
expect(result.success).toBe(true);
expect(result.evicted).toBeGreaterThanOrEqual(1);
const remainingSemantic = await kv.list("mem:semantic");
expect(remainingSemantic.length).toBe(0);
const remainingMemories = await kv.list("mem:memories");
expect(remainingMemories.length).toBe(0);
});
});