Integrating with an agent

Four steps, and only the first and last touch the network. If your agent already has a place where the system prompt is assembled and a place where a finished exchange is recorded, you are most of the way there.

1. Build context for the message about to be answered. Returns text to prepend, plus the accounting.
2. Run the model with that text. Your call, unchanged.
3. Learn from the finished exchange.
4. Record the outcome for the domain, so the self-model has something to average.

Step 1 is keyed on the message being answered, not the one being sent. That is what makes the trigger work: the block is built from what the user just said, so a fact they named gets its full body in the prompt that responds to it.

Framework-agnostic, and the shape any framework wrapper should have. Note which lines cannot fail the turn.

import { createClient, runTurn } from "cogmemory"const memory = createClient({  apiKey: process.env.COGNITIVE_MEMORY_KEY!,  baseUrl: process.env.COGNITIVE_MEMORY_URL ?? "https://cogmem.astracollab.app"})export async function handleTurn(userMessage: string, sessionId: string) {  const { context, learning, learningSkipped } = await runTurn(    memory,    {      userMessage,      run: (block) => callYourModel({ system: block, userMessage })    },    { domain: "database", sessionId }  )  // context.entries says which memory produced each line and what it cost.  if (context.truncated) log.warn("context truncated", { tokens: context.totalTokens })  // A silent skip looks exactly like a working one until the thing you taught it  // never comes back. Check it.  if (learningSkipped) log.warn("memory did not learn", { learningSkipped })  return context}async function recordOutcome(domain: string, ok: boolean, note?: string) {  // success is false exactly when a failurePattern is present.  await memory.selfModel.record({    domain,    success: ok,    ...(ok ? { strategy: note } : { failurePattern: note })  })}

runTurn does all four in the right order and never lets a memory failure reach your model call. The two things it cannot do for you are the two it leaves to you: deciding what the turn was about, and deciding whether it went well.

Most agents already have the two hooks this needs. Wire memory to the point where the system prompt is assembled and the point where the turn is finished, and leave everything else alone.

import { streamText } from "ai"import { createClient } from "cogmemory"const memory = createClient({ apiKey: process.env.COGNITIVE_MEMORY_KEY! })const result = streamText({  model,  system: [    { role: "system", content: "You are a careful engineer." },    // The memory block goes last so it sits closest to the user's message.    { role: "system", content: context.text }  ],  messages})// After the stream finishes, learn from the exchange that actually happened.result.onFinish(async ({ text }) => {  await memory.turns    .learn({ userMessage: last(messages), assistantResponse: text, sessionId })    .catch((error) => log.warn("memory did not learn", { error }))})

// No dependency on the SDK at all. Same two calls.const before = await fetch(`${baseUrl}/api/v1/context`, {  method: "POST",  headers: {    authorization: `Bearer ${apiKey}`,    "content-type": "application/json"  },  body: JSON.stringify({ userMessage })})const { text } = await before.json()const answer = await callYourModel({ system: text, userMessage })await fetch(`${baseUrl}/api/v1/turns`, {  method: "POST",  headers: {    authorization: `Bearer ${apiKey}`,    "content-type": "application/json"  },  body: JSON.stringify({ userMessage, assistantResponse: answer, sessionId })})

Keep the memory block adjacent to the user message rather than at the top of the prompt. It is the part that is specific to this turn, and models weight the end of a long prompt more heavily than the beginning.

Sometimes the right move is to ask rather than to be handed everything.recallOrExplain exists for that: it returns a string ready to put in a prompt, and it handles the empty case by saying so out loud.

import { recallOrExplain } from "cogmemory"const brief = await recallOrExplain(memory, "where do we deploy", { limit: 5 })const answer = await callYourModel({ system: brief, userMessage })

With nothing matching, it returns Nothing in memory matches “where do we deploy”. If you were not told, say so rather than guessing. That sentence is the point. A silent miss is how a model invents a fact it was never given, so the helper makes the miss visible and phrased as an instruction.

Override it when your agent has a house style for that:

await recallOrExplain(memory, "where do we deploy", {  limit: 5,  emptyMessage: "Check src/infra, and say so if it is ambiguous."})

An agent that starts empty has to be taught before it is useful. If you already have the facts — a conventions file, a runbook, a table of hosts — load them once.

import { seedMemories } from "cogmemory"const { stored, merged, rejected } = await seedMemories(memory, {  facts: [    "Deploys go out on Tuesdays and Thursdays only",    "The staging database is internal-hbr-2291.pineapple.example",    "Never run migrations against production"  ],  sessionId: "onboarding"})

Re-running it is safe. Restatements come back under merged rather than as errors, and anything refused comes back under rejected with a reason — so a seed that quietly stored nothing tells you why instead of looking like success.

Seed through memories.create, not through turns. A seed string is a statement, not an exchange, and a message phrased as a question will be skipped as a lookup.

The self-model is the one feature that has to be fed deliberately, and the rule that catches people is that success is false whenever a failurePattern is present. There is no third state.

// A turn that went badly: name the pattern, and it counts as a failure.await memory.selfModel.record({  domain: "database",  success: false,  failurePattern: "migrated the schema without a dry run"})// A turn that went well: record what worked, so it can be suggested again.await memory.selfModel.record({  domain: "database",  success: true,  strategy: "dry run against a restored snapshot first"})

Patterns and strategies accumulate on the domain and are what gets injected as a guardrail once a domain drops below 75%. The strings are deduplicated, so recording the same pattern on every failure does not pad the prompt.

If you are using runTurn, pass failurePattern or strategy in the options and the outcome is recorded for you.

Memory is an enhancement. A read failure should degrade to no memory this turn, and a write failure should never reach a model call that already succeeded.

The client already retries transient failures and throws CognitiveMemoryError otherwise, with predicates so you branch on meaning rather than on status numbers:

import { CognitiveMemoryError } from "cogmemory"try {  await memory.turns.learn({ userMessage, assistantResponse })} catch (error) {  if (!(error instanceof CognitiveMemoryError)) throw error  if (error.isAuthError()) alert("the memory key stopped working")  else if (error.isScopeError()) alert(`widen the key: ${error.requiredScope}`)  else if (error.isServerError()) metric("memory.degraded")  else throw error}

Do not test an integration by pointing it at a running service. The package ships the deterministic layer — tiering, ranking, reconciliation, the fast gate — as pure functions, so the decisions can be asserted without a database, a network, or a model.

import { extractDeterministic, relevanceTokens, isLossyRewrite } from "cogmemory"import { describe, expect, it } from "vitest"it("learns a stated requirement in the user's own words", () => {  const found = extractDeterministic("Never force push to main")  expect(found).toHaveLength(1)  expect(found[0].content).toContain("force push")})it("refuses a merge that would drop a qualifier", () => {  expect(    isLossyRewrite(      "Always run migrations against staging",      "Always run migrations against staging, never production"    )  ).toBe(true)})it("ranks the memory that shares tokens with the query", () => {  expect(relevanceTokens("where do we deploy")).toContain("deploy")})

For the HTTP surface, construct the client with a stub HttpClient — it is exported for exactly this — and assert on the calls. The resources are thin, so a fake transport is a few lines.

The behaviours worth pinning in your own tests are the three that fail quietly: that context is built before the model runs, that a turn with a question does not learn, and that a learning failure does not fail the turn.

• The key is scoped to what the agent does. A read-only agent should hold memories:read and nothing else. Credentials.
• You are logging learningSkipped and truncated. Both are silent by default and both are the reason memory looks broken.
• You set a domain. Without it the self-model stays at its priors forever. The self-model.
• You seeded the obvious facts so the agent is useful on turn one instead of turn fifty.
• You checked the context preview against a real message from your logs — the context preview is the fastest way to find out what the agent is actually being told.

And the whole thing is two calls, so a full integration is a page of code — see the quickstart for the smallest version that works, and the API reference for the exact contract.