Skip to content

fix: resilient worker prompting with agent-not-found retry #67

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

Merged
BYK merged 1 commit into from
Apr 14, 2026
Merged

fix: resilient worker prompting with agent-not-found retry #67

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
3 changes: 3 additions & 0 deletions AGENTS.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,9 @@
<!-- lore:019d15f7-4d00-781e-9512-a4f3e3109f18 -->
* **OpenCode plugin SDK has no embedding API — vector search blocked**: The OpenCode plugin SDK (\`@opencode-ai/plugin\`, \`@opencode-ai/sdk\`) exposes only session/chat/tool operations. There is no \`client.embed()\`, embeddings endpoint, or raw model inference API. The only LLM access is \`client.session.prompt()\` which creates full chat roundtrips through the agentic loop. This means Lore cannot do vector/embedding search without either: (1) OpenCode adding an embedding API, or (2) direct \`fetch()\` to provider APIs bypassing the SDK (fragile — requires key extraction from \`client.config.providers()\`). The FTS5 + RRF search infrastructure is designed to be additive — vector search would layer on top as another RRF input list, not replace BM25.

<!-- lore:019d8c54-e51c-7fe8-87ba-273269c39b7a -->
* **Worker session prompt helper with agent-not-found retry**: src/worker.ts owns workerSessionIDs Set, isWorkerSession(), and promptWorker(). promptWorker() calls session.prompt() and uses the return value directly (no redundant session.messages() call). On 'agent not found' errors (detected via regex on JSON.stringify(result.error)), it retries once without the agent parameter on a fresh session. All callers (distillation×2, curator×2, search×1) use this shared helper. Session rotation (deleting from the caller's Map) happens after every call. The retry creates a new child session via client.session.create() and registers its ID in workerSessionIDs.

### Decision

<!-- lore:019c904b-7924-7187-8471-8ad2423b8946 -->
Expand Down
69 changes: 22 additions & 47 deletions src/curator.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,8 +2,9 @@ import type { createOpencodeClient } from "@opencode-ai/sdk";
import { config } from "./config";
import * as temporal from "./temporal";
import * as ltm from "./ltm";
import * as log from "./log";
import { CURATOR_SYSTEM, curatorUser, CONSOLIDATION_SYSTEM, consolidationUser } from "./prompt";
import { workerSessionIDs } from "./distillation";
import { workerSessionIDs, promptWorker } from "./worker";

/**
* Maximum length (chars) for a single knowledge entry's content.
Expand Down Expand Up @@ -103,33 +104,18 @@ export async function run(input: {
{ type: "text" as const, text: `${CURATOR_SYSTEM}\n\n${userContent}` },
];

await input.client.session.prompt({
path: { id: workerID },
body: {
parts,
agent: "lore-curator",
...(model ? { model } : {}),
},
const responseText = await promptWorker({
client: input.client,
workerID,
parts,
agent: "lore-curator",
model,
sessionMap: workerSessions,
sessionKey: input.sessionID,
});
if (!responseText) return { created: 0, updated: 0, deleted: 0 };

const msgs = await input.client.session.messages({
path: { id: workerID },
query: { limit: 2 },
});
// Rotate worker session so the next call starts fresh — prevents
// accumulating multiple assistant messages with reasoning/thinking parts,
// which providers reject ("Multiple reasoning_opaque values").
workerSessions.delete(input.sessionID);

const last = msgs.data?.at(-1);
if (!last || last.info.role !== "assistant")
return { created: 0, updated: 0, deleted: 0 };

const responsePart = last.parts.find((p) => p.type === "text");
if (!responsePart || responsePart.type !== "text")
return { created: 0, updated: 0, deleted: 0 };

const ops = parseOps(responsePart.text);
const ops = parseOps(responseText);
let created = 0;
let updated = 0;
let deleted = 0;
Expand Down Expand Up @@ -230,29 +216,18 @@ export async function consolidate(input: {
{ type: "text" as const, text: `${CONSOLIDATION_SYSTEM}\n\n${userContent}` },
];

await input.client.session.prompt({
path: { id: workerID },
body: {
parts,
agent: "lore-curator",
...(model ? { model } : {}),
},
const responseText = await promptWorker({
client: input.client,
workerID,
parts,
agent: "lore-curator",
model,
sessionMap: workerSessions,
sessionKey: input.sessionID,
});
if (!responseText) return { updated: 0, deleted: 0 };

const msgs = await input.client.session.messages({
path: { id: workerID },
query: { limit: 2 },
});
// Rotate worker session — see run() comment.
workerSessions.delete(input.sessionID);

const last = msgs.data?.at(-1);
if (!last || last.info.role !== "assistant") return { updated: 0, deleted: 0 };

const responsePart = last.parts.find((p) => p.type === "text");
if (!responsePart || responsePart.type !== "text") return { updated: 0, deleted: 0 };

const ops = parseOps(responsePart.text);
const ops = parseOps(responseText);
let updated = 0;
let deleted = 0;

Expand Down
94 changes: 38 additions & 56 deletions src/distillation.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,21 +11,17 @@ import {
recursiveUser,
} from "./prompt";
import { needsUrgentDistillation } from "./gradient";
import { workerSessionIDs, promptWorker } from "./worker";

// Re-export for backwards compat — index.ts and others may still import from here.
export { workerSessionIDs };

type Client = ReturnType<typeof createOpencodeClient>;
type TemporalMessage = temporal.TemporalMessage;

// Worker sessions keyed by parent session ID — hidden children, one per source session
const workerSessions = new Map<string, string>();

// Set of worker session IDs — used to skip storage and distillation for worker sessions
// Exported so curator.ts can register its own worker sessions here too
export const workerSessionIDs = new Set<string>();

export function isWorkerSession(sessionID: string): boolean {
return workerSessionIDs.has(sessionID);
}

async function ensureWorkerSession(
client: Client,
parentID: string,
Expand Down Expand Up @@ -108,6 +104,17 @@ function latestObservations(
return row?.observations || undefined;
}

/** Safely parse the source_ids JSON column. Defaults to [] on corrupt data. */
export function parseSourceIds(raw: string): string[] {
try {
const parsed = JSON.parse(raw);
return Array.isArray(parsed) ? parsed : [];
} catch {
log.warn("corrupt source_ids in distillation, defaulting to []");
return [];
}
}

export type Distillation = {
id: string;
project_id: string;
Expand Down Expand Up @@ -141,7 +148,7 @@ export function loadForSession(
}>;
return rows.map((r) => ({
...r,
source_ids: JSON.parse(r.source_ids) as string[],
source_ids: parseSourceIds(r.source_ids),
}));
}

Expand Down Expand Up @@ -208,7 +215,7 @@ function loadGen0(projectPath: string, sessionID: string): Distillation[] {
}>;
return rows.map((r) => ({
...r,
source_ids: JSON.parse(r.source_ids) as string[],
source_ids: parseSourceIds(r.source_ids),
}));
}

Expand Down Expand Up @@ -242,7 +249,7 @@ function resetOrphans(projectPath: string, sessionID: string): number {
.all(pid, sessionID) as Array<{ source_ids: string }>;
const covered = new Set<string>();
for (const r of rows) {
for (const id of JSON.parse(r.source_ids) as string[]) covered.add(id);
for (const id of parseSourceIds(r.source_ids)) covered.add(id);
}
if (rows.length === 0) {
// No distillations at all — reset everything to undistilled
Expand Down Expand Up @@ -375,32 +382,18 @@ async function distillSegment(input: {
{ type: "text" as const, text: `${DISTILLATION_SYSTEM}\n\n${userContent}` },
];

await input.client.session.prompt({
path: { id: workerID },
body: {
parts,
agent: "lore-distill",
...(model ? { model } : {}),
},
const responseText = await promptWorker({
client: input.client,
workerID,
parts,
agent: "lore-distill",
model,
sessionMap: workerSessions,
sessionKey: input.sessionID,
});
if (!responseText) return null;

// Read the response
const msgs = await input.client.session.messages({
path: { id: workerID },
query: { limit: 2 },
});
// Rotate worker session so the next call starts fresh — prevents
// accumulating multiple assistant messages with reasoning/thinking parts,
// which providers reject ("Multiple reasoning_opaque values").
workerSessions.delete(input.sessionID);

const last = msgs.data?.at(-1);
if (!last || last.info.role !== "assistant") return null;

const responsePart = last.parts.find((p) => p.type === "text");
if (!responsePart || responsePart.type !== "text") return null;

const result = parseDistillationResult(responsePart.text);
const result = parseDistillationResult(responseText);
if (!result) return null;

const distillId = storeDistillation({
Expand Down Expand Up @@ -437,29 +430,18 @@ async function metaDistill(input: {
{ type: "text" as const, text: `${RECURSIVE_SYSTEM}\n\n${userContent}` },
];

await input.client.session.prompt({
path: { id: workerID },
body: {
parts,
agent: "lore-distill",
...(model ? { model } : {}),
},
const responseText = await promptWorker({
client: input.client,
workerID,
parts,
agent: "lore-distill",
model,
sessionMap: workerSessions,
sessionKey: input.sessionID,
});
if (!responseText) return null;

const msgs = await input.client.session.messages({
path: { id: workerID },
query: { limit: 2 },
});
// Rotate worker session — see distillSegment() comment.
workerSessions.delete(input.sessionID);

const last = msgs.data?.at(-1);
if (!last || last.info.role !== "assistant") return null;

const responsePart = last.parts.find((p) => p.type === "text");
if (!responsePart || responsePart.type !== "text") return null;

const result = parseDistillationResult(responsePart.text);
const result = parseDistillationResult(responseText);
if (!result) return null;

// Store the meta-distillation at generation N+1
Expand Down
3 changes: 2 additions & 1 deletion src/index.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@ import { shouldImport, importFromFile, exportToFile } from "./agents-file";
import * as latReader from "./lat-reader";
import * as embedding from "./embedding";
import * as log from "./log";
import { isWorkerSession } from "./worker";

/**
* Detect whether an error from session.error is a context overflow ("prompt too long").
Expand Down Expand Up @@ -167,7 +168,7 @@ export const LorePlugin: Plugin = async (ctx) => {
const skipSessions = new Set<string>();

async function shouldSkip(sessionID: string): Promise<boolean> {
if (distillation.isWorkerSession(sessionID)) return true;
if (isWorkerSession(sessionID)) return true;
if (skipSessions.has(sessionID)) return true;
if (activeSessions.has(sessionID)) return false; // already known good
// First encounter — check if this is a child session.
Expand Down
39 changes: 13 additions & 26 deletions src/search.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -272,7 +272,7 @@ export function reciprocalRankFusion<T>(
// ---------------------------------------------------------------------------

import type { createOpencodeClient } from "@opencode-ai/sdk";
import { workerSessionIDs } from "./distillation";
import { workerSessionIDs, promptWorker } from "./worker";
import { QUERY_EXPANSION_SYSTEM } from "./prompt";
import * as log from "./log";

Expand Down Expand Up @@ -326,39 +326,26 @@ export async function expandQuery(
];

// Race the LLM call against a timeout
const result = await Promise.race([
client.session.prompt({
path: { id: workerID },
body: {
parts,
agent: "lore-query-expand",
...(model ? { model } : {}),
},
const responseText = await Promise.race([
promptWorker({
client,
workerID,
parts,
agent: "lore-query-expand",
model,
sessionMap: expansionWorkerSessions,
sessionKey: sessionID,
}),
new Promise<null>((resolve) => setTimeout(() => resolve(null), TIMEOUT_MS)),
]);

// Rotate worker session so the next call starts fresh
expansionWorkerSessions.delete(sessionID);

if (!result) {
log.info("query expansion timed out, using original query");
if (!responseText) {
log.info("query expansion timed out or failed, using original query");
return [query];
}

// Read the response
const msgs = await client.session.messages({
path: { id: workerID },
query: { limit: 2 },
});
const last = msgs.data?.at(-1);
if (!last || last.info.role !== "assistant") return [query];

const responsePart = last.parts.find((p) => p.type === "text");
if (!responsePart || responsePart.type !== "text") return [query];

// Parse JSON array from response
const cleaned = responsePart.text
const cleaned = responseText
.trim()
.replace(/^```json?\s*/i, "")
.replace(/\s*```$/i, "");
Expand Down
Loading
Loading