Skip to content

Harden recall and capture reliability #78

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’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
ishaanxgupta wants to merge 1 commit into
base: codex/human-memory-capture
Choose a base branch
from
Open

Harden recall and capture reliability #78

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
24 changes: 17 additions & 7 deletions src/index.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,10 @@ import { tool } from "@opencode-ai/plugin";

import { AGENT_ENTITY_CONTEXT } from "./services/entity-context.js";
import { supermemoryClient } from "./services/client.js";
import { formatContextForPrompt } from "./services/context.js";
import {
formatContextForPrompt,
getInjectedProfileFactTexts,
} from "./services/context.js";
import { createCaptureHook } from "./services/capture.js";
import {
buildDirectRecallContext,
Expand Down Expand Up @@ -204,10 +207,7 @@ export const SupermemoryPlugin: Plugin = async (ctx: PluginInput) => {
suppressTexts: isFirstMessage
? profileRequest.then((result) =>
result?.success && result.profile
? [
...result.profile.static,
...result.profile.dynamic,
]
? getInjectedProfileFactTexts(result)
: [],
)
: undefined,
Expand Down Expand Up @@ -615,7 +615,15 @@ export const SupermemoryPlugin: Plugin = async (ctx: PluginInput) => {
await compactionHook.event(input);
}
if (captureHook) {
await captureHook.event(input);
if (input.event.type === "session.idle") {
void captureHook.event(input).catch((error) => {
log("[capture] background idle capture failed", {
error: String(error),
});
});
} else {
await captureHook.event(input);
}
}
},
};
Expand Down Expand Up @@ -654,7 +662,9 @@ function formatSearchResults(
const r = hit.result;
const result = {
content: formatRecallHit(hit),
similarity: Math.round(hit.similarity * 100),
...(hit.similarity === undefined
? {}
: { similarity: Math.round(hit.similarity * 100) }),
...(hit.title ? { title: hit.title } : {}),
...(hit.filepath ? { filepath: hit.filepath } : {}),
};
Expand Down
76 changes: 76 additions & 0 deletions src/services/capture.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,13 +2,15 @@ import { describe, expect, test } from "bun:test";
import type { Part } from "@opencode-ai/sdk";

import {
AUTOMATIC_CAPTURE_TIMEOUT_MS,
buildCadenceBatches,
buildCaptureTurns,
buildSessionEndBatch,
createCaptureHook,
getCaptureId,
type SessionMessage,
} from "./capture.js";
import { SupermemoryClient } from "./client.js";
import type { ResolvedTags } from "./tags.js";

function textPart(
Expand Down Expand Up @@ -122,6 +124,7 @@ describe("automatic conversation capture", () => {
conversationId: string;
metadata?: Record<string, string | number | boolean>;
customId?: string;
timeoutMs?: number;
}> = [];
const ctx = {
directory: "/repo",
Expand Down Expand Up @@ -155,6 +158,7 @@ describe("automatic conversation capture", () => {
conversationId,
metadata,
customId: options?.customId,
timeoutMs: options?.timeoutMs,
});
return { success: true };
},
Expand Down Expand Up @@ -183,6 +187,7 @@ describe("automatic conversation capture", () => {

expect(writes).toHaveLength(1);
expect(writes[0]?.metadata?.captureReason).toBe("cadence");
expect(writes[0]?.timeoutMs).toBe(AUTOMATIC_CAPTURE_TIMEOUT_MS);

messages = conversation(4);
await hook.event({
Expand All @@ -208,4 +213,75 @@ describe("automatic conversation capture", () => {
expect(writes[1]?.metadata?.captureReason).toBe("session_end");
expect(writes[0]?.customId).not.toBe(writes[1]?.customId);
});

test("retains terminal capture state after failure and retries with bounded SDK options", async () => {
const sdkOptions: Array<{ timeout?: number; maxRetries?: number }> = [];
let attempts = 0;
let readAttempts = 0;
const memoryClient = new SupermemoryClient();
(
memoryClient as unknown as {
client: {
memories: {
add: (
payload: unknown,
options?: { timeout?: number; maxRetries?: number },
) => Promise<{ id: string }>;
};
};
}
).client = {
memories: {
add: async (_payload, options) => {
sdkOptions.push(options ?? {});
attempts += 1;
if (attempts === 1) throw new Error("temporary capture failure");
return { id: "memory-1" };
},
},
};

const hook = createCaptureHook(
{
directory: "/repo",
client: {
session: {
messages: async () => {
readAttempts += 1;
if (readAttempts === 1) {
throw new Error("temporary transcript read failure");
}
return { data: conversation(1) };
},
},
},
},
{
canonical: "repo_test__0123456789abcdef",
user: "repo_test__0123456789abcdef",
project: "repo_test__0123456789abcdef",
projectId: "0123456789abcdef",
projectName: "test",
personalReads: [],
projectReads: [],
allReads: [],
},
{ captureEveryNTurns: 0, memoryClient },
);

for (let attempt = 0; attempt < 4; attempt += 1) {
await hook.event({
event: {
type: "session.deleted",
properties: { info: { id: "session-1" } },
},
});
}

expect(sdkOptions).toEqual([
{ timeout: AUTOMATIC_CAPTURE_TIMEOUT_MS, maxRetries: 0 },
{ timeout: AUTOMATIC_CAPTURE_TIMEOUT_MS, maxRetries: 0 },
]);
expect(readAttempts).toBe(3);
});
});
108 changes: 70 additions & 38 deletions src/services/capture.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,8 @@ import { log } from "./logger.js";
import { isFullyPrivate, stripPrivateContent } from "./privacy.js";
import type { ResolvedTags } from "./tags.js";

export const AUTOMATIC_CAPTURE_TIMEOUT_MS = 3_000;

interface CaptureMessageInfo {
id: string;
role: string;
Expand Down Expand Up @@ -56,6 +58,7 @@ interface ConversationWriter {
options?: {
defaultEntityContext?: string;
customId?: string;
timeoutMs?: number;
},
) => Promise<{ success: boolean; error?: string }>;
}
Expand Down Expand Up @@ -237,35 +240,44 @@ export function createCaptureHook(
sessionID: string,
batch: CaptureBatch,
reason: "cadence" | "session_end",
): Promise<void> {
): Promise<boolean> {
const captureId = getCaptureId(sessionID, batch);
if (completedCaptureIds.has(captureId)) return;
if (completedCaptureIds.has(captureId)) return true;

const messages = batch.turns.flatMap((turn) => turn.messages);
if (messages.length === 0) {
completedCaptureIds.add(captureId);
return;
return true;
}

const result = await memoryClient.ingestConversation(
`${sessionID}:${batch.startTurn}-${batch.endTurn}`,
messages,
[tags.canonical],
{
project: tags.projectName,
sm_project_id: tags.projectId,
sm_scope: "personal",
sm_capture_mode: "automatic",
captureReason: reason,
sessionId: sessionID,
turnStart: batch.startTurn,
turnEnd: batch.endTurn,
},
{
defaultEntityContext: AGENT_ENTITY_CONTEXT,
customId: captureId,
},
);
let result: { success: boolean; error?: string };
try {
result = await memoryClient.ingestConversation(
`${sessionID}:${batch.startTurn}-${batch.endTurn}`,
messages,
[tags.canonical],
{
project: tags.projectName,
sm_project_id: tags.projectId,
sm_scope: "personal",
sm_capture_mode: "automatic",
captureReason: reason,
sessionId: sessionID,
turnStart: batch.startTurn,
turnEnd: batch.endTurn,
},
{
defaultEntityContext: AGENT_ENTITY_CONTEXT,
customId: captureId,
timeoutMs: AUTOMATIC_CAPTURE_TIMEOUT_MS,
},
);
} catch (error) {
result = {
success: false,
error: error instanceof Error ? error.message : String(error),
};
}

if (result.success) {
completedCaptureIds.add(captureId);
Expand All @@ -275,7 +287,7 @@ export function createCaptureHook(
startTurn: batch.startTurn,
endTurn: batch.endTurn,
});
return;
return true;
}

log("[capture] failed to save conversation batch", {
Expand All @@ -285,26 +297,42 @@ export function createCaptureHook(
endTurn: batch.endTurn,
error: result.error,
});
return false;
}

async function captureCadence(
sessionID: string,
turns: CaptureTurn[],
): Promise<void> {
): Promise<boolean> {
let complete = true;
for (const batch of buildCadenceBatches(turns, captureEveryNTurns)) {
await saveBatch(sessionID, batch, "cadence");
if (!(await saveBatch(sessionID, batch, "cadence"))) {
complete = false;
}
}
return complete;
}

async function captureSessionEnd(sessionID: string): Promise<void> {
const turns = snapshots.get(sessionID);
if (!turns) return;
async function captureSessionEnd(sessionID: string): Promise<boolean> {
let turns = snapshots.get(sessionID);
if (!turns) {
try {
turns = await refreshSnapshot(sessionID);
} catch (error) {
log("[capture] failed to read terminal session", {
sessionID,
error: String(error),
});
return false;
}
}

await captureCadence(sessionID, turns);
const cadenceComplete = await captureCadence(sessionID, turns);
const finalBatch = buildSessionEndBatch(turns, captureEveryNTurns);
if (finalBatch) {
await saveBatch(sessionID, finalBatch, "session_end");
}
const finalComplete = finalBatch
? await saveBatch(sessionID, finalBatch, "session_end")
: true;
return cadenceComplete && finalComplete;
}

async function runExclusive(
Expand Down Expand Up @@ -336,6 +364,7 @@ export function createCaptureHook(
if (event.type === "session.idle") {
const sessionID = props?.sessionID as string | undefined;
if (!sessionID) return;
activeSessions.add(sessionID);

await runExclusive(sessionID, async () => {
try {
Expand All @@ -355,11 +384,13 @@ export function createCaptureHook(
const sessionInfo = props?.info as { id?: string } | undefined;
const sessionID = sessionInfo?.id;
if (!sessionID) return;
activeSessions.add(sessionID);

await runExclusive(sessionID, async () => {
await captureSessionEnd(sessionID);
snapshots.delete(sessionID);
activeSessions.delete(sessionID);
if (await captureSessionEnd(sessionID)) {
snapshots.delete(sessionID);
activeSessions.delete(sessionID);
}
});
return;
}
Expand All @@ -368,9 +399,10 @@ export function createCaptureHook(
await Promise.all(
[...activeSessions].map((sessionID) =>
runExclusive(sessionID, async () => {
await captureSessionEnd(sessionID);
snapshots.delete(sessionID);
activeSessions.delete(sessionID);
if (await captureSessionEnd(sessionID)) {
snapshots.delete(sessionID);
activeSessions.delete(sessionID);
}
}),
),
);
Expand Down
Loading
Loading