alvis/AgapHost
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

Adolf memory: migrate Cognee -> Hindsight + Kimi quota tooling

Browse Source 
Memory migration (H1-H5, kb#73-77,84):
- hindsight service in openai/docker-compose.yml: LLM via Kimi (cognee-llm
  wrapper), multilingual GPU embeddings (bge-m3 via ollama), jina multilingual
  reranker; pg0 + model cache persisted
- openclaw.json/shared-mcp.json: mcp.servers cognee -> hindsight (bank "adolf")
- hindsight-openclaw-plugin: forced-hook memory (before_prompt_build recall +
  agent_end retain), replacing cognee's hook layer; cognify-sweep dropped
- verified live: Russian retain->recall, cross-session recall, bank isolation

Kimi quota (kb#62):
- adolf-llm/server.js: LLM-free GET /usage route (Kimi managed-usage API)
- quota-command-openclaw-plugin: /quota readout command

Cognee stack left running (decommission is H4/kb#76). Kimi-quota-footer
auto-append abandoned (streamed Matrix replies bypass outbound hooks).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_014t8Qg9gi7H7HtT8MncoXAB
This commit is contained in:
alvis
2026-07-15 19:53:21 +00:00
parent 544637c073
commit 4ac595a3a9
11 changed files with 1254 additions and 63 deletions
Download Patch File Download Diff File Expand all files Collapse all files

340
openai/adolf-llm/server.js
View File

@@ -65,23 +65,16 @@ function writeMcpConfig(dir) {
}
// ---------------------------------------------------------------------------
// Cognee auto-memory hooks (§3.2). Both are STUBS today — the cognee service is
// P4 and does not exist yet. They are deliberately non-blocking: a turn must
// never wait on (or fail because of) memory. Wire the real cognee-mcp calls in
// P4 and the rest of the turn pipeline stays unchanged.
// Pre-turn auto-retrieve: returns a string of relevant memories to inject ahead
// of the user prompt, or null for "nothing to inject".
// TODO(P4): call cognee memory_search / cognee.search() scoped by chatId.
async function cogneeSearch(_query, _chatId) {
return null;
}
// Post-turn auto-ingest: fire-and-forget; never awaited by the turn path.
// TODO(P4): async cognee.add(user + assistant) via cognee-mcp, scoped by chatId.
async function cogneeAdd(_userText, _assistantText, _chatId) {
return;
}
// Memory lives at the OpenClaw layer, not here (P8). The Adolf gateway loads
// the `cognee-memory` OpenClaw plugin, which owns all memory touchpoints:
// - before_prompt_build => LLM-free cognee graph recall, injected into the
// prompt this wrapper then receives from OpenClaw.
// - agent_end => raw `add` of the turn to cognee.
// - background sweep => async `cognify` on cognee-llm (Kimi).
// - cognee_recall tool + cognee-mcp `recall` for on-demand deep queries.
// This wrapper is therefore a dumb model endpoint again: it must never call
// cognee itself. The former cogneeSearch/cogneeAdd stubs (and their call sites)
// were deleted when the plugin took over (P8).
// ---------------------------------------------------------------------------
// Persistent conversation -> Kimi session map.
@@ -242,9 +235,8 @@ async function persistImage(url, dir, n) {
}
// Build the prompt for the current user turn: join text parts, persist any
// image parts, append path references. `injectedMemory` (from cogneeSearch) is
// prepended when present.
async function buildPrompt(userMsg, dir, injectedMemory) {
// image parts, append path references.
async function buildPrompt(userMsg, dir) {
const content = userMsg.content;
const textParts = [];
const imageRefs = [];
@@ -275,9 +267,6 @@ async function buildPrompt(userMsg, dir, injectedMemory) {
if (imageRefs.length) {
prompt += '\n\n' + imageRefs.map(r => `See attached image: ${r}`).join('\n');
}
if (injectedMemory) {
prompt = `Relevant memories (retrieved automatically):\n${injectedMemory}\n\n---\n\n${prompt}`;
}
return prompt;
}
@@ -288,8 +277,9 @@ async function buildPrompt(userMsg, dir, injectedMemory) {
// {"type":"session.resume_hint","session_id":"..."} -> capture session id
// onDelta(chunk) is called per assistant content fragment as it arrives.
// Resolves { text, sessionId } once the process closes.
function runKimi({ prompt, cwd, resumeId, onDelta }) {
function runKimi({ prompt, cwd, resumeId, onDelta, signal }) {
return new Promise((resolve, reject) => {
if (signal?.aborted) { reject(new Error('aborted before start')); return; }
const args = [];
if (resumeId) args.push('-r', resumeId);
args.push('-p', prompt, '--output-format', 'stream-json');
@@ -300,6 +290,19 @@ function runKimi({ prompt, cwd, resumeId, onDelta }) {
let stderr = '';
const parts = [];
let sessionId = null;
let settled = false;
let aborted = false;
// If the caller aborts (the gateway/client disconnected — e.g. its idle
// watchdog gave up), kill the child so it doesn't keep grinding an
// orphaned agent turn to completion, wasting Kimi quota and streaming into
// a dead socket. SIGTERM first, hard SIGKILL if it lingers.
const onAbort = () => {
aborted = true;
try { child.kill('SIGTERM'); } catch {}
setTimeout(() => { try { child.kill('SIGKILL'); } catch {} }, 3000).unref();
};
if (signal) signal.addEventListener('abort', onAbort, { once: true });
function handleLine(line) {
const t = line.trim();
@@ -324,11 +327,21 @@ function runKimi({ prompt, cwd, resumeId, onDelta }) {
});
child.stderr.on('data', d => { stderr += d; });
child.on('error', reject);
child.on('error', err => {
if (settled) return;
settled = true;
if (signal) signal.removeEventListener('abort', onAbort);
reject(err);
});
child.on('close', code => {
if (settled) return;
settled = true;
if (signal) signal.removeEventListener('abort', onAbort);
if (buf) handleLine(buf); // flush any trailing partial line
const text = parts.join('').trim();
if (!text && code !== 0) {
if (aborted) {
reject(new Error('aborted: client disconnected'));
} else if (!text && code !== 0) {
reject(new Error(`kimi exited ${code}: ${stderr.slice(0, 2000)}`));
} else {
resolve({ text, sessionId });
@@ -341,7 +354,7 @@ function runKimi({ prompt, cwd, resumeId, onDelta }) {
// One turn: resolve session (chat_id primary, history-hash fallback), persist
// media + .mcp.json, run kimi (streaming through onDelta), record the mapping,
// and fire the async cognee ingest. Returns { text }.
async function handleTurn(messages, onDelta) {
async function handleTurn(messages, onDelta, signal) {
const turns = convTurns(messages);
let lastUserIdx = -1;
for (let i = turns.length - 1; i >= 0; i--) {
@@ -396,24 +409,21 @@ async function handleTurn(messages, onDelta) {
fs.mkdirSync(dir, { recursive: true });
writeMcpConfig(dir); // Gate 1: shared MCP via project-root .mcp.json
// Pre-turn auto-retrieve (STUB no-op today; never blocks meaningfully).
const injectedMemory = await cogneeSearch(textOf(userMsg), chatId);
let prompt;
if (reseed) {
// Rebuild the whole conversation for a fresh session, plus current media/memory.
// Rebuild the whole conversation for a fresh session, plus current media.
const base = renderTranscript(turns.slice(0, lastUserIdx + 1));
const media = await buildPrompt(userMsg, dir, injectedMemory);
const media = await buildPrompt(userMsg, dir);
// buildPrompt already includes the current user text; for reseed we want the
// transcript to carry it, so only append image refs / injected memory extras.
// transcript to carry it, so only append image refs.
prompt = base;
const extra = media.replace(textOf(userMsg), '').trim();
if (extra) prompt += `\n\n${extra}`;
} else {
prompt = await buildPrompt(userMsg, dir, injectedMemory);
prompt = await buildPrompt(userMsg, dir);
}
const { text, sessionId } = await runKimi({ prompt, cwd: dir, resumeId, onDelta });
const { text, sessionId } = await runKimi({ prompt, cwd: dir, resumeId, onDelta, signal });
// Record the forward mapping.
const entry = { convId, sessionId: sessionId || resumeId, dir, ts: Date.now() };
@@ -425,12 +435,187 @@ async function handleTurn(messages, onDelta) {
}
persistMap();
// Post-turn auto-ingest (STUB no-op today; fire-and-forget, never awaited).
cogneeAdd(textOf(userMsg), text, chatId).catch(() => {});
return { text };
}
// ---------------------------------------------------------------------------
// Kimi quota readout (kb #62). GET /usage — the claude-usage analog for
// Adolf. LLM-free: hits Kimi's own managed-usage endpoint directly, never
// spawns `kimi`. Mirrors the parsing logic of the installed
// @moonshot-ai/kimi-code CLI itself (decompiled from dist/main.mjs's
// parseManagedUsagePayload/toUsageRow/limitLabel/resetHintFrom — same
// endpoint, same response shape) so bucket labels/derivations stay in sync
// with what `kimi` would show via its own /usage-equivalent.
//
// Token source: the CLI's own OAuth creds file, kept fresh by the running
// `kimi` process (adolf-llm-home volume). We only ever READ that file. If
// its access_token is stale/expired we refresh in memory (POST
// https://auth.kimi.com/api/oauth/token, form-encoded, grant_type=
// refresh_token — endpoint + client_id taken from the same decompiled
// KIMI_CODE_FLOW_CONFIG/refreshAccessToken) and cache the result in a
// module-level variable ONLY — we deliberately never write the refreshed
// token back to the creds file, since the live CLI owns that file and a
// racing write from here could corrupt/rotate state it depends on.
const KIMI_CREDS_PATH = '/root/.kimi-code/credentials/kimi-code.json';
const KIMI_OAUTH_HOST = 'https://auth.kimi.com';
const KIMI_CLIENT_ID = '17e5f671-d194-4dfb-9706-5516cb48c098';
const KIMI_USAGES_URL = 'https://api.kimi.com/coding/v1/usages';
let kimiMemToken = null; // { access_token, expires_at } — in-memory only, never persisted
async function loadKimiCreds() {
const raw = await fs.promises.readFile(KIMI_CREDS_PATH, 'utf8');
return JSON.parse(raw);
}
async function refreshKimiToken(refreshToken) {
const body = new URLSearchParams({
client_id: KIMI_CLIENT_ID,
grant_type: 'refresh_token',
refresh_token: refreshToken,
}).toString();
const res = await fetch(`${KIMI_OAUTH_HOST}/api/oauth/token`, {
method: 'POST',
headers: { 'Content-Type': 'application/x-www-form-urlencoded', Accept: 'application/json' },
body,
});
const data = await res.json().catch(() => ({}));
if (!res.ok || typeof data.access_token !== 'string') {
throw new Error(`kimi oauth refresh failed (HTTP ${res.status}): ${data.error || data.error_description || 'unknown error'}`);
}
return {
access_token: data.access_token,
expires_at: Math.floor(Date.now() / 1000) + Number(data.expires_in || 900),
};
}
// Resolve a usable access token, preferring the creds-file token (kept fresh
// by the live CLI) and falling back to an in-memory refresh only when that
// one is stale/expired.
async function getKimiAccessToken(forceRefresh) {
const creds = await loadKimiCreds();
const now = Math.floor(Date.now() / 1000);
if (!forceRefresh && creds.access_token && creds.expires_at && now < creds.expires_at - 30) {
return creds.access_token;
}
if (!forceRefresh && kimiMemToken && now < kimiMemToken.expires_at - 30) {
return kimiMemToken.access_token;
}
if (!creds.refresh_token) throw new Error('no refresh_token in kimi credentials file');
kimiMemToken = await refreshKimiToken(creds.refresh_token);
return kimiMemToken.access_token;
}
async function fetchKimiUsagesRaw() {
let token = await getKimiAccessToken(false);
let res = await fetch(KIMI_USAGES_URL, { headers: { Authorization: `Bearer ${token}`, Accept: 'application/json' } });
if (res.status === 401) {
token = await getKimiAccessToken(true); // force one in-memory refresh + retry
res = await fetch(KIMI_USAGES_URL, { headers: { Authorization: `Bearer ${token}`, Accept: 'application/json' } });
}
if (!res.ok) {
const text = await res.text().catch(() => '');
throw new Error(`kimi /usages HTTP ${res.status}: ${text.slice(0, 500)}`);
}
return res.json();
}
function isRecord(v) { return v !== null && typeof v === 'object' && !Array.isArray(v); }
function toInt(v) {
if (typeof v === 'number') return Number.isFinite(v) ? Math.trunc(v) : null;
if (typeof v === 'string') { const n = Number(v); return Number.isFinite(n) ? Math.trunc(n) : null; }
return null;
}
// Port of the CLI's limitLabel(): prefer an explicit name/title/scope field,
// else derive "<N>h limit" / "<N>m limit" / "<N>d limit" from the window's
// duration+timeUnit.
function kimiLimitLabel(item, detail, window, idx) {
for (const key of ['name', 'title', 'scope']) {
const v = item[key] ?? detail[key];
if (typeof v === 'string' && v) return v;
}
const duration = toInt(window.duration ?? item.duration ?? detail.duration);
const rawUnit = window.timeUnit ?? item.timeUnit ?? detail.timeUnit;
const timeUnit = typeof rawUnit === 'string' ? rawUnit : '';
if (duration !== null) {
if (timeUnit.includes('MINUTE')) {
if (duration >= 60 && duration % 60 === 0) return `${duration / 60}h limit`;
return `${duration}m limit`;
}
if (timeUnit.includes('HOUR')) return `${duration}h limit`;
if (timeUnit.includes('DAY')) return `${duration}d limit`;
return `${duration}s limit`;
}
return `Limit #${idx + 1}`;
}
function kimiResetIso(raw) {
for (const key of ['reset_at', 'resetAt', 'reset_time', 'resetTime']) {
const v = raw[key];
if (typeof v === 'string' && v) return v;
}
return null;
}
// Port of the CLI's toUsageRow(): used = raw.used, or limit-remaining when
// used is absent.
function kimiUsageRow(raw, defaultLabel) {
if (!isRecord(raw)) return null;
const limit = toInt(raw.limit);
let used = toInt(raw.used);
const remaining = toInt(raw.remaining);
if (used === null && remaining !== null && limit !== null) used = limit - remaining;
if (used === null && limit === null) return null;
const name = typeof raw.name === 'string' ? raw.name : (typeof raw.title === 'string' ? raw.title : defaultLabel);
return {
label: name,
used: used ?? 0,
limit: limit ?? 0,
remaining: remaining !== null ? remaining : (limit !== null && used !== null ? limit - used : null),
resets: kimiResetIso(raw),
};
}
function kimiRowOut(row) {
if (!row) return null;
const pct = row.limit > 0 ? Math.round((row.used / row.limit) * 100) : null;
return { pct, used: row.used, limit: row.limit, remaining: row.remaining, resets: row.resets };
}
// Normalize Kimi's /usages payload ({ usage, limits: [...] }) into the
// claude-usage-analog shape: weekly / window_5h / window_7d, each
// pct/used/limit/remaining/resets, plus a raw `limits` passthrough so no
// bucket is lost if label text ever drifts from what we match on below.
function normalizeKimiUsage(payload) {
const rec = isRecord(payload) ? payload : {};
const summaryRow = kimiUsageRow(rec.usage, 'Weekly limit');
const limitRows = [];
const rawLimits = Array.isArray(rec.limits) ? rec.limits : [];
rawLimits.forEach((item, idx) => {
if (!isRecord(item)) return;
const detail = isRecord(item.detail) ? item.detail : item;
const window = isRecord(item.window) ? item.window : {};
const label = kimiLimitLabel(item, detail, window, idx);
const row = kimiUsageRow(detail, label);
if (row) limitRows.push(row);
});
const findByLabel = re => limitRows.find(r => re.test(r.label));
const weekly = summaryRow || findByLabel(/week/i) || null;
const window5h = findByLabel(/^5\s*h(our)?\b|5h limit/i) || null;
const window7d = findByLabel(/^7\s*d(ay)?\b|7d limit/i) || null;
return {
timestamp: new Date().toISOString(),
weekly: kimiRowOut(weekly),
window_5h: kimiRowOut(window5h),
window_7d: kimiRowOut(window7d),
limits: limitRows.map(r => ({ label: r.label, ...kimiRowOut(r) })),
};
}
// ---------------------------------------------------------------------------
// OpenAI-compatible HTTP surface.
function completionBody(text) {
@@ -464,6 +649,21 @@ const server = http.createServer((req, res) => {
return;
}
if (req.method === 'GET' && req.url === '/usage') {
(async () => {
try {
const raw = await fetchKimiUsagesRaw();
const out = normalizeKimiUsage(raw);
res.writeHead(200, { 'Content-Type': 'application/json' });
res.end(JSON.stringify(out));
} catch (err) {
res.writeHead(502, { 'Content-Type': 'application/json' });
res.end(JSON.stringify({ error: String(err.message || err) }));
}
})();
return;
}
if (req.method === 'POST' && req.url === '/v1/chat/completions') {
let body = '';
req.on('data', d => { body += d; });
@@ -488,30 +688,68 @@ const server = http.createServer((req, res) => {
});
const id = `chatcmpl-${Date.now()}`;
const created = Math.floor(Date.now() / 1000);
res.write(sseChunk(id, created, { role: 'assistant' }, null));
// Heartbeat keepalive: OpenClaw's LLM idle watchdog aborts a turn on
// any >120s gap between SSE stream events (default timeoutSeconds),
// not on total run length. During long thinking/tool/MCP phases Kimi
// emits stream-json events we don't forward, so the SSE stream can go
// silent well past that window. Every write resets `lastWrite`; a 5s
// ticker emits an empty-content delta once 25s of silence elapses —
// still a stream event (resets the watchdog) but appends nothing
// visible to the rendered reply or cognee-persisted text.
let lastWrite = Date.now();
const write = (delta, finish) => {
if (res.writableEnded || res.destroyed) return;
res.write(sseChunk(id, created, delta, finish));
lastWrite = Date.now();
};
write({ role: 'assistant' }, null);
const hb = setInterval(() => {
if (Date.now() - lastWrite >= 25_000) write({ content: '' }, null);
}, 5_000);
// Propagate a client/gateway disconnect down to the kimi child so an
// abandoned turn (e.g. OpenClaw's idle watchdog gave up) is killed
// instead of finishing invisibly and burning quota. `done` guards
// against the normal res.end() 'close' also aborting.
const ac = new AbortController();
let done = false;
res.on('close', () => { if (!done) ac.abort(); });
try {
await handleTurn(messages, delta => {
res.write(sseChunk(id, created, { content: delta }, null));
});
res.write(sseChunk(id, created, {}, 'stop'));
await handleTurn(messages, delta => write({ content: delta }, null), ac.signal);
done = true;
write({}, 'stop');
res.write('data: [DONE]\n\n');
res.end();
} catch (err) {
// Headers already sent — surface the error inside the stream.
res.write(sseChunk(id, created, { content: `\n[error: ${String(err.message || err)}]` }, 'stop'));
res.write('data: [DONE]\n\n');
res.end();
done = true;
// Headers already sent — surface the error inside the stream (unless
// the socket is already gone, in which case there is nowhere to write).
if (!res.writableEnded && !res.destroyed) {
write({ content: `\n[error: ${String(err.message || err)}]` }, 'stop');
res.write('data: [DONE]\n\n');
}
} finally {
clearInterval(hb);
if (!res.writableEnded) res.end();
}
return;
}
const ac = new AbortController();
let done = false;
res.on('close', () => { if (!done) ac.abort(); });
try {
const { text } = await handleTurn(messages, null);
const { text } = await handleTurn(messages, null, ac.signal);
done = true;
res.writeHead(200, { 'Content-Type': 'application/json' });
res.end(JSON.stringify(completionBody(text)));
} catch (err) {
res.writeHead(500, { 'Content-Type': 'application/json' });
res.end(JSON.stringify({ error: String(err.message || err) }));
done = true;
if (!res.writableEnded && !res.destroyed) {
res.writeHead(500, { 'Content-Type': 'application/json' });
res.end(JSON.stringify({ error: String(err.message || err) }));
}
}
});
return;