diff --git a/packages/opencode/src/altimate/training/store.ts b/packages/opencode/src/altimate/training/store.ts index 75256ce928..8597465d85 100644 --- a/packages/opencode/src/altimate/training/store.ts +++ b/packages/opencode/src/altimate/training/store.ts @@ -1,6 +1,7 @@ // altimate_change - Training store wrapping MemoryStore for learned knowledge import { MemoryStore, type MemoryBlock } from "../../memory" import { + TRAINING_META_COMMENT, TRAINING_TAG, TRAINING_MAX_PATTERNS_PER_KIND, TrainingKind, @@ -164,5 +165,5 @@ export namespace TrainingStore { } function stripTrainingMeta(content: string): string { - return content.replace(/^\n*/, "").trim() + return content.replace(TRAINING_META_COMMENT, "").trim() } diff --git a/packages/opencode/src/altimate/training/types.ts b/packages/opencode/src/altimate/training/types.ts index a5e90f1985..6f4e38c44e 100644 --- a/packages/opencode/src/altimate/training/types.ts +++ b/packages/opencode/src/altimate/training/types.ts @@ -2,6 +2,12 @@ import z from "zod" export const TRAINING_TAG = "training" + +/** Matches the metadata comment ``embedTrainingMeta`` writes at the head of a + * training block's content. Exported so readers that normalise it away — the + * workspace mirror hashes content without it, because the applied counter is + * rewritten every session — cannot drift from the writer. */ +export const TRAINING_META_COMMENT = /^\n*/ export const TRAINING_ID_PREFIX = "training" // altimate_change start — increase training limits for enterprise teams // 20 entries per kind is too restrictive for teams with 200+ dbt models spanning @@ -70,6 +76,6 @@ export function embedTrainingMeta(content: string, meta: TrainingBlockMeta): str "-->", ].join("\n") // Strip existing training meta block if present - const stripped = content.replace(/^\n*/, "") + const stripped = content.replace(TRAINING_META_COMMENT, "") return header + "\n" + stripped } diff --git a/packages/opencode/src/altimate/workspace/api-client.ts b/packages/opencode/src/altimate/workspace/api-client.ts index 10a429e385..99c2681e58 100644 --- a/packages/opencode/src/altimate/workspace/api-client.ts +++ b/packages/opencode/src/altimate/workspace/api-client.ts @@ -17,6 +17,10 @@ const REQUEST_TIMEOUT_MS = 15_000 export interface DatamateRef { id: number name: string + /** Whether the workspace has memory switched on. Surfaced as a user-facing + * toggle in the workspace app, so callers that write memory must respect it. + * Undefined when the backend omitted the field. */ + memoryEnabled?: boolean } export interface Binding { @@ -237,6 +241,13 @@ async function req( return json as T } +/** Shared wire helper for sibling Altimate routers. Exported so callers that + * need the same credential resolution, abort budget and typed-error mapping do + * not duplicate any of it — see ./memory-api.ts, which drives + * ``/datamates/memory/*`` through this exact path. Always pass an explicit + * ``base``; the default is this module's own namespace. */ +export { req as altimateRequest } + export namespace WorkspaceApi { /** Server-authoritative pre-check by git remote. Returns null on 404. */ export async function getBindingForRemote(remote: string): Promise { @@ -350,7 +361,7 @@ export namespace WorkspaceApi { // bare ``[...]``, and a generic ``{data: [...]}`` — so a backend // contract change (or compat layer) doesn't silently empty the picker. // (cubic-dev-ai round 3.) - type Row = { id: number | string; name: string } + type Row = { id: number | string; name: string; memory_enabled?: boolean } const body = await req("GET", "/", { base: "/datamates", }) @@ -378,7 +389,7 @@ export namespace WorkspaceApi { // per-element rather than per-envelope malformed value. return rows .filter((d): d is Row => d !== null && typeof d === "object") - .map((d) => ({ id: Number(d.id), name: d.name })) + .map((d) => ({ id: Number(d.id), name: d.name, memoryEnabled: d.memory_enabled })) .filter((d) => Number.isInteger(d.id) && d.id > 0 && typeof d.name === "string") } } diff --git a/packages/opencode/src/altimate/workspace/memory-api.ts b/packages/opencode/src/altimate/workspace/memory-api.ts new file mode 100644 index 0000000000..1729f613cc --- /dev/null +++ b/packages/opencode/src/altimate/workspace/memory-api.ts @@ -0,0 +1,177 @@ +// altimate_change - new file +// +// Wire client for the memory routes in altimate-backend (/datamates/memory/*). +// Those routes proxy to a Lambda which runs an LLM extractor before storing, so +// the behaviour here is shaped by two facts established against the live +// service rather than inferred: +// +// * A create runs the extractor. It rewrites the submitted text and +// overwrites the ``memory_type`` and ``title`` metadata keys, and it can +// decline the content entirely and store nothing while still answering 200. +// ``update`` does NOT run the extractor — it writes text and metadata +// verbatim. Every create is therefore followed by an update that restores +// the block exactly as the user wrote it. +// * ``extracted`` is not a stored/not-stored signal. It is false both when +// the extractor declined (no ``result``) and when the extractor errored and +// the raw messages were stored as a fallback (``result`` present, content +// stored). Presence of ``result`` is the only reliable signal, which is why +// ``add`` reports an id rather than a boolean. +// +// Records are tagged with ``metadata.source`` so the backend can keep them out +// of ordinary Datamate reads; this client opts back in explicitly on every read. +import { altimateRequest } from "./api-client" + +/** Stamped on every record this CLI writes, and the value the backend filters + * on. Reads must opt in by name or they come back empty. */ +export const MIRROR_SOURCE = "altimate-code" + +const BASE = "/datamates/memory" + +/** Upper bound on records requested per read. The service does not honour + * paging parameters — a repeated request re-runs the identical query — so this + * is a single bounded fetch rather than the first page of several. */ +export const LIST_LIMIT = 200 + +/** A record as returned by ``/list``. */ +export interface CloudMemoryRecord { + id: string + memory: string + created_at?: string + updated_at?: string + metadata?: Record | null +} + +/** Metadata written on every mirrored record. Names and types are fixed so + * that lookup and filtering agree across independent call sites. */ +export interface MirrorMetadata { + source: typeof MIRROR_SOURCE + block_id: string + block_scope: "global" | "project" + /** Always "private" in v0. Written so a later sharing feature can promote a + * record without a backfill; nothing reads it today. */ + visibility: "private" + block_created: string + block_updated: string + /** Absent for global blocks — that is what makes them span workspaces. */ + datamate_id?: string + datamate_name?: string + /** Project provenance and the cross-machine convergence key. */ + repo_remote?: string + project_path?: string + block_tags?: string + /** ISO-8601. Mirrored so a TTL'd block expires everywhere rather than living + * forever in the workspace once it has left the machine that wrote it. */ + block_expires?: string + archived?: "true" + archived_at?: string +} + +export function isMirrorRecord(record: CloudMemoryRecord): boolean { + const meta = record.metadata + if (!meta || typeof meta !== "object") return false + return meta.source === MIRROR_SOURCE +} + +export function isArchived(record: CloudMemoryRecord): boolean { + return record.metadata?.archived === "true" +} + +/** Pull every created record id out of a create response. + * + * A create runs an extractor server-side, and an extractor is free to split one + * submission into several records — each carrying the metadata we sent, so each + * looks like our block. Returning only the first would leave the rest holding + * rewritten text, never repaired, never indexed and never archived, while the + * read path injected all of them under one block id. + * + * The payload shape is not a published contract, so this accepts the observed + * forms rather than guessing: a bare array, or an object wrapping one under + * ``results``/``memories``/``data``. An empty result is the expected outcome + * when the extractor declines the content. */ +export function extractRecordIds(result: unknown): string[] { + const rows = (() => { + if (Array.isArray(result)) return result + if (result && typeof result === "object") { + const obj = result as Record + for (const key of ["results", "memories", "data"]) { + if (Array.isArray(obj[key])) return obj[key] as unknown[] + } + } + return [] + })() + + const ids: string[] = [] + for (const row of rows) { + if (!row || typeof row !== "object") continue + const id = (row as Record).id ?? (row as Record).memory_id + if (typeof id === "string" && id) ids.push(id) + } + return ids +} + +/** Convenience for callers that only need to know whether anything was stored. */ +export function extractRecordId(result: unknown): string | undefined { + return extractRecordIds(result)[0] +} + +export namespace MemoryApi { + /** Create a record and report the ids it produced. + * + * Returns an empty array when the service stored nothing. That is not an error — + * the extractor declines content it judges unremarkable — so the caller + * should leave the block unindexed and let a later edit retry, rather than + * treating it as a failure. */ + export async function add(content: string, metadata: MirrorMetadata): Promise { + const res = await altimateRequest<{ message?: string; result?: unknown }>("POST", "/", { + base: BASE, + allowEmptyBody: true, + body: { + messages: [{ role: "user", content }], + memory_options: { metadata }, + }, + }) + return extractRecordIds(res?.result) + } + + /** Overwrite a record verbatim. Does not run the extractor, and replaces the + * metadata dict wholesale, so callers must pass the complete metadata. */ + export async function update( + memoryId: string, + content: string, + metadata: MirrorMetadata, + ): Promise { + await altimateRequest<{ message?: string }>("PATCH", `/${encodeURIComponent(memoryId)}`, { + base: BASE, + allowEmptyBody: true, + body: { memory: content, metadata }, + }) + } + + /** Read this user's mirrored records. + * + * ``include_sources`` is required: the backend excludes this client's records + * from list/search by default so they do not surface in Datamate sessions. + * No workspace filter is sent — the service's own query for a caller's + * records is not scoped by workspace, so narrowing happens in the caller. + * + * Deliberately NOT capped at ``LIST_LIMIT``. The service ignores paging, so a + * cap here would discard real records before anything could rank them, and + * the user would lose memory with no signal. Session context is already + * bounded downstream: ``MemoryPrompt.inject`` scores every block and appends + * only while it fits the caller's budget. ``LIST_LIMIT`` is used to recognise + * a possibly-cut-short read (see ``fetchKnownRecords``), not to trim one. */ + export async function list(): Promise { + const rows = await altimateRequest( + "GET", + "/list", + { + base: BASE, + allowEmptyBody: true, + query: { include_sources: MIRROR_SOURCE, page_size: String(LIST_LIMIT) }, + }, + ) + if (!rows) return [] + if (Array.isArray(rows)) return rows + return Array.isArray(rows.memories) ? rows.memories : [] + } +} diff --git a/packages/opencode/src/altimate/workspace/memory-backfill.ts b/packages/opencode/src/altimate/workspace/memory-backfill.ts new file mode 100644 index 0000000000..cafe214bca --- /dev/null +++ b/packages/opencode/src/altimate/workspace/memory-backfill.ts @@ -0,0 +1,46 @@ +// altimate_change - new file +// +// Seeds a freshly bound workspace with the memory this machine already holds. +// Without it only blocks written AFTER the bind would ever reach the store, and +// a user's existing memory would stay invisible in the workspace. +// +// Lives in its own module rather than inside ./state.ts because the sweep needs +// MemoryStore, whose write path already reaches ./memory-sync — importing it +// directly from state.ts would close an eval-order cycle +// (state -> backfill -> memory -> store -> memory-sync -> state). state.ts +// reaches this through a lazy dynamic import instead. +import { MemoryStore } from "@/memory/store" +import { Log } from "@/altimate/util/log" +import { backfill, isEnabled } from "./memory-sync" +import type { CachedBinding } from "./state" + +const log = Log.create({ service: "altimate-workspace-memory-backfill" }) + +/** Push every non-expired local block. Throttled and resumable inside + * ``backfill`` — blocks already synced at their current payload are skipped, so + * repeated binds cost index reads rather than uploads. + * + * Covers both scopes: project blocks attach to the workspace just bound, and + * global blocks go up account-level. A bind is the only moment global memory is + * swept; blocks written later ride the ordinary per-write mirror. */ +export async function backfillOnBind(directory: string, binding: CachedBinding): Promise { + if (!isEnabled()) return false + try { + // The directory and binding are passed in rather than rediscovered. The + // `link` subcommand binds from a plain yargs handler with no instance + // context, so resolving project scope from the ambient instance throws + // there — silently, because this catch turns it into a log line while the + // CLI still prints "Linked". Reading project memory was the entire point. + const blocks = await MemoryStore.listAll({ directory }) + if (blocks.length === 0) return true + const result = await backfill(blocks, binding) + log.info("workspace memory seeded after bind", result) + // Only a sweep that stored everything it meant to counts as seeded. A + // failure here must leave the binding eligible for a retry, or local blocks + // stay absent from the workspace until a rebind or an unrelated edit. + return !result.gated && result.failed === 0 + } catch (err) { + log.warn("workspace memory backfill after bind failed", { err: String(err) }) + return false + } +} diff --git a/packages/opencode/src/altimate/workspace/memory-index.ts b/packages/opencode/src/altimate/workspace/memory-index.ts new file mode 100644 index 0000000000..e82b37b689 --- /dev/null +++ b/packages/opencode/src/altimate/workspace/memory-index.ts @@ -0,0 +1,188 @@ +// altimate_change - new file +// +// Maps a memory block's logical identity to the id of the cloud record holding +// it, so a later save updates that record instead of creating a second one. +// +// Scoped to (tenant, apiUrl, credential) at the top level. The credential is +// included because records are per-user: two people sharing a machine and a +// tenant would otherwise share this map and update each other's records. Only a +// short digest of the API key is stored — never the key. +// +// Same file conventions as ./state.ts — Global.Path.state, 0o600, atomic write, +// discard on corrupt — so the two behave identically under the same failures. +import { createHash } from "node:crypto" +import { chmodSync, existsSync, readFileSync } from "node:fs" +import path from "node:path" +import { AltimateApi } from "@/altimate/api/client" +import { Global } from "@/global" +import { Filesystem } from "@/util/filesystem" +import { Log } from "@/altimate/util/log" + +const INDEX_VERSION = 1 + +const log = Log.create({ service: "altimate-workspace-memory-index" }) + +export interface IndexEntry { + /** The cloud record's id. */ + memoryId: string + /** Digest of the block's mirrored payload, used to skip a push when nothing + * a reader would see has changed. Deliberately not part of the logical key — + * a key that moves when content changes could never find the record it means + * to update. */ + contentHash: string + syncedAt: number +} + +interface IndexFile { + version: 1 + tenant: string + apiUrl: string + /** Digest of the API key. Identifies the account without storing a secret. */ + account: string + records: Record +} + +export function indexPath(): string { + return path.join(Global.Path.state, "altimate-workspace-memory-index.json") +} + +/** The logical identity of a mirrored block. + * + * Project blocks include the workspace AND the originating project: block ids + * are only unique within a project directory, so two projects bound to one + * workspace can each hold a ``warehouse/snowflake``. Global blocks carry + * neither — they belong to the account and are the same block everywhere. */ +export function indexKey(input: { + scope: "global" | "project" + blockId: string + datamateId?: number + projectKey?: string +}): string { + if (input.scope === "global") return `global::${input.blockId}` + return `project::${input.datamateId ?? "unbound"}::${input.projectKey ?? "unknown"}::${input.blockId}` +} + +function digest(value: string): string { + return createHash("sha256").update(value).digest("hex").slice(0, 16) +} + +function isValidIndexFile(raw: unknown): raw is IndexFile { + if (!raw || typeof raw !== "object") return false + const r = raw as Record + if (r.version !== INDEX_VERSION) return false + for (const key of ["tenant", "apiUrl", "account"]) { + if (typeof r[key] !== "string" || !r[key]) return false + } + if (!r.records || typeof r.records !== "object" || Array.isArray(r.records)) return false + for (const v of Object.values(r.records as Record)) { + if (!v || typeof v !== "object") return false + const e = v as Record + if (typeof e.memoryId !== "string" || !e.memoryId) return false + if (typeof e.contentHash !== "string") return false + if (typeof e.syncedAt !== "number") return false + } + return true +} + +function readFile(): IndexFile | null { + const p = indexPath() + if (!existsSync(p)) return null + try { + const raw = JSON.parse(readFileSync(p, "utf8")) as unknown + if (!isValidIndexFile(raw)) return null + return raw + } catch (err) { + log.warn("workspace memory index is corrupt, discarding", { + code: (err as NodeJS.ErrnoException)?.code, + }) + return null + } +} + +function writeFile(next: IndexFile): void { + const p = indexPath() + Filesystem.writeJsonAtomic(p, next) + try { + chmodSync(p, 0o600) + } catch (err) { + log.warn("could not chmod workspace memory index", { + code: (err as NodeJS.ErrnoException)?.code, + }) + } +} + +interface Scope { + tenant: string + apiUrl: string + account: string +} + +async function currentScope(): Promise { + // Defensive for the same reason as state.ts::tenantKey — a corrupt + // credentials file or schema drift must read as "no credentials", never as a + // rejection into a fire-and-forget mirror call. + try { + if (!(await AltimateApi.isConfigured())) return null + const c = await AltimateApi.getCredentials() + return { + tenant: c.altimateInstanceName, + apiUrl: c.altimateUrl, + account: digest(c.altimateApiKey), + } + } catch (err) { + log.warn("could not resolve credentials for memory index scoping", { err: String(err) }) + return null + } +} + +function matches(file: IndexFile, scope: Scope): boolean { + return file.tenant === scope.tenant && file.apiUrl === scope.apiUrl && file.account === scope.account +} + +// Serializes read-modify-write within this process. Two mirror tasks finishing +// at once would otherwise both read the pre-write file and the second would +// drop the first's entry. +// +// Cross-process races remain — the same known gap as ./state.ts. Two CLI +// processes writing the same block concurrently can create two records, and +// that duplicate is NOT self-repairing: a later save resolves one of them and +// updates it, leaving the other live indefinitely. Closing it needs a file lock +// here, or an idempotency key at the service. +let writeChain: Promise = Promise.resolve() + +export async function readIndex(): Promise> { + const scope = await currentScope() + if (!scope) return {} + const file = readFile() + if (!file || !matches(file, scope)) return {} + return file.records +} + +export async function readIndexEntry(key: string): Promise { + return (await readIndex())[key] ?? null +} + +/** Upsert one entry. Best-effort: losing the map costs a duplicate record on + * the next save, never a failed local memory write. */ +export async function recordIndexEntry(key: string, entry: IndexEntry): Promise { + const scope = await currentScope() + if (!scope) return + const task = writeChain.then(async () => { + try { + const existing = readFile() + const file: IndexFile = + existing && matches(existing, scope) + ? existing + : { version: INDEX_VERSION, ...scope, records: {} } + file.records[key] = entry + writeFile(file) + } catch (err) { + log.warn("could not persist workspace memory index", { + code: (err as NodeJS.ErrnoException)?.code, + err: String(err), + }) + } + }) + writeChain = task.catch(() => {}) + return task +} diff --git a/packages/opencode/src/altimate/workspace/memory-sync.ts b/packages/opencode/src/altimate/workspace/memory-sync.ts new file mode 100644 index 0000000000..ad0385830b --- /dev/null +++ b/packages/opencode/src/altimate/workspace/memory-sync.ts @@ -0,0 +1,806 @@ +// altimate_change - new file +// +// Mirrors Altimate Code memory blocks to the workspace memory store, and loads +// a workspace's memory back into a session. +// +// Additive by design: the local Markdown files remain authoritative and every +// operation here is fire-and-forget, so a cloud failure can neither fail nor +// slow a local memory save. +// +// Write — a saved block is pushed to the store, tagged with the workspace the +// project is bound to. A create is followed by an update that restores the text +// verbatim, because a create runs an extractor that rewrites it. Blocks the +// extractor declines are left unindexed so a later edit retries them. +// +// Read — one fetch per session, held in memory and merged at injection time. +// Nothing is written to disk: local files are the source of truth, and a cloud +// record can have been edited elsewhere by a client that does not preserve this +// CLI's metadata. +import { createHash } from "node:crypto" +import { Flag as CoreFlag } from "@opencode-ai/core/flag/flag" +import { Flag } from "@/flag/flag" +import { Instance } from "@/project/instance" +import { Log } from "@/altimate/util/log" +import type { MemoryBlock } from "@/memory/types" +import { TRAINING_META_COMMENT } from "@/altimate/training/types" +import { readLocalBinding, type CachedBinding } from "./state" +import { indexKey, readIndex, readIndexEntry, recordIndexEntry } from "./memory-index" +import { WorkspaceApi } from "./api-client" +import { + LIST_LIMIT, + MemoryApi, + MIRROR_SOURCE, + isArchived, + isMirrorRecord, + type CloudMemoryRecord, + type MirrorMetadata, +} from "./memory-api" + +const log = Log.create({ service: "altimate-workspace-memory-sync" }) + +/** How long an injection waits on an in-flight hydration before proceeding + * without it. Sessions are often a single turn, so injecting one turn late + * would read as the feature not working; blocking on the full request budget + * would be worse. */ +const HYDRATION_WAIT_MS = 3_000 + +/** Parallelism for the bind-time backfill. Low on purpose: a backfill can be + * every block the machine holds, and each create costs an LLM pass server-side. */ +const BACKFILL_CONCURRENCY = 2 + +/** A cloud block merged into a session's injection. */ +export interface RemoteMemoryBlock extends MemoryBlock { + /** Marks the block as cloud-sourced. A remote training block must never + * drive TrainingStore.incrementApplied, which writes to the LOCAL store and + * would fabricate a file for a block this machine never had. */ + remote: true + /** Human-readable origin, when the block came from a different project in + * the same workspace. Absent for this project's own blocks and for globals. */ + origin?: string +} + +/** Per-session hydration state. + * + * Keyed by session id rather than held in module scope: the server runs + * concurrent sessions, potentially in different projects bound to different + * workspaces, and a single shared overlay would let one session read another's + * workspace memory. Binding resolution needs no such keying — ``Instance`` is + * AsyncLocalStorage-backed, so ``Instance.directory`` is already correct for + * whichever session's async context is running. */ +interface SessionMemory { + overlay: RemoteMemoryBlock[] + hydration: Promise | null + touchedAt: number + /** Set once a bounded wait expired, so later injections do not re-wait. */ + waitTimedOut?: boolean +} + +const sessions = new Map() + +/** Sessions are evicted oldest-first rather than on a session-end hook, which + * does not exist here. Well above any plausible concurrent-session count, and + * an eviction only costs a refetch. */ +const MAX_TRACKED_SESSIONS = 32 + +function sessionState(sessionID: string): SessionMemory { + let state = sessions.get(sessionID) + if (!state) { + if (sessions.size >= MAX_TRACKED_SESSIONS) { + const oldest = [...sessions.entries()].sort((a, b) => a[1].touchedAt - b[1].touchedAt)[0] + if (oldest) sessions.delete(oldest[0]) + } + state = { overlay: [], hydration: null, touchedAt: Date.now() } + sessions.set(sessionID, state) + } + state.touchedAt = Date.now() + return state +} + +/** The mirror rides the workspace pilot flag and honours the memory opt-out. + * Never active for anyone who has not opted into the pilot. */ +export function isEnabled(): boolean { + return CoreFlag.ALTIMATE_WORKSPACE && !Flag.ALTIMATE_DISABLE_MEMORY +} + +/** Test seam. Production leaves this unset and resolves the binding from the + * active instance; tests set it so they need not boot one. */ +export const syncInternals: { + resolveBinding?: () => Promise + /** Test seam for the local-existence check. Production reads the store. */ + blockExists?: (block: MemoryBlock) => Promise +} = {} + +/** Instance.directory throws synchronously with no instance context, so a + * trailing .catch() on a promise built from it never fires. */ +function currentDirectory(): string | null { + try { + return Instance.directory + } catch { + return null + } +} + +export function projectKeyFor(binding: CachedBinding): string { + return binding.repoRemote ?? binding.projectPath ?? "unknown" +} + +async function currentBinding(): Promise { + if (syncInternals.resolveBinding) return syncInternals.resolveBinding() + const directory = currentDirectory() + if (!directory) return null + try { + return await readLocalBinding(directory) + } catch (err) { + log.warn("could not resolve binding for memory mirror", { err: String(err) }) + return null + } +} + +/** How long an ENABLED workspace is trusted before re-checking. Without this + * every memory write would cost a workspace lookup, including writes that turn + * out to be no-ops. + * + * Only positives are cached. A workspace starts with memory disabled, so + * caching that verdict would leave the CLI ignoring the setting for a full TTL + * after the user switches it on — and a disabled workspace writes nothing, so + * re-checking it costs a lookup on an operation that was going to be a no-op + * anyway. */ +const MEMORY_ENABLED_TTL_MS = 60_000 + +const memoryEnabledCache = new Map() + +/** Warn once per workspace, not once per write. */ +const missingFieldWarned = new Set() + +/** Whether the bound workspace has memory switched on. + * + * The workspace app exposes this as a user-facing toggle, so mirroring into a + * workspace with memory disabled would contradict what the user is shown. + * Fails closed: if the check cannot be made, nothing is mirrored. */ +async function memoryEnabled(binding: CachedBinding): Promise { + const cached = memoryEnabledCache.get(binding.datamateId) + if (cached && Date.now() - cached.checkedAt < MEMORY_ENABLED_TTL_MS) return true + try { + const workspaces = await WorkspaceApi.listDatamates() + const match = workspaces.find((w) => w.id === binding.datamateId) + if (match && match.memoryEnabled === undefined && !missingFieldWarned.has(binding.datamateId)) { + // Fail-closed is right, but a backend that has not shipped the field + // turns the whole feature into a silent no-op. Say so once. + missingFieldWarned.add(binding.datamateId) + log.warn("workspace has no memory_enabled field; treating memory as disabled", { + workspace: binding.datamateId, + }) + } + const value = match?.memoryEnabled === true + if (value) memoryEnabledCache.set(binding.datamateId, { checkedAt: Date.now() }) + else memoryEnabledCache.delete(binding.datamateId) + return value + } catch (err) { + log.warn("could not confirm workspace memory setting, skipping mirror", { err: String(err) }) + // Not cached either way: a transient failure should neither disable the + // mirror for a minute nor keep it enabled. Failing closed already prevents + // this particular write. + return false + } +} + +/** Strips the training metadata comment. + * + * TrainingStore.incrementApplied rewrites a training block on EVERY session + * start to bump a counter embedded in that comment. Mirroring those rewrites + * would fire a request per training block per session for a change no reader + * can see, so the counter is normalised away before hashing. */ +function stripTrainingMeta(content: string): string { + return content.replace(TRAINING_META_COMMENT, "").trim() +} + +/** Fingerprint of everything about a block that reaches the store. Tags and + * expiry are included because both are mirrored. */ +function contentHash(block: MemoryBlock): string { + const payload = JSON.stringify([ + stripTrainingMeta(block.content), + [...block.tags].sort(), + block.expires ?? "", + ]) + // sha-256, not a 32-bit non-cryptographic hash. This value is the single + // gate deciding whether a save is sent at all: on a collision ``push`` + // returns "unchanged" and a real edit is silently never mirrored, with no + // retry. A 32-bit space makes that reachable; node:crypto is already a + // dependency of this feature (see ./memory-index.ts). + return createHash("sha256").update(payload).digest("hex") +} + +/** Read tags written by {@link buildMetadata}. + * + * Accepts the legacy comma-joined form so records written before the JSON + * encoding still load; those cannot represent a tag containing a comma, which + * is the defect the JSON form fixes. */ +export function decodeTags(raw: unknown): string[] { + if (typeof raw !== "string" || !raw) return [] + if (raw.startsWith("[")) { + try { + const parsed = JSON.parse(raw) + if (Array.isArray(parsed)) return parsed.filter((t): t is string => typeof t === "string" && t.length > 0) + } catch { + // Fall through to the legacy form. + } + } + return raw + .split(",") + .map((t) => t.trim()) + .filter(Boolean) +} + +export function buildMetadata(block: MemoryBlock, binding: CachedBinding | null): MirrorMetadata { + const meta: MirrorMetadata = { + source: MIRROR_SOURCE, + block_id: block.id, + block_scope: block.scope, + visibility: "private", + block_created: block.created, + block_updated: block.updated, + } + // JSON, not a comma join: a tag containing a comma split into two on read. + if (block.tags.length > 0) meta.block_tags = JSON.stringify(block.tags) + if (block.expires) meta.block_expires = block.expires + // Global blocks deliberately carry no workspace: they belong to the account + // and apply in every workspace the user has. + if (block.scope === "project" && binding) { + meta.datamate_id = String(binding.datamateId) + meta.datamate_name = binding.datamateName + if (binding.repoRemote) meta.repo_remote = binding.repoRemote + if (binding.projectPath) meta.project_path = binding.projectPath + } + return meta +} + +/** Does a cloud record hold this block? + * + * Matches on logical identity only. Content is deliberately excluded: an + * identity that moved when content changed could never find the record it + * means to update. */ +function isSameBlock(record: CloudMemoryRecord, block: MemoryBlock, binding: CachedBinding | null): boolean { + if (!isMirrorRecord(record) || isArchived(record)) return false + const m = record.metadata ?? {} + if (m.block_id !== block.id) return false + if (m.block_scope !== block.scope) return false + if (block.scope !== "project") return true + if (String(m.datamate_id ?? "") !== String(binding?.datamateId ?? "")) return false + const recordProject = (m.repo_remote as string | undefined) ?? (m.project_path as string | undefined) + return !recordProject || !binding || recordProject === projectKeyFor(binding) +} + +/** Fetch the record set once, and report whether it was cut short. + * + * The service ignores paging parameters, so a full result at the limit means + * records exist that no request can reach. Callers use ``truncated`` to avoid + * acting on a partial view — creating a duplicate, or concluding a record is + * absent when it is merely out of reach. */ +async function fetchKnownRecords(): Promise { + const records = await MemoryApi.list() + // Exactly at the limit is the only ambiguous case. Treating ">= limit" as + // truncated would permanently block creates for any user whose record set is + // larger than the limit -- the service ignores paging, so a response LARGER + // than the limit is positive proof the set came back whole. + const truncated = records.length === LIST_LIMIT + if (truncated) { + log.warn("workspace memory read hit the service limit; some records are unreachable", { + limit: LIST_LIMIT, + }) + } + return { records, truncated } +} + +/** Records already known to the caller, so a sweep does not re-list per block. */ +type KnownRecords = { records: CloudMemoryRecord[]; truncated: boolean } + +/** What a push actually did. ``declined`` means the service kept nothing — + * counting it as success made a sweep report blocks it had not stored. */ +type PushOutcome = "stored" | "unchanged" | "declined" | "skipped" + +/** Is this block still in the local store? + * + * Imported lazily: ``@/memory/store`` reaches this module on its write path, so + * a static import would close an eval-order cycle (see ./memory-backfill.ts). + * A read failure answers "yes" — refusing to mirror on a transient read error + * would silently drop a memory the user does have. */ +async function existsLocally(block: MemoryBlock): Promise { + if (syncInternals.blockExists) return syncInternals.blockExists(block) + try { + const { MemoryStore } = await import("@/memory/store") + return !!(await MemoryStore.read(block.scope, block.id)) + } catch (err) { + log.warn("could not confirm a block still exists locally; mirroring anyway", { + id: block.id, + err: String(err), + }) + return true + } +} + +async function push( + block: MemoryBlock, + binding: CachedBinding | null, + known?: KnownRecords, +): Promise { + const key = indexKey({ + scope: block.scope, + blockId: block.id, + datamateId: binding?.datamateId, + projectKey: binding ? projectKeyFor(binding) : undefined, + }) + const hash = contentHash(block) + const existing = await readIndexEntry(key) + if (existing?.contentHash === hash) return "unchanged" + + const metadata = buildMetadata(block, binding) + + // Resolve the record set even when the index already names a record. Every + // safety check below needs it, and previously only ``backfill`` passed one: + // on the ordinary per-save path the truncation guard, the lookup-failure + // path and the newer-remote guard were all unreachable. Fetched only after + // the hash check above, so an unchanged block still costs nothing. + let view = known + if (!view) { + try { + view = await fetchKnownRecords() + } catch (err) { + // A failed lookup must not read as "no record exists" -- that is how a + // duplicate gets created. Leave the block unindexed so a later save + // retries it. + log.warn("could not read the workspace record set; deferring", { id: block.id, err: String(err) }) + return "skipped" + } + } + + // The local store is the authority on whether this block still exists. + // `backfill` registers a block on the serialize queue only when a worker + // dequeues it, so a delete issued mid-sweep runs first and this push would + // otherwise undo it -- recreating a record the user deleted, or reviving an + // archived one, and then marking it synced so no later sweep re-archives it. + if (!(await existsLocally(block))) { + log.warn("skipping mirror for a block that no longer exists locally", { + id: block.id, + scope: block.scope, + }) + return "skipped" + } + + // An archived record is a tombstone. Ignoring the index here sends us to the + // identity search below, which excludes archived records -- so a block the + // user recreated under an old id gets a fresh record instead of un-archiving + // the old one (`MemoryApi.update` replaces metadata wholesale, which would + // drop the archived marker). + const indexed = existing?.memoryId + ? view.records.find((r) => r.id === existing.memoryId) + : undefined + const indexUsable = existing?.memoryId && (!indexed || !isArchived(indexed)) + const match = (indexUsable ? existing?.memoryId : undefined) ?? view.records.find((r) => isSameBlock(r, block, binding))?.id + if (match) { + // Refuse to move a record backwards. Two machines editing the same block, + // or a stale clone running a sweep, would otherwise overwrite a newer cloud + // value with older local content — last-request-wins rather than + // convergence. This narrows the window rather than closing it; closing it + // needs a conditional update the service does not offer. + const remote = view.records.find((r) => r.id === match) + const remoteUpdated = remote && (remote.metadata ?? {}).block_updated + if (typeof remoteUpdated === "string" && remoteUpdated > block.updated) { + log.warn("declining to overwrite a newer workspace record with older local content", { + id: block.id, + localUpdated: block.updated, + remoteUpdated, + }) + return "skipped" + } + await MemoryApi.update(match, block.content, metadata) + await recordIndexEntry(key, { memoryId: match, contentHash: hash, syncedAt: Date.now() }) + return "stored" + } + + // Creating against a list we know was cut short risks a duplicate: the + // record may exist just beyond the window. Refusing leaves the block + // unindexed, so a later save retries once the set is readable. + if (view.truncated) { + log.warn("skipping create against a truncated record set", { id: block.id, scope: block.scope }) + return "skipped" + } + + const created = await MemoryApi.add(block.content, metadata) + if (created.length === 0) { + // The extractor declined the content and stored nothing. Leaving the block + // unindexed means a later edit retries it rather than silently skipping. + log.warn("workspace declined to store memory block", { id: block.id, scope: block.scope }) + return "declined" + } + + // Repair the create: the extractor rewrote the text and replaced our + // memory_type/title. update() is verbatim and replaces the whole metadata + // dict, restoring the block exactly as written. + const [primary, ...extras] = created + await MemoryApi.update(primary, block.content, metadata) + await recordIndexEntry(key, { memoryId: primary, contentHash: hash, syncedAt: Date.now() }) + + // An extractor may split one submission into several records, each carrying + // the metadata we sent. Only one can represent the block; the rest would + // otherwise be injected as duplicates under the same block id, holding text + // the user never wrote. + for (const extra of extras) { + log.warn("archiving an extra record produced by one create", { id: block.id, memoryId: extra }) + await MemoryApi.update(extra, "", { ...metadata, archived: "true", archived_at: new Date().toISOString() }) + } + return "stored" +} + +/** Cloud operations for one logical block, run one at a time. + * + * Both callers are fire-and-forget from the store's write and delete paths, so + * without this a delete issued while a mirror is still in flight finds no + * record to archive, and the mirror then creates a live one -- resurrecting + * memory the user deleted. Two rapid saves race the same way and create + * duplicates. Keyed by scope+id, so unrelated blocks still mirror in parallel. + */ +const blockQueues = new Map>() + +function serialize(scope: "global" | "project", blockId: string, op: () => Promise): Promise { + const key = `${scope}:${blockId}` + const prior = blockQueues.get(key) ?? Promise.resolve() + const next = prior.then(op, op) + // Retain only while this op is the tail, so the map cannot grow unbounded. + blockQueues.set(key, next) + const settled = next.catch(() => undefined) + void settled.then(() => { + if (blockQueues.get(key) === next) blockQueues.delete(key) + }) + return next +} + +/** Mirror one block. Safe to call unconditionally — returns immediately when + * the pilot flag is off, the project is unbound, or the workspace has memory + * disabled. */ +export async function mirrorBlock(block: MemoryBlock): Promise { + if (!isEnabled()) return + // Queued BEFORE the binding lookup, not after. Both are async, so resolving + // them first let two operations on one block reach `serialize` in the + // opposite order to the writes that triggered them. + await serialize(block.scope, block.id, async () => { + // A binding is required for EVERY scope, not just project. Memories are + // associated with a workspace, and the workspace is what carries the + // memory_enabled setting — mirroring from an unbound directory would upload + // with nothing to consult and nothing to attribute it to. Global blocks still + // carry no workspace themselves, so they apply everywhere on read; the + // binding governs only whether we upload at all. + const binding = await currentBinding() + if (!binding) return + if (!(await memoryEnabled(binding))) return + await push(block, binding) + }) +} + +/** Archive a block's cloud record rather than deleting it, so the workspace + * keeps the history. Only this client filters the marker — other readers do + * not — so an archived record stays visible elsewhere. */ +export async function archiveBlock(scope: "global" | "project", blockId: string): Promise { + if (!isEnabled()) return + // Queued behind any in-flight mirror for the same block, so a delete cannot + // run before the create it is meant to undo. The binding lookup happens + // inside the queued op for the same reason as in `mirrorBlock`. + return serialize(scope, blockId, async () => { + const binding = await currentBinding() + if (!binding) return + if (!(await memoryEnabled(binding))) return + await archiveNow(scope, blockId, binding) + }) +} + +async function archiveNow( + scope: "global" | "project", + blockId: string, + binding: CachedBinding, +): Promise { + const key = indexKey({ + scope, + blockId, + datamateId: binding?.datamateId, + projectKey: binding ? projectKeyFor(binding) : undefined, + }) + const entry = await readIndexEntry(key) + + // The read exists to recover the record's current text so the archive can + // preserve it, and to find the record at all when the index cannot answer. + // It is also why archiving fails when the result is truncated. + const records = await MemoryApi.list() + + // The index is per-machine and is discarded on account switch, corruption or + // a wiped state directory. Without a fallback, deleting a block on a machine + // that has lost its index leaves the record live, and every later session + // re-injects it with no way to remove it. Match on logical identity instead, + // exactly as the write path does. + const current = entry + ? records.find((r) => r.id === entry.memoryId) + : // Matching on workspace alone is too coarse: two projects in one + // workspace may hold the same block id, and deleting one would archive + // the other's record. Use the write path's own identity test so the + // fallback can only ever hit the record this block would have written. + records.find((r) => + isSameBlock(r, { id: blockId, scope, tags: [], content: "", created: "", updated: "" } as MemoryBlock, binding), + ) + if (!current) return + + const now = new Date().toISOString() + const metadata: MirrorMetadata = { + ...((current.metadata ?? {}) as unknown as MirrorMetadata), + source: MIRROR_SOURCE, + block_id: blockId, + block_scope: scope, + visibility: "private", + archived: "true", + archived_at: now, + } + // Keep the text — archiving hides a record from injection, it does not erase + // what it said. + await MemoryApi.update(current.id, current.memory ?? "", metadata) + // Written only after the remote archive lands, so a failure leaves the entry + // pointing at a record that is still live rather than losing track of it. + await recordIndexEntry(key, { memoryId: current.id, contentHash: "", syncedAt: Date.now() }) +} + +async function runQueue( + items: T[], + worker: (item: T) => Promise, + concurrency: number, +): Promise<{ ok: number; failed: number; declined: number; skipped: number }> { + let cursor = 0 + let ok = 0 + let failed = 0 + let declined = 0 + let skipped = 0 + const runners = Array.from({ length: Math.min(concurrency, items.length) }, async () => { + while (cursor < items.length) { + const item = items[cursor++] + try { + const outcome = await worker(item) + if (outcome === "declined") declined++ + else if (outcome === "skipped" || outcome === "unchanged") skipped++ + else ok++ + } catch (err) { + failed++ + log.warn("memory mirror task failed", { err: String(err) }) + } + } + }) + await Promise.all(runners) + return { ok, failed, declined, skipped } +} + +/** Push a set of blocks — the sweep that runs when a project is bound to a + * workspace. Throttled and resumable: blocks whose payload is already synced + * are skipped, so a re-run after a partial failure sends only what is missing. */ +export async function backfill( + blocks: MemoryBlock[], + explicitBinding?: CachedBinding, +): Promise<{ ok: number; failed: number; skipped: number; declined: number; gated: boolean }> { + // ``gated`` says the sweep never ran, as opposed to running and storing + // nothing. A caller recording "this binding is seeded" must be able to tell + // those apart: memory being off is not a completed seed. + if (!isEnabled()) return { ok: 0, failed: 0, skipped: 0, declined: 0, gated: true } + // The bind path passes the binding it just recorded; there is no ambient + // instance to resolve one from on the `link` subcommand. + const binding = explicitBinding ?? (await currentBinding()) + if (!binding || !(await memoryEnabled(binding))) + return { ok: 0, failed: 0, skipped: blocks.length, declined: 0, gated: true } + const index = await readIndex() + + const pending: { block: MemoryBlock; binding: CachedBinding | null }[] = [] + let skipped = 0 + for (const block of blocks) { + const target = block.scope === "project" ? binding : null + if (block.scope === "project" && !target) { + skipped++ + continue + } + const key = indexKey({ + scope: block.scope, + blockId: block.id, + datamateId: target?.datamateId, + projectKey: target ? projectKeyFor(target) : undefined, + }) + if (index[key]?.contentHash === contentHash(block)) { + skipped++ + continue + } + pending.push({ block, binding: target }) + } + + if (pending.length === 0) return { ok: 0, failed: 0, skipped, declined: 0, gated: false } + + // One read for the whole sweep. Every block in a first bind is an index miss, + // so resolving each through its own lookup made a bind cost one full record + // fetch per block. + let known: KnownRecords | undefined + try { + known = await fetchKnownRecords() + } catch (err) { + log.warn("could not prefetch records for backfill; falling back per block", { + err: String(err), + }) + } + + log.info("workspace memory backfill starting", { pending: pending.length, skipped }) + const result = await runQueue( + pending, + (item) => serialize(item.block.scope, item.block.id, () => push(item.block, item.binding, known)), + BACKFILL_CONCURRENCY, + ) + // `skipped` combines blocks filtered before the queue (already synced, or + // project blocks with no workspace) with those the queue itself declined to + // act on — a truncated read, or a remote copy that is newer. + const totals = { ...result, skipped: result.skipped + skipped, gated: false } + log.info("workspace memory backfill finished", totals) + return totals +} + +/** Short label for a record's originating project. */ +function originLabel(meta: Record): string | undefined { + const remote = typeof meta.repo_remote === "string" ? meta.repo_remote : undefined + if (remote) { + const trimmed = remote.replace(/[/]+$/, "").replace(/\.git$/, "").replace(/[/]+$/, "") + const last = trimmed.split(/[/:]/).pop() + if (last) return last + } + const projectPath = typeof meta.project_path === "string" ? meta.project_path : undefined + if (projectPath) { + const base = projectPath.replace(/\/$/, "").split("/").pop() + if (base) return base + } + return undefined +} + +/** Map a cloud record back to an injectable block, or null if it is not a + * well-formed mirrored block. */ +export function toBlock( + record: CloudMemoryRecord, + ownProjectKey: string | undefined, +): RemoteMemoryBlock | null { + const meta = record.metadata + if (!meta || typeof meta !== "object") return null + const blockId = typeof meta.block_id === "string" ? meta.block_id : undefined + const scope = meta.block_scope === "global" || meta.block_scope === "project" ? meta.block_scope : undefined + if (!blockId || !scope || !record.memory) return null + + const tags = decodeTags(meta.block_tags) + const updated = + (typeof meta.block_updated === "string" ? meta.block_updated : undefined) ?? + record.updated_at ?? + new Date().toISOString() + + const recordProjectKey = + (typeof meta.repo_remote === "string" ? meta.repo_remote : undefined) ?? + (typeof meta.project_path === "string" ? meta.project_path : undefined) + const fromSibling = scope === "project" && !!recordProjectKey && recordProjectKey !== ownProjectKey + + return { + id: blockId, + scope, + tags, + created: (typeof meta.block_created === "string" ? meta.block_created : undefined) ?? record.created_at ?? updated, + updated, + expires: typeof meta.block_expires === "string" ? meta.block_expires : undefined, + content: record.memory, + remote: true, + ...(fromSibling ? { origin: originLabel(meta) } : {}), + } +} + +/** Does this record apply to the workspace the current project is bound to? + * + * Project blocks apply to their workspace regardless of which project wrote + * them — a session anywhere in the workspace sees them. Global blocks carry no + * workspace and apply everywhere; omitting that arm would make them + * write-only. */ +export function belongsHere(record: CloudMemoryRecord, ownWorkspace: string | undefined): boolean { + const meta = record.metadata ?? {} + if (meta.block_scope === "global") return true + if (!ownWorkspace) return false + return String(meta.datamate_id ?? "") === ownWorkspace +} + +/** Fetch a session's workspace memory once. + * + * Idempotent: safe to call on every turn, which matters because the caller's + * enclosing block runs per user turn rather than once per session. Repeat calls + * return the in-flight or completed hydration instead of refetching, and the + * existing overlay is never cleared first — clearing before a refetch made + * workspace memory blink out of the prompt whenever a fetch ran long. */ +export async function hydrate(sessionID: string): Promise { + if (!isEnabled()) return + const state = sessionState(sessionID) + if (state.hydration) return state.hydration + state.hydration = doHydrate(sessionID) + return state.hydration +} + +/** Wait for a session's in-flight hydration, capped. Resolves immediately once + * the fetch has settled, so only the first injection of a session pays. */ +export async function whenHydrated( + sessionID: string, + timeoutMs: number = HYDRATION_WAIT_MS, +): Promise { + const state = sessions.get(sessionID) + const pending = state?.hydration + if (!pending || !state) return + // A hydration that already blew the budget must not be waited on again: the + // promise stays unresolved, so every later injection in the session would pay + // the full timeout. Wait once, then let the overlay fill in whenever it lands. + if (state.waitTimedOut) return + let timer: ReturnType | undefined + let timedOut = false + try { + await Promise.race([ + pending, + new Promise((resolve) => { + timer = setTimeout(() => { + timedOut = true + resolve() + }, timeoutMs) + timer.unref?.() + }), + ]) + } finally { + if (timer) clearTimeout(timer) + if (timedOut) state.waitTimedOut = true + } +} + +async function doHydrate(sessionID: string): Promise { + try { + const binding = await currentBinding() + if (!binding || !(await memoryEnabled(binding))) { + sessionState(sessionID).overlay = [] + return + } + const ownProjectKey = binding ? projectKeyFor(binding) : undefined + const ownWorkspace = binding ? String(binding.datamateId) : undefined + + const records = await MemoryApi.list() + + const blocks: RemoteMemoryBlock[] = [] + for (const record of records) { + if (!record?.id) continue + if (!isMirrorRecord(record) || isArchived(record)) continue + if (!belongsHere(record, ownWorkspace)) continue + const block = toBlock(record, ownProjectKey) + if (!block) continue + // A TTL'd block must expire everywhere, not just on the machine that + // wrote it. The cloud copy is not swept, so honour it on read. + if (block.expires && new Date(block.expires) <= new Date()) continue + blocks.push(block) + } + + sessionState(sessionID).overlay = blocks + if (blocks.length > 0) { + log.info("workspace memory hydrated", { blocks: blocks.length, workspace: binding?.datamateName }) + } + } catch (err) { + log.warn("workspace memory hydration failed", { err: String(err) }) + sessionState(sessionID).overlay = [] + } +} + +/** A session's cloud overlay. Returns a copy so a caller cannot mutate the + * cached state in place. */ +export function overlayBlocks(sessionID: string): RemoteMemoryBlock[] { + return [...(sessions.get(sessionID)?.overlay ?? [])] +} + +/** Forget a session's hydration, or all of them. + * + * Not called per turn: doing so defeated ``hydrate``'s idempotence and made + * every turn refetch. Exposed for tests and for a future session-end hook. */ +export function resetOverlay(sessionID?: string): void { + if (sessionID === undefined) { + sessions.clear() + memoryEnabledCache.clear() + return + } + sessions.delete(sessionID) +} diff --git a/packages/opencode/src/altimate/workspace/state.ts b/packages/opencode/src/altimate/workspace/state.ts index 075f861c79..6ec5e3f027 100644 --- a/packages/opencode/src/altimate/workspace/state.ts +++ b/packages/opencode/src/altimate/workspace/state.ts @@ -30,6 +30,10 @@ export interface CachedBinding { repoRemote: string | null projectPath: string | null linkedAt: number + /** Set once a bind-time seed completed without failures. Absent means the + * seed has not run, errored, or was skipped because memory was off — all of + * which must stay retryable, so a later warm sweeps again. */ + seededAt?: number } interface CacheFile { @@ -72,6 +76,8 @@ function isValidCacheFile(raw: unknown): raw is CacheFile { (typeof b.projectPath === "string" && b.projectPath.length > 0) if (!hasIdentity) return false if (typeof b.linkedAt !== "number") return false + // A corrupt marker must not read as "already seeded" and suppress the sweep. + if (b.seededAt !== undefined && typeof b.seededAt !== "number") return false } return true } @@ -182,15 +188,54 @@ export async function readLocalBinding(directory: string): Promise { const key = await tenantKey() if (!key) return @@ -200,18 +245,61 @@ export async function recordApprovedBinding( // duplicate retries against a workspace that IS bound server-side. // (cubic round 3.) canonicalizeKey resolves symlinks so writes and reads // funnel through the same key (macOS ``/tmp`` → ``/private/tmp``). + // Whether this call actually changes the binding. A flow that merely warms + // the cache with the binding already on disk must not trigger a sweep: the + // seed exists for a NEW or CHANGED bind, and re-running it on every warm + // costs a full read of local memory and a round trip per block -- now paid + // synchronously on the `link` path, which awaits the seed. + let bindingChanged = true + let alreadySeeded = false try { const existing = readCache() const cache: CacheFile = existing && existing.tenant === key.tenant && existing.apiUrl === key.apiUrl ? existing : { version: CACHE_VERSION, tenant: key.tenant, apiUrl: key.apiUrl, bindings: {} } - cache.bindings[canonicalizeKey(directory)] = binding + const prior = cache.bindings[canonicalizeKey(directory)] + bindingChanged = !prior || !sameBinding(prior, binding) + alreadySeeded = !bindingChanged && !!prior?.seededAt + // Carry the seed marker across a warm so a completed seed is not repeated. + cache.bindings[canonicalizeKey(directory)] = + !bindingChanged && prior?.seededAt ? { ...binding, seededAt: prior.seededAt } : binding writeCache(cache) } catch (err) { + // An unreadable cache means the prior binding is unknown, so fall back to + // seeding: a missed seed is worse than a redundant one. + bindingChanged = true log.warn("could not persist workspace binding cache", { code: (err as NodeJS.ErrnoException)?.code, err: String(err), }) } + + // altimate_change start - seed the workspace with the memory this machine + // already holds. Deliberately OUTSIDE the try above: a failed cache write + // must not skip the backfill, and a failed backfill must not read as a failed + // link. The dynamic import keeps the module graph acyclic — see the header of + // ./memory-backfill.ts for why a static import cannot be used. + // + // ``awaitBackfill`` exists because the CLI calls ``process.exit()`` as soon + // as a command handler returns (src/index.ts): a detached sweep is killed + // mid-flight there, so a bind that reported success could seed nothing. The + // TUI stays resident and leaves it detached so the dialog closes at once. + // Skip only when this exact binding has already been seeded successfully. A + // warm after a failed or skipped seed must try again, or the blocks this + // machine already holds never reach the workspace. + if (alreadySeeded) return + const seeded = import("./memory-backfill") + .then((m) => m.backfillOnBind(canonicalizeKey(directory), binding)) + .then((ok) => { + if (ok) markSeeded(directory, binding) + return ok + }) + .catch((err) => { + log.warn("could not start workspace memory backfill", { err: String(err) }) + return false + }) + if (opts?.awaitBackfill) await seeded + else void seeded + // altimate_change end } diff --git a/packages/opencode/src/cli/cmd/link.ts b/packages/opencode/src/cli/cmd/link.ts index 0da0543ded..d420d7d4a9 100644 --- a/packages/opencode/src/cli/cmd/link.ts +++ b/packages/opencode/src/cli/cmd/link.ts @@ -254,7 +254,7 @@ async function runBrowserHandoff( repoRemote: res.binding.repo_remote, projectPath: res.binding.project_path, linkedAt: Date.now(), - }) + }, { awaitBackfill: true }) bindSpin.stop(`Linked to "${res.binding.datamate_name}".`) const manageUrl = await manageUrlFor(res.binding.datamate_id) if (manageUrl) prompts.log.info(`Manage it at: ${manageUrl}`) @@ -380,7 +380,7 @@ async function createThenBindOrRebind( repoRemote: created.binding.repo_remote, projectPath: created.binding.project_path, linkedAt: Date.now(), - }) + }, { awaitBackfill: true }) prompts.log.info(`Manage it at: ${created.manage_url}`) // Guard against a server that hands back a non-http(s) manage_url — ``open`` // delegates to the OS handler, so a rogue value could launch an unrelated @@ -494,7 +494,7 @@ async function bindOrRebind( repoRemote: res.binding.repo_remote, projectPath: res.binding.project_path, linkedAt: Date.now(), - }) + }, { awaitBackfill: true }) spin.stop( isRebind ? `Re-linked to "${res.binding.datamate_name}".` diff --git a/packages/opencode/src/memory/prompt.ts b/packages/opencode/src/memory/prompt.ts index 1826e131b5..9d7ba753b2 100644 --- a/packages/opencode/src/memory/prompt.ts +++ b/packages/opencode/src/memory/prompt.ts @@ -14,8 +14,11 @@ import { trainingKind, parseTrainingMeta, type TrainingKind, + TRAINING_META_COMMENT, } from "@/altimate/training/types" import { TrainingStore } from "@/altimate/training/store" +// altimate_change - workspace memory overlay (read side of the cloud mirror) +import { overlayBlocks, whenHydrated, type RemoteMemoryBlock } from "@/altimate/workspace/memory-sync" // Training kind display headers (moved from training/prompt.ts) const KIND_HEADERS: Record = { @@ -50,17 +53,47 @@ const KIND_ORDER: TrainingKind[] = ["rule", "pattern", "standard", "glossary", " // Track which training entries have been applied this session (prevents double-counting) const appliedThisSession = new Set() +// altimate_change start - merge the workspace overlay into the local block set. +// +// Local files win: for a block this project also holds on disk, the local copy +// is authoritative, because a cloud record can have been edited elsewhere by a +// client that does not preserve this CLI's metadata. +// +// A block from a SIBLING project in the same workspace is kept even when its id +// collides with a local one — block ids are only unique within a project +// directory, so a collision there means two genuinely different blocks. +function isRemote(block: MemoryBlock): block is RemoteMemoryBlock { + return (block as RemoteMemoryBlock).remote === true +} + +function mergeOverlay(local: MemoryBlock[], remote: RemoteMemoryBlock[]): MemoryBlock[] { + if (remote.length === 0) return local + const localKeys = new Set(local.map((b) => `${b.scope}:${b.id}`)) + const merged: MemoryBlock[] = [...local] + for (const block of remote) { + const fromSibling = block.origin !== undefined + if (!fromSibling && localKeys.has(`${block.scope}:${block.id}`)) continue + merged.push(block) + } + merged.sort((a, b) => b.updated.localeCompare(a.updated)) + return merged +} +// altimate_change end + export namespace MemoryPrompt { /** Reset per-session applied tracking. Call at session start (step === 1). */ export function resetSession(): void { appliedThisSession.clear() } - /** Format a non-training memory block for display. */ - export function formatBlock(block: MemoryBlock): string { + /** Format a non-training memory block for display. A block carried in from a + * SIBLING project in the same workspace is labelled with its origin — + * without it the model would read another project's memory as this one's. */ + export function formatBlock(block: MemoryBlock & { origin?: string }): string { const tagsStr = block.tags.length > 0 ? ` [${block.tags.join(", ")}]` : "" const expiresStr = block.expires ? ` (expires: ${block.expires})` : "" - let result = `### ${block.id} (${block.scope})${tagsStr}${expiresStr}\n${block.content}` + const originStr = originSuffix(block.origin) + let result = `### ${block.id} (${block.scope})${tagsStr}${expiresStr}${originStr}\n${block.content}` if (block.citations && block.citations.length > 0) { const citationLines = block.citations.map((c) => { @@ -75,13 +108,24 @@ export namespace MemoryPrompt { } /** Format a training entry for display (with applied count). */ - function formatTrainingEntry(block: MemoryBlock): string { + /** Label for a block that came from a sibling project in the workspace. + * Shared so ``formatBlock`` and ``formatTrainingEntry`` cannot drift — they + * rendered the same literal twice. */ + function originSuffix(origin?: string): string { + return origin ? ` — from workspace project \`${origin}\`` : "" + } + + function formatTrainingEntry(block: MemoryBlock & { origin?: string }): string { const meta = parseTrainingMeta(block.content) const appliedStr = meta && meta.applied > 0 ? ` (applied ${meta.applied}x)` : "" // Strip the training metadata comment from content for display - const content = block.content.replace(/^\n*/, "").trim() + const content = block.content.replace(TRAINING_META_COMMENT, "").trim() const name = block.id.split("/").slice(2).join("/") || block.id - return `#### ${name}${appliedStr}\n${content}` + // altimate_change - label a sibling project's entry, exactly as formatBlock + // does. mergeOverlay deliberately keeps both blocks when a sibling shares an + // id with this project's, so without this the model sees two identical + // headings it cannot tell apart — the mis-reading the label exists to stop. + return `#### ${name}${appliedStr}${originSuffix(block.origin)}\n${content}` } /** Score a block for relevance to the current agent context. */ @@ -133,7 +177,13 @@ export namespace MemoryPrompt { budget: number = MEMORY_DEFAULT_INJECTION_BUDGET, ctx?: InjectionContext, ): Promise { - const blocks = await MemoryStore.listAll() + // altimate_change - fold in this session's workspace memory overlay. The + // bounded wait only ever costs the first injection of a session, and the + // overlay is per-session so concurrent sessions in different workspaces + // cannot read each other's memory. + if (ctx?.sessionID) await whenHydrated(ctx.sessionID) + const remote = ctx?.sessionID ? overlayBlocks(ctx.sessionID) : [] + const blocks = mergeOverlay(await MemoryStore.listAll(), remote) if (blocks.length === 0) return "" // Score and filter @@ -219,6 +269,9 @@ export namespace MemoryPrompt { // Fire-and-forget: increment applied count for training blocks (once per session) for (const block of injectedTraining) { + // altimate_change - a remote block has no local file; incrementApplied + // would fabricate one from the cloud copy. Skip them. + if (isRemote(block)) continue if (!appliedThisSession.has(block.id)) { appliedThisSession.add(block.id) const kind = trainingKind(block) diff --git a/packages/opencode/src/memory/store.ts b/packages/opencode/src/memory/store.ts index aa22ddf118..da4b0bf64b 100644 --- a/packages/opencode/src/memory/store.ts +++ b/packages/opencode/src/memory/store.ts @@ -6,6 +6,13 @@ import { Global } from "@opencode-ai/core/global" import { Instance } from "@/project/instance" import { MEMORY_MAX_BLOCK_SIZE, MEMORY_MAX_BLOCKS_PER_SCOPE, MemoryBlockSchema, type MemoryBlock, type Citation } from "./types" import { Telemetry } from "@/altimate/telemetry" +// altimate_change start - workspace memory mirror. Additive: local files stay +// authoritative and every call below is fire-and-forget. +import { archiveBlock, mirrorBlock } from "@/altimate/workspace/memory-sync" +import { Log } from "@/altimate/util/log" + +const mirrorLog = Log.create({ service: "altimate-memory-mirror" }) +// altimate_change end const FRONTMATTER_REGEX = /^---\n([\s\S]*?)\n---\n([\s\S]*)$/ @@ -16,8 +23,8 @@ function globalDir(): string { // altimate_change start - use .altimate-code (primary) with .opencode (fallback) // Cache keyed by Instance.directory to avoid stale paths when context changes const _projectDirCache = new Map() -function projectDir(): string { - const dir = Instance.directory +function projectDir(directory?: string): string { + const dir = directory ?? Instance.directory const cached = _projectDirCache.get(dir) if (cached) return cached const primary = path.join(dir, ".altimate-code", "memory") @@ -35,12 +42,12 @@ function projectDir(): string { } // altimate_change end -function dirForScope(scope: "global" | "project"): string { - return scope === "global" ? globalDir() : projectDir() +function dirForScope(scope: "global" | "project", directory?: string): string { + return scope === "global" ? globalDir() : projectDir(directory) } -function blockPath(scope: "global" | "project", id: string): string { - const base = dirForScope(scope) +function blockPath(scope: "global" | "project", id: string, directory?: string): string { + const base = dirForScope(scope, directory) const result = path.join(base, ...id.split("/").slice(0, -1), `${id.split("/").pop()}.md`) // Defense-in-depth: verify the resolved path stays within the memory directory const resolved = path.resolve(result) @@ -119,8 +126,12 @@ function auditEntry(action: string, id: string, scope: string, extra?: string): } export namespace MemoryStore { - export async function read(scope: "global" | "project", id: string): Promise { - const filepath = blockPath(scope, id) + export async function read( + scope: "global" | "project", + id: string, + directory?: string, + ): Promise { + const filepath = blockPath(scope, id, directory) let raw: string try { raw = await fs.readFile(filepath, "utf-8") @@ -156,8 +167,15 @@ export namespace MemoryStore { return validated.data } - export async function list(scope: "global" | "project", opts?: { includeExpired?: boolean }): Promise { - const dir = dirForScope(scope) + /** ``opts.directory`` resolves project scope explicitly instead of from the + * ambient instance. Callers outside an instance context — the ``link`` + * subcommand is one — have no ambient directory, and reading project scope + * without it throws. */ + export async function list( + scope: "global" | "project", + opts?: { includeExpired?: boolean; directory?: string }, + ): Promise { + const dir = dirForScope(scope, opts?.directory) const blocks: MemoryBlock[] = [] async function scanDir(currentDir: string, prefix: string) { @@ -176,7 +194,7 @@ export namespace MemoryStore { } else if (entry.name.endsWith(".md")) { const baseName = entry.name.slice(0, -3) const id = prefix ? `${prefix}/${baseName}` : baseName - const block = await read(scope, id) + const block = await read(scope, id, opts?.directory) if (block) { if (!opts?.includeExpired && isExpired(block)) continue blocks.push(block) @@ -190,8 +208,25 @@ export namespace MemoryStore { return blocks } - export async function listAll(opts?: { includeExpired?: boolean }): Promise { - const [global, project] = await Promise.all([list("global", opts), list("project", opts)]) + /** One scope failing must not take the other down with it: a caller with no + * usable project directory should still get global blocks. */ + export async function listAll(opts?: { includeExpired?: boolean; directory?: string }): Promise { + const [globalResult, projectResult] = await Promise.allSettled([ + list("global", opts), + list("project", opts), + ]) + if (globalResult.status === "rejected") { + Log.create({ service: "memory.store" }).warn("could not read global memory", { + err: String(globalResult.reason), + }) + } + if (projectResult.status === "rejected") { + Log.create({ service: "memory.store" }).warn("could not read project memory", { + err: String(projectResult.reason), + }) + } + const global = globalResult.status === "fulfilled" ? globalResult.value : [] + const project = projectResult.status === "fulfilled" ? projectResult.value : [] const all = [...project, ...global] all.sort((a, b) => b.updated.localeCompare(a.updated)) return all @@ -264,6 +299,19 @@ export namespace MemoryStore { tags_count: block.tags.length, }) + // altimate_change start - mirror to the bound workspace. The local file is + // already durable here, so a cloud failure must not surface as a failed + // memory write. No-ops unless the pilot flag is on, the project is bound, + // and the workspace has memory enabled. + void mirrorBlock(block).catch((e) => { + mirrorLog.warn("failed to mirror memory block to workspace", { + id: block.id, + scope: block.scope, + err: String(e), + }) + }) + // altimate_change end + // Auto-clean expired blocks AFTER successful write to avoid data loss if (needsCleanup) { const expiredBlocks = allBlocks.filter((b) => isExpired(b)) @@ -291,6 +339,16 @@ export namespace MemoryStore { duplicate_count: 0, tags_count: 0, }) + // altimate_change start - archive rather than delete the cloud record so + // the workspace keeps the history. Fire-and-forget, as with write. + void archiveBlock(scope, id).catch((e) => { + mirrorLog.warn("failed to archive workspace memory record", { + id, + scope, + err: String(e), + }) + }) + // altimate_change end return true } catch (e: any) { if (e.code === "ENOENT") return false diff --git a/packages/opencode/src/memory/types.ts b/packages/opencode/src/memory/types.ts index 3d4df2206c..8f3e34880c 100644 --- a/packages/opencode/src/memory/types.ts +++ b/packages/opencode/src/memory/types.ts @@ -48,5 +48,8 @@ export const AGENT_TRAINING_RELEVANCE: Record {}) + // altimate_change end SessionSummary.summarize({ sessionID: sessionID, messageID: lastUser.id, @@ -1176,6 +1189,7 @@ export namespace SessionPrompt { : await MemoryPrompt.inject(UNIFIED_INJECTION_BUDGET, { agent: agent.name, disableTraining: Flag.ALTIMATE_DISABLE_TRAINING, + sessionID, }) // altimate_change end const system = [ diff --git a/packages/opencode/test/altimate/plugin/workspace.test.ts b/packages/opencode/test/altimate/plugin/workspace.test.ts index 02172c67e7..6be367ea79 100644 --- a/packages/opencode/test/altimate/plugin/workspace.test.ts +++ b/packages/opencode/test/altimate/plugin/workspace.test.ts @@ -6,7 +6,7 @@ // This file focuses on the deterministic layer: URL parsing, git detection, // state read/write + chmod, latch semantics, and error classification. import { afterAll, afterEach, beforeEach, describe, expect, test } from "bun:test" -import { existsSync, mkdirSync, rmSync, statSync } from "node:fs" +import { existsSync, mkdirSync, rmSync, statSync, writeFileSync } from "node:fs" import path from "node:path" import os from "node:os" @@ -154,6 +154,151 @@ describe("workspace binding cache", () => { expect(read!.datamateName).toBe("Marketing") }) + test("awaitBackfill holds the bind open until the seed has run", async () => { + // `altimate-code link` runs in a plain yargs handler and src/index.ts calls + // process.exit() the moment it returns, so a detached seed is killed + // mid-flight: the bind reports success having stored nothing. + const ORIGINAL_FLAG = process.env.ALTIMATE_WORKSPACE + process.env.ALTIMATE_WORKSPACE = "1" + const proj = path.join(SANDBOX, "seed-proj") + mkdirSync(path.join(proj, ".altimate-code", "memory"), { recursive: true }) + const now = new Date().toISOString() + writeFileSync( + path.join(proj, ".altimate-code", "memory", "seed.md"), + ["---", "id: seed", "scope: project", `created: ${now}`, `updated: ${now}`, "---", "", "A fact.", ""].join("\n"), + ) + + let release: (() => void) | undefined + const gate = new Promise((r) => (release = r)) + const originalFetch = globalThis.fetch + globalThis.fetch = (async (_input?: unknown, _init?: unknown) => { + await gate + return new Response(JSON.stringify({ datamates: [] }), { + status: 200, + headers: { "Content-Type": "application/json" }, + }) + }) as typeof fetch + + try { + const binding = { + datamateId: 7, + datamateName: "Seeded", + repoRemote: null, + projectPath: proj, + linkedAt: 1, + } + const pending = recordApprovedBinding(proj, binding, { awaitBackfill: true }) + const outcome = await Promise.race([ + pending.then(() => "resolved"), + new Promise((r) => setTimeout(() => r("still-pending"), 200)), + ]) + expect(outcome).toBe("still-pending") + release?.() + await pending + } finally { + release?.() + globalThis.fetch = originalFetch + if (ORIGINAL_FLAG === undefined) delete process.env.ALTIMATE_WORKSPACE + else process.env.ALTIMATE_WORKSPACE = ORIGINAL_FLAG + } + }) + + test("re-recording an unchanged binding does not re-seed", async () => { + // A flow that merely warms the cache must not sweep: the seed is for a new + // or changed bind. `link` now awaits the seed, so a redundant one is paid + // synchronously by the user. + const ORIGINAL_FLAG = process.env.ALTIMATE_WORKSPACE + process.env.ALTIMATE_WORKSPACE = "1" + const proj = path.join(SANDBOX, "warm-proj") + mkdirSync(path.join(proj, ".altimate-code", "memory"), { recursive: true }) + const now = new Date().toISOString() + writeFileSync( + path.join(proj, ".altimate-code", "memory", "warm.md"), + ["---", "id: warm", "scope: project", `created: ${now}`, `updated: ${now}`, "---", "", "A fact.", ""].join("\n"), + ) + const binding = { + datamateId: 9, + datamateName: "Warm", + repoRemote: null, + projectPath: proj, + linkedAt: 1, + } + + let calls = 0 + const originalFetch = globalThis.fetch + globalThis.fetch = (async (_input?: unknown, _init?: unknown) => { + calls++ + return new Response(JSON.stringify({ datamates: [{ id: 9, name: "Warm", memory_enabled: true }] }), { + status: 200, + headers: { "Content-Type": "application/json" }, + }) + }) as typeof fetch + + try { + await recordApprovedBinding(proj, binding, { awaitBackfill: true }) + const afterFirst = calls + expect(afterFirst).toBeGreaterThan(0) + + // Same workspace, same project, later timestamp: a warm, not a rebind. + await recordApprovedBinding(proj, { ...binding, linkedAt: 2 }, { awaitBackfill: true }) + expect(calls).toBe(afterFirst) + + // A genuine rebind to another workspace must still seed. + await recordApprovedBinding(proj, { ...binding, datamateId: 10 }, { awaitBackfill: true }) + expect(calls).toBeGreaterThan(afterFirst) + } finally { + globalThis.fetch = originalFetch + if (ORIGINAL_FLAG === undefined) delete process.env.ALTIMATE_WORKSPACE + else process.env.ALTIMATE_WORKSPACE = ORIGINAL_FLAG + } + }) + + test("a seed that never ran stays retryable on the next warm", async () => { + // Memory disabled at bind time means the sweep is a no-op, not a completed + // seed. Treating it as done left the blocks this machine already holds + // absent from the workspace until a rebind or an unrelated edit. + const ORIGINAL_FLAG = process.env.ALTIMATE_WORKSPACE + process.env.ALTIMATE_WORKSPACE = "1" + const proj = path.join(SANDBOX, "gated-proj") + mkdirSync(path.join(proj, ".altimate-code", "memory"), { recursive: true }) + const now = new Date().toISOString() + writeFileSync( + path.join(proj, ".altimate-code", "memory", "gated.md"), + ["---", "id: gated", "scope: project", `created: ${now}`, `updated: ${now}`, "---", "", "A fact.", ""].join("\n"), + ) + const binding = { + datamateId: 11, + datamateName: "Gated", + repoRemote: null, + projectPath: proj, + linkedAt: 1, + } + + let memoryEnabled = false + let calls = 0 + const originalFetch = globalThis.fetch + globalThis.fetch = (async (_input?: unknown, _init?: unknown) => { + calls++ + return new Response( + JSON.stringify({ datamates: [{ id: 11, name: "Gated", memory_enabled: memoryEnabled }] }), + { status: 200, headers: { "Content-Type": "application/json" } }, + ) + }) as typeof fetch + + try { + await recordApprovedBinding(proj, binding, { awaitBackfill: true }) + const afterGated = calls + // Memory switched on later: the same binding must sweep this time. + memoryEnabled = true + await recordApprovedBinding(proj, { ...binding, linkedAt: 2 }, { awaitBackfill: true }) + expect(calls).toBeGreaterThan(afterGated) + } finally { + globalThis.fetch = originalFetch + if (ORIGINAL_FLAG === undefined) delete process.env.ALTIMATE_WORKSPACE + else process.env.ALTIMATE_WORKSPACE = ORIGINAL_FLAG + } + }) + test("chmods the cache file to 0o600 after write", async () => { await recordApprovedBinding("/work/proj-a", { datamateId: 1, diff --git a/packages/opencode/test/altimate/workspace/memory-sync.test.ts b/packages/opencode/test/altimate/workspace/memory-sync.test.ts new file mode 100644 index 0000000000..5c80a4f6bb --- /dev/null +++ b/packages/opencode/test/altimate/workspace/memory-sync.test.ts @@ -0,0 +1,1059 @@ +// altimate_change - new file +// Unit coverage for the workspace memory mirror +// (src/altimate/workspace/memory-{api,index,sync}.ts). +// +// Network is stubbed at globalThis.fetch, so assertions are about the requests +// the mirror actually issues — method, path, body — rather than a mock's call +// log. Cases claiming "nothing was sent" check a zero request count, not merely +// the absence of a throw. +import { afterAll, afterEach, beforeEach, describe, expect, test } from "bun:test" +import { mkdirSync, rmSync, statSync } from "node:fs" +import path from "node:path" +import os from "node:os" + +// Global.Path.state resolves at module load, so the sandbox must exist before +// the modules under test are imported. +const ORIGINAL_XDG_STATE_HOME = process.env.XDG_STATE_HOME +const ORIGINAL_WORKSPACE_FLAG = process.env.ALTIMATE_WORKSPACE +const SANDBOX = path.join(os.tmpdir(), `altimate-memsync-${process.pid}-${Date.now()}`) +mkdirSync(path.join(SANDBOX, "state"), { recursive: true }) +process.env.XDG_STATE_HOME = path.join(SANDBOX, "state") +process.env.ALTIMATE_WORKSPACE = "1" + +afterAll(() => { + if (ORIGINAL_XDG_STATE_HOME === undefined) delete process.env.XDG_STATE_HOME + else process.env.XDG_STATE_HOME = ORIGINAL_XDG_STATE_HOME + if (ORIGINAL_WORKSPACE_FLAG === undefined) delete process.env.ALTIMATE_WORKSPACE + else process.env.ALTIMATE_WORKSPACE = ORIGINAL_WORKSPACE_FLAG + try { + rmSync(SANDBOX, { recursive: true, force: true }) + } catch { + /* best effort */ + } +}) + +const { indexKey, indexPath, readIndexEntry, recordIndexEntry } = await import( + "../../../src/altimate/workspace/memory-index" +) +const { + archiveBlock, + backfill, + belongsHere, + buildMetadata, + hydrate, + isEnabled, + mirrorBlock, + overlayBlocks, + resetOverlay, + syncInternals, + toBlock, + whenHydrated, +} = await import("../../../src/altimate/workspace/memory-sync") +const { MIRROR_SOURCE, extractRecordId, LIST_LIMIT } = await import( + "../../../src/altimate/workspace/memory-api" +) + +import { AltimateApi } from "../../../src/altimate/api/client" + +const originalIsConfigured = AltimateApi.isConfigured +const originalGetCreds = AltimateApi.getCredentials +type Creds = Awaited> +function stubCreds(tenant: string, apiUrl: string, apiKey = "key-a") { + ;(AltimateApi as unknown as { isConfigured: () => Promise }).isConfigured = async () => true + ;(AltimateApi as unknown as { getCredentials: () => Promise }).getCredentials = async () => + ({ altimateInstanceName: tenant, altimateUrl: apiUrl, altimateApiKey: apiKey }) as Creds +} + +interface Captured { + method: string + url: string + body: any +} +const originalFetch = globalThis.fetch +let captured: Captured[] = [] +let listResponse: any[] = [] +let createResult: any = [{ id: "mem-new" }] +/** Workspaces returned by GET /datamates/ — drives the memory_enabled gate. */ +let workspaces: any[] = [{ id: 42, name: "acme", memory_enabled: true }] +/** When set, GET /list fails -- used to prove a failed lookup never creates. */ +let listFails = false + +function stubFetch() { + globalThis.fetch = (async (input: any, init?: any) => { + const url = String(input) + const method = (init?.method ?? "GET").toUpperCase() + captured.push({ method, url, body: init?.body ? JSON.parse(init.body) : undefined }) + if (listFails && url.includes("/datamates/memory/list")) { + return new Response(JSON.stringify({ detail: "boom" }), { status: 500 }) + } + const payload = (() => { + if (url.includes("/datamates/memory/list")) return listResponse + if (url.includes("/datamates/memory/")) { + // A created record becomes visible to later reads, as it would on the + // real service. Without this, anything that creates and then looks the + // record up again sees an empty store. + if (method === "POST") { + const body = init?.body ? JSON.parse(init.body) : {} + for (const rec of Array.isArray(createResult) ? createResult : []) { + if (rec?.id) listResponse.push({ id: rec.id, memory: body.messages?.[0]?.content ?? "", metadata: body.metadata }) + } + } + return { message: "ok", result: createResult } + } + if (url.includes("/datamates/")) return { datamates: workspaces } + return { message: "ok" } + })() + return new Response(JSON.stringify(payload), { + status: 200, + headers: { "Content-Type": "application/json" }, + }) + }) as typeof fetch +} + +function callsTo(fragment: string, method?: string): Captured[] { + return captured.filter((c) => c.url.includes(fragment) && (!method || c.method === method)) +} + +const SES = "ses_test" +const NOW = "2026-08-19T00:00:00.000Z" +function block(over: Partial = {}): any { + return { + id: "warehouse/snowflake", + scope: "global", + tags: ["warehouse"], + created: NOW, + updated: NOW, + content: "Snowflake account is acme-prod.", + ...over, + } +} + +const BINDING = { + datamateId: 42, + datamateName: "acme", + repoRemote: "ssh://git@github.com/acme/analytics.git", + projectPath: "/work/analytics", + linkedAt: 1, +} + +beforeEach(() => { + captured = [] + listResponse = [] + listFails = false + createResult = [{ id: "mem-new" }] + workspaces = [{ id: 42, name: "acme", memory_enabled: true }] + stubCreds("acme", "https://api.example.com") + stubFetch() + resetOverlay() + syncInternals.resolveBinding = async () => BINDING as any + // The store is not on disk in these tests; the delete-race guard has its + // own coverage below. + syncInternals.blockExists = async () => true +}) + +afterEach(() => { + globalThis.fetch = originalFetch + ;(AltimateApi as unknown as { isConfigured: typeof originalIsConfigured }).isConfigured = + originalIsConfigured + ;(AltimateApi as unknown as { getCredentials: typeof originalGetCreds }).getCredentials = + originalGetCreds + process.env.ALTIMATE_WORKSPACE = "1" + delete syncInternals.resolveBinding + delete syncInternals.blockExists +}) + +// ── record id extraction ──────────────────────────────────────────────────── +describe("extractRecordId", () => { + test("reads an id from each envelope the store is known to use", () => { + expect(extractRecordId([{ id: "a" }])).toBe("a") + expect(extractRecordId({ results: [{ id: "b" }] })).toBe("b") + expect(extractRecordId({ memories: [{ id: "c" }] })).toBe("c") + expect(extractRecordId({ data: [{ memory_id: "d" }] })).toBe("d") + }) + + test("reports nothing rather than guessing when no id is present", () => { + // A declined create stores nothing and returns no usable id; inventing one + // would index a record that does not exist. + expect(extractRecordId(undefined)).toBeUndefined() + expect(extractRecordId([])).toBeUndefined() + expect(extractRecordId({ message: "nothing stored" })).toBeUndefined() + expect(extractRecordId([{ nope: 1 }])).toBeUndefined() + }) +}) + +// ── index keying ──────────────────────────────────────────────────────────── +describe("indexKey", () => { + test("global keys ignore workspace and project", () => { + expect(indexKey({ scope: "global", blockId: "b" })).toBe( + indexKey({ scope: "global", blockId: "b", datamateId: 7, projectKey: "x" }), + ) + }) + + test("one block id in two projects of a workspace does not collide", () => { + // Block ids are unique only within a project directory, so two bound + // projects can each hold a 'warehouse/snowflake'. + const a = indexKey({ scope: "project", blockId: "w/s", datamateId: 42, projectKey: "repo-a" }) + const b = indexKey({ scope: "project", blockId: "w/s", datamateId: 42, projectKey: "repo-b" }) + expect(a).not.toBe(b) + }) + + test("one project in two workspaces does not collide", () => { + const a = indexKey({ scope: "project", blockId: "x", datamateId: 1, projectKey: "repo" }) + const b = indexKey({ scope: "project", blockId: "x", datamateId: 2, projectKey: "repo" }) + expect(a).not.toBe(b) + }) +}) + +describe("index persistence", () => { + test("round-trips an entry and writes it 0600", async () => { + const key = indexKey({ scope: "global", blockId: "round-trip" }) + await recordIndexEntry(key, { memoryId: "mem-1", contentHash: "h", syncedAt: 1 }) + expect((await readIndexEntry(key))?.memoryId).toBe("mem-1") + expect(statSync(indexPath()).mode & 0o777).toBe(0o600) + }) + + test("a tenant switch invalidates the map", async () => { + const key = indexKey({ scope: "global", blockId: "tenant-scoped" }) + await recordIndexEntry(key, { memoryId: "mem-tenant-a", contentHash: "h", syncedAt: 1 }) + stubCreds("other-tenant", "https://api.example.com") + expect(await readIndexEntry(key)).toBeNull() + }) + + test("a different account on the same tenant does not inherit the map", async () => { + // Records are per-user. Two people sharing a machine and a tenant would + // otherwise update each other's records. + const key = indexKey({ scope: "global", blockId: "account-scoped" }) + await recordIndexEntry(key, { memoryId: "mem-user-a", contentHash: "h", syncedAt: 1 }) + expect((await readIndexEntry(key))?.memoryId).toBe("mem-user-a") + + stubCreds("acme", "https://api.example.com", "key-b") + expect(await readIndexEntry(key)).toBeNull() + }) +}) + +// ── gating ────────────────────────────────────────────────────────────────── +describe("gating", () => { + test("disabled unless the pilot flag is set", () => { + delete process.env.ALTIMATE_WORKSPACE + expect(isEnabled()).toBe(false) + process.env.ALTIMATE_WORKSPACE = "1" + expect(isEnabled()).toBe(true) + }) + + test("a write issues no request at all when the flag is off", async () => { + delete process.env.ALTIMATE_WORKSPACE + await mirrorBlock(block({ id: "flag-off" })) + expect(captured.length).toBe(0) + }) + + test("hydrate issues no request when the flag is off", async () => { + delete process.env.ALTIMATE_WORKSPACE + await hydrate("ses_flag_off") + expect(captured.length).toBe(0) + expect(overlayBlocks(SES)).toEqual([]) + }) + + test("nothing is mirrored from an unbound directory, at either scope", async () => { + // The mirror is inert until the user binds a project. Global blocks carry + // no workspace themselves, but the binding is what makes the mirror active + // and what carries the memory_enabled setting. + syncInternals.resolveBinding = async () => null + await mirrorBlock(block({ id: "unbound", scope: "project" })) + await mirrorBlock(block({ id: "unbound-global", scope: "global" })) + expect(captured.length).toBe(0) + }) + + test("hydration returns nothing from an unbound directory", async () => { + syncInternals.resolveBinding = async () => null + listResponse = [{ id: "1", memory: "x", metadata: { source: MIRROR_SOURCE, block_id: "x", block_scope: "global" } }] + await hydrate(SES) + expect(overlayBlocks(SES)).toEqual([]) + }) +}) + +// ── metadata ──────────────────────────────────────────────────────────────── +describe("buildMetadata", () => { + test("project scope carries the workspace and the originating project", () => { + const meta = buildMetadata(block({ scope: "project", tags: ["a", "b"] }), BINDING as any) + expect(meta.datamate_id).toBe("42") + expect(meta.datamate_name).toBe("acme") + expect(meta.repo_remote).toBe(BINDING.repoRemote) + expect(meta.project_path).toBe(BINDING.projectPath) + expect(meta.source).toBe(MIRROR_SOURCE) + expect(meta.block_scope).toBe("project") + expect(meta.block_tags).toBe('["a","b"]') + }) + + test("global scope carries no workspace even when a binding exists", () => { + // Global memory belongs to the account and applies in every workspace; + // stamping a workspace would pin it to one and hide it from the others. + const meta = buildMetadata(block({ scope: "global" }), BINDING as any) + expect(meta.datamate_id).toBeUndefined() + expect(meta.datamate_name).toBeUndefined() + expect(meta.repo_remote).toBeUndefined() + expect(meta.block_scope).toBe("global") + }) + + test("expiry is mirrored, so a TTL'd block can expire everywhere", () => { + // Without this a TTL'd block lives forever in the workspace once it has + // left the machine that wrote it, and is injected into every session. + const meta = buildMetadata(block({ expires: "2027-01-01T00:00:00.000Z" }), null) + expect(meta.block_expires).toBe("2027-01-01T00:00:00.000Z") + }) + + test("every record is marked private", () => { + expect(buildMetadata(block(), null).visibility).toBe("private") + }) + + test("created and updated timestamps are carried", () => { + const meta = buildMetadata(block({ created: NOW, updated: NOW }), null) + expect(meta.block_created).toBe(NOW) + expect(meta.block_updated).toBe(NOW) + }) +}) + +// ── write path ────────────────────────────────────────────────────────────── +describe("mirrorBlock", () => { + test("a create is repaired with a verbatim update", async () => { + // A create runs an extractor that rewrites the text; update() is verbatim, + // so every create is followed by one. + const b = block({ id: "repair-me" }) + createResult = [{ id: "mem-repair" }] + await mirrorBlock(b) + + expect(callsTo("/datamates/memory/", "POST").length).toBe(1) + const patches = callsTo("/datamates/memory/mem-repair", "PATCH") + expect(patches.length).toBe(1) + expect(patches[0].body.memory).toBe(b.content) + expect(patches[0].body.metadata.block_id).toBe("repair-me") + }) + + test("a known block updates the record the index names", async () => { + // The index supplies the id, so the update must never create. The record + // set is still read: every safety check below the lookup needs it, and + // when only `backfill` supplied one those checks were unreachable on the + // ordinary per-save path. + const b = block({ id: "known" }) + createResult = [{ id: "mem-known" }] + await mirrorBlock(b) + captured = [] + await mirrorBlock({ ...b, updated: "2026-08-20T00:00:00.000Z", content: "changed" }) + expect(callsTo("/datamates/memory/", "POST").length).toBe(0) + expect(callsTo("/datamates/memory/mem-known", "PATCH").length).toBe(1) + }) + + test("a failed lookup defers instead of creating a duplicate", async () => { + // Treating an unreadable record set as "no record exists" is how the same + // block gets created twice. + listFails = true + await mirrorBlock(block({ id: "unreadable" })) + expect(callsTo("/datamates/memory/", "POST").length).toBe(0) + }) + + test("a truncated read blocks a create on the ordinary save path", async () => { + // The guard previously only ran under `backfill`, so per-save mirroring + // still created duplicates of records sitting past the window. + listResponse = Array.from({ length: LIST_LIMIT }, (_, i) => ({ + id: `mem-${i}`, + memory: "other", + metadata: { source: MIRROR_SOURCE, block_id: `other-${i}`, block_scope: "global" }, + })) + await mirrorBlock(block({ id: "past-the-window" })) + expect(callsTo("/datamates/memory/", "POST").length).toBe(0) + }) + + test("a record set larger than the limit is complete, not truncated", async () => { + // The service ignores paging, so more rows than the limit proves the set + // came back whole. Treating that as truncated would permanently block + // creates for any user with a large workspace. + listResponse = Array.from({ length: LIST_LIMIT + 1 }, (_, i) => ({ + id: `mem-${i}`, + memory: "other", + metadata: { source: MIRROR_SOURCE, block_id: `other-${i}`, block_scope: "global" }, + })) + await mirrorBlock(block({ id: "still-creatable" })) + expect(callsTo("/datamates/memory/", "POST").length).toBe(1) + }) + + test("the newer-remote guard applies on the ordinary save path", async () => { + // Previously `remote` was only resolved from the backfill prefetch, so a + // stale machine could overwrite a newer cloud value on every normal save. + listResponse = [ + { + id: "mem-newer", + memory: "newer text from another machine", + metadata: { + source: MIRROR_SOURCE, + block_id: "contested", + block_scope: "global", + block_updated: "2027-01-01T00:00:00.000Z", + }, + }, + ] + await mirrorBlock(block({ id: "contested", updated: NOW, content: "older local text" })) + expect(callsTo("/datamates/memory/mem-newer", "PATCH").length).toBe(0) + }) + + test("an edit is still mirrored when its content collides under a 32-bit hash", async () => { + // "fact loczw" and "fact vfbpa" share an FNV-1a 32-bit digest for the exact + // payload contentHash builds. The hash is the only gate deciding whether a + // save is sent, so under a 32-bit digest this edit is silently dropped with + // no retry -- the block's cloud copy just stays wrong. + const b = block({ id: "collide", tags: ["warehouse"], content: "fact loczw" }) + createResult = [{ id: "mem-collide" }] + await mirrorBlock(b) + captured = [] + await mirrorBlock({ ...b, content: "fact vfbpa", updated: "2026-08-20T00:00:00.000Z" }) + expect(callsTo("/datamates/memory/mem-collide", "PATCH").length).toBe(1) + }) + + test("a block deleted mid-sweep is not recreated", async () => { + // `backfill` registers a block on the serialize queue only when a worker + // dequeues it, so a delete issued during the sweep runs first. Without the + // local-existence check this push creates a live record for a block the + // user just deleted -- and indexes it, so no later sweep re-archives it. + syncInternals.blockExists = async () => false + await mirrorBlock(block({ id: "deleted-mid-sweep" })) + expect(callsTo("/datamates/memory/", "POST").length).toBe(0) + }) + + test("a block deleted mid-sweep does not revive its archived record", async () => { + const b = block({ id: "tombstoned" }) + createResult = [{ id: "mem-tomb" }] + await mirrorBlock(b) + // Archived remotely, and gone locally. + listResponse = listResponse.map((r: any) => + r.id === "mem-tomb" ? { ...r, metadata: { ...r.metadata, archived: "true" } } : r, + ) + syncInternals.blockExists = async () => false + captured = [] + await mirrorBlock({ ...b, content: "changed", updated: "2027-01-01T00:00:00.000Z" }) + expect(callsTo("/datamates/memory/mem-tomb", "PATCH").length).toBe(0) + }) + + test("a recreated block gets a fresh record rather than un-archiving the old one", async () => { + // `MemoryApi.update` replaces metadata wholesale, so updating the tombstone + // would drop `archived` and bring the deleted record back to life. + const b = block({ id: "reborn" }) + createResult = [{ id: "mem-old" }] + await mirrorBlock(b) + listResponse = listResponse.map((r: any) => + r.id === "mem-old" ? { ...r, metadata: { ...r.metadata, archived: "true" } } : r, + ) + createResult = [{ id: "mem-new" }] + captured = [] + await mirrorBlock({ ...b, content: "written again", updated: "2027-01-01T00:00:00.000Z" }) + expect(callsTo("/datamates/memory/mem-old", "PATCH").length).toBe(0) + expect(callsTo("/datamates/memory/", "POST").length).toBe(1) + }) + + test("an unchanged block costs no read at all", async () => { + // The content hash short-circuits before the record set is fetched, so a + // no-op save stays free. + const b = block({ id: "cheap" }) + createResult = [{ id: "mem-cheap" }] + await mirrorBlock(b) + captured = [] + await mirrorBlock(b) + expect(callsTo("/datamates/memory/list").length).toBe(0) + expect(callsTo("/datamates/memory/mem-cheap", "PATCH").length).toBe(0) + }) + + test("an unknown block looks for an existing record before creating one", async () => { + // This is the convergence guarantee: a second machine, a reinstalled CLI, + // or a create whose response was lost would otherwise duplicate the record. + listResponse = [ + { + id: "mem-elsewhere", + memory: "written by another machine", + metadata: { source: MIRROR_SOURCE, block_id: "converge", block_scope: "global" }, + }, + ] + await mirrorBlock(block({ id: "converge" })) + expect(callsTo("/datamates/memory/", "POST").length).toBe(0) + expect(callsTo("/datamates/memory/mem-elsewhere", "PATCH").length).toBe(1) + }) + + test("re-saving an unchanged block sends nothing", async () => { + const b = block({ id: "unchanged" }) + await mirrorBlock(b) + captured = [] + await mirrorBlock(b) + expect(captured.length).toBe(0) + }) + + test("a changed block updates the known record instead of creating a second", async () => { + const b = block({ id: "changed" }) + createResult = [{ id: "mem-changed" }] + await mirrorBlock(b) + captured = [] + + await mirrorBlock({ ...b, updated: "2026-08-20T00:00:00.000Z", content: "Now acme-staging." }) + expect(callsTo("/datamates/memory/", "POST").length).toBe(0) + const patches = callsTo("/datamates/memory/mem-changed", "PATCH") + expect(patches.length).toBe(1) + expect(patches[0].body.memory).toBe("Now acme-staging.") + }) + + test("a block the store declines is not indexed, so a later edit retries", async () => { + createResult = [] + await mirrorBlock(block({ id: "declined" })) + expect(callsTo("/datamates/memory/", "POST").length).toBe(1) + expect(captured.filter((c) => c.method === "PATCH").length).toBe(0) + + captured = [] + createResult = [{ id: "mem-later" }] + await mirrorBlock(block({ id: "declined" })) + expect(callsTo("/datamates/memory/", "POST").length).toBe(1) + }) + + test("an applied-count bump on a training block sends nothing", async () => { + // incrementApplied rewrites a training block on EVERY session start to bump + // a counter inside the content body. Mirroring that would cost a request + // per training block per session. + const meta = (applied: number) => + `\nAlways alias CTEs.` + const b = block({ id: "training/rule/aliases", tags: ["training", "rule"], content: meta(1) }) + await mirrorBlock(b) + captured = [] + await mirrorBlock({ ...b, updated: "2026-08-20T00:00:00.000Z", content: meta(2) }) + expect(captured.length).toBe(0) + }) + + test("a real edit to a training block still syncs", async () => { + const meta = (n: number, body: string) => `\n${body}` + const b = block({ id: "training/rule/real", tags: ["training"], content: meta(1, "Alias CTEs.") }) + createResult = [{ id: "mem-real" }] + await mirrorBlock(b) + captured = [] + await mirrorBlock({ ...b, updated: "2026-08-20T00:00:00.000Z", content: meta(2, "Never alias.") }) + expect(callsTo("/datamates/memory/mem-real", "PATCH").length).toBe(1) + }) + + test("a tag-only change still syncs, since tags are mirrored", async () => { + const b = block({ id: "tag-change" }) + createResult = [{ id: "mem-tags" }] + await mirrorBlock(b) + captured = [] + await mirrorBlock({ ...b, updated: "2026-08-20T00:00:00.000Z", tags: ["warehouse", "prod"] }) + expect(callsTo("/datamates/memory/mem-tags", "PATCH").length).toBe(1) + }) +}) + +// ── memory_enabled gate ───────────────────────────────────────────────────── +describe("memory_enabled", () => { + test("nothing is written when the workspace has memory disabled", async () => { + // The workspace app shows this as a toggle, so writing into a disabled + // workspace would contradict what the user sees. + workspaces = [{ id: 42, name: "acme", memory_enabled: false }] + await mirrorBlock(block({ id: "disabled", scope: "global" })) + expect(callsTo("/datamates/memory/", "POST").length).toBe(0) + }) + + test("enabling memory takes effect immediately, without waiting out a cache", async () => { + // A workspace starts with memory disabled, so a user's first action is + // often to switch it on. Caching the disabled verdict would ignore that + // for a full TTL. + workspaces = [{ id: 42, name: "acme", memory_enabled: false }] + await mirrorBlock(block({ id: "before-enable" })) + expect(callsTo("/datamates/memory/", "POST").length).toBe(0) + + workspaces = [{ id: 42, name: "acme", memory_enabled: true }] + captured = [] + await mirrorBlock(block({ id: "after-enable" })) + expect(callsTo("/datamates/memory/", "POST").length).toBe(1) + }) + + test("the gate fails closed when the workspace cannot be read", async () => { + workspaces = [] + await mirrorBlock(block({ id: "unknown-ws", scope: "global" })) + expect(callsTo("/datamates/memory/", "POST").length).toBe(0) + }) +}) + +// ── read path ─────────────────────────────────────────────────────────────── +describe("toBlock", () => { + const record = (metadata: any, over: any = {}) => ({ + id: "rec", + memory: "content here", + created_at: NOW, + updated_at: NOW, + metadata, + ...over, + }) + + test("maps metadata back onto block fields", () => { + const b = toBlock( + record({ + source: MIRROR_SOURCE, + block_id: "warehouse/snowflake", + block_scope: "project", + block_updated: NOW, + block_tags: "warehouse,prod", + }), + undefined, + ) + expect(b?.id).toBe("warehouse/snowflake") + expect(b?.scope).toBe("project") + expect(b?.tags).toEqual(["warehouse", "prod"]) + expect(b?.content).toBe("content here") + expect(b?.remote).toBe(true) + }) + + test("a record from a sibling project is labelled with its origin", () => { + const b = toBlock( + record({ + source: MIRROR_SOURCE, + block_id: "x", + block_scope: "project", + repo_remote: "ssh://git@github.com/acme/other-repo.git", + }), + "ssh://git@github.com/acme/analytics.git", + ) + expect(b?.origin).toBe("other-repo") + }) + + test("a record from this same project is not labelled", () => { + const own = "ssh://git@github.com/acme/analytics.git" + const b = toBlock( + record({ source: MIRROR_SOURCE, block_id: "x", block_scope: "project", repo_remote: own }), + own, + ) + expect(b?.origin).toBeUndefined() + }) + + test("restores expiry from metadata", () => { + const b = toBlock( + record({ + source: MIRROR_SOURCE, + block_id: "ttl", + block_scope: "global", + block_expires: "2027-01-01T00:00:00.000Z", + }), + undefined, + ) + expect(b?.expires).toBe("2027-01-01T00:00:00.000Z") + }) + + test("rejects records that are not well-formed mirrored blocks", () => { + expect(toBlock(record(null), undefined)).toBeNull() + expect(toBlock(record({ source: MIRROR_SOURCE, block_scope: "global" }), undefined)).toBeNull() + expect(toBlock(record({ source: MIRROR_SOURCE, block_id: "x", block_scope: "nope" }), undefined)).toBeNull() + expect( + toBlock(record({ source: MIRROR_SOURCE, block_id: "x", block_scope: "global" }, { memory: "" }), undefined), + ).toBeNull() + }) +}) + +describe("tag encoding", () => { + test("a tag containing a comma survives the round trip", () => { + // The previous comma-joined form split this tag into two on read. + const tags = ["warehouse", "owner: data, platform", "prod"] + const meta = buildMetadata(block({ tags }), null) + const back = toBlock( + { id: "1", memory: "x", metadata: { ...meta, source: MIRROR_SOURCE } } as any, + undefined, + ) + expect(back?.tags).toEqual(tags) + }) + + test("records written in the legacy comma form still decode", () => { + const back = toBlock( + { + id: "1", + memory: "x", + metadata: { source: MIRROR_SOURCE, block_id: "old", block_scope: "global", block_tags: "a,b" }, + } as any, + undefined, + ) + expect(back?.tags).toEqual(["a", "b"]) + }) +}) + +describe("belongsHere", () => { + const rec = (metadata: any) => ({ id: "r", memory: "m", metadata }) + + test("global records apply everywhere, including with no workspace bound", () => { + // Omitting this arm would make global memory write-only. + expect(belongsHere(rec({ block_scope: "global" }), "42")).toBe(true) + expect(belongsHere(rec({ block_scope: "global" }), undefined)).toBe(true) + }) + + test("a project record applies only to its own workspace", () => { + expect(belongsHere(rec({ block_scope: "project", datamate_id: "42" }), "42")).toBe(true) + expect(belongsHere(rec({ block_scope: "project", datamate_id: "99" }), "42")).toBe(false) + }) + + test("a project record is excluded when nothing is bound", () => { + expect(belongsHere(rec({ block_scope: "project", datamate_id: "42" }), undefined)).toBe(false) + }) +}) + +describe("hydrate", () => { + test("keeps only this CLI's records", async () => { + // The discriminating case is the last one: block-shaped metadata written by + // something else. Only the source check rejects it. + listResponse = [ + { id: "1", memory: "cli", metadata: { source: MIRROR_SOURCE, block_id: "cli", block_scope: "global" } }, + { id: "2", memory: "chat", metadata: { datamate_id: "42" } }, + { id: "3", memory: "none", metadata: null }, + { id: "4", memory: "impostor", metadata: { block_id: "imp", block_scope: "global" } }, + ] + await hydrate(SES) + expect(overlayBlocks(SES).map((b) => b.id)).toEqual(["cli"]) + }) + + test("drops records whose mirrored expiry has passed", async () => { + // The cloud copy is not swept, so an expired block must be honoured on + // read or it outlives its TTL on every other machine. + listResponse = [ + { id: "1", memory: "live", metadata: { source: MIRROR_SOURCE, block_id: "live", block_scope: "global" } }, + { + id: "2", + memory: "stale", + metadata: { + source: MIRROR_SOURCE, + block_id: "stale", + block_scope: "global", + block_expires: "2020-01-01T00:00:00.000Z", + }, + }, + ] + await hydrate(SES) + expect(overlayBlocks(SES).map((b) => b.id)).toEqual(["live"]) + }) + + test("drops archived records", async () => { + listResponse = [ + { id: "1", memory: "live", metadata: { source: MIRROR_SOURCE, block_id: "live", block_scope: "global" } }, + { id: "2", memory: "gone", metadata: { source: MIRROR_SOURCE, block_id: "gone", block_scope: "global", archived: "true" } }, + ] + await hydrate(SES) + expect(overlayBlocks(SES).map((b) => b.id)).toEqual(["live"]) + }) + + test("drops project records belonging to another workspace", async () => { + listResponse = [ + { id: "1", memory: "g", metadata: { source: MIRROR_SOURCE, block_id: "g", block_scope: "global" } }, + { id: "2", memory: "other", metadata: { source: MIRROR_SOURCE, block_id: "other", block_scope: "project", datamate_id: "999" } }, + ] + await hydrate(SES) + expect(overlayBlocks(SES).map((b) => b.id)).toEqual(["g"]) + }) + + test("opts in explicitly, or the backend returns nothing", async () => { + // The backend excludes this client's records from list by default. + listResponse = [{ id: "1", memory: "x", metadata: { source: MIRROR_SOURCE, block_id: "x", block_scope: "global" } }] + await hydrate(SES) + expect(callsTo("/datamates/memory/list")[0].url).toContain("include_sources=altimate-code") + }) + + test("is one-shot per session", async () => { + listResponse = [{ id: "1", memory: "x", metadata: { source: MIRROR_SOURCE, block_id: "x", block_scope: "global" } }] + await hydrate("ses_once") + expect(callsTo("/datamates/memory/list").length).toBe(1) + await hydrate("ses_once") + expect(callsTo("/datamates/memory/list").length).toBe(1) + }) + + test("a failing fetch leaves an empty overlay rather than throwing", async () => { + globalThis.fetch = (async () => { + throw new Error("network down") + }) as unknown as typeof fetch + await hydrate(SES) + expect(overlayBlocks(SES)).toEqual([]) + }) + + test("returns nothing when the workspace has memory disabled", async () => { + workspaces = [{ id: 42, name: "acme", memory_enabled: false }] + listResponse = [{ id: "1", memory: "x", metadata: { source: MIRROR_SOURCE, block_id: "x", block_scope: "global" } }] + await hydrate(SES) + expect(overlayBlocks(SES)).toEqual([]) + }) +}) + +describe("whenHydrated", () => { + test("a hydration that blew its budget is not waited on again", async () => { + // The promise stays unresolved, so without the timed-out latch every later + // injection in the session would pay the full timeout again. + let release: (() => void) | undefined + const stall = new Promise((r) => (release = r)) + const original = globalThis.fetch + globalThis.fetch = (async (input: any, init?: any) => { + if (String(input).includes("/datamates/memory/list")) { + await stall + } + return original(input, init) + }) as typeof fetch + + try { + void hydrate("stalled") + const first = Date.now() + await whenHydrated("stalled", 40) + expect(Date.now() - first).toBeGreaterThanOrEqual(30) + + const second = Date.now() + await whenHydrated("stalled", 40) + expect(Date.now() - second).toBeLessThan(20) + } finally { + release?.() + globalThis.fetch = original + } + }) +}) + +describe("archiveBlock", () => { + test("archives in place and never issues a DELETE", async () => { + const b = block({ id: "to-archive" }) + createResult = [{ id: "mem-archive" }] + await mirrorBlock(b) + captured = [] + listResponse = [ + { id: "mem-archive", memory: b.content, metadata: { source: MIRROR_SOURCE, block_id: "to-archive", block_scope: "global" } }, + ] + + await archiveBlock("global", "to-archive") + expect(captured.filter((c) => c.method === "DELETE").length).toBe(0) + const patches = callsTo("/datamates/memory/mem-archive", "PATCH") + expect(patches.length).toBe(1) + expect(patches[0].body.metadata.archived).toBe("true") + // Archiving hides a record from injection; it does not erase what it said. + expect(patches[0].body.memory).toBe(b.content) + }) + + test("archives by logical identity when the local index cannot answer", async () => { + // The index is per-machine and is discarded on account switch, corruption, + // or a wiped state directory. Without this fallback a delete leaves the + // record live and every later session re-injects it, with no way to remove it. + listResponse = [ + { + id: "mem-elsewhere", + memory: "written by another machine", + metadata: { source: MIRROR_SOURCE, block_id: "orphaned", block_scope: "global" }, + }, + ] + await archiveBlock("global", "orphaned") + const patches = callsTo("/datamates/memory/mem-elsewhere", "PATCH") + expect(patches.length).toBe(1) + expect(patches[0].body.metadata.archived).toBe("true") + }) + + test("does not archive another workspace's record with the same block id", async () => { + listResponse = [ + { + id: "mem-other-ws", + memory: "different workspace", + metadata: { source: MIRROR_SOURCE, block_id: "shared-id", block_scope: "project", datamate_id: "999" }, + }, + ] + await archiveBlock("project", "shared-id") + expect(captured.filter((c) => c.method === "PATCH").length).toBe(0) + }) + + test("does not archive a sibling project's record in the same workspace", async () => { + // Same workspace, same block id, different project. Matching on workspace + // alone would archive the wrong project's memory. + listResponse = [ + { + id: "mem-sibling", + memory: "belongs to another project in this workspace", + metadata: { + source: MIRROR_SOURCE, + block_id: "shared-id", + block_scope: "project", + datamate_id: "42", + repo_remote: "ssh://git@github.com/acme/other.git", + }, + }, + ] + await archiveBlock("project", "shared-id") + expect(captured.filter((c) => c.method === "PATCH").length).toBe(0) + }) + + test("archives this project's own record when the index is unavailable", async () => { + listResponse = [ + { + id: "mem-mine", + memory: "belongs to this project", + metadata: { + source: MIRROR_SOURCE, + block_id: "shared-id", + block_scope: "project", + datamate_id: "42", + repo_remote: BINDING.repoRemote, + }, + }, + ] + await archiveBlock("project", "shared-id") + expect(callsTo("/datamates/memory/mem-mine", "PATCH").length).toBe(1) + }) + + test("a delete issued during an in-flight mirror cannot resurrect the block", async () => { + // Both hooks are fire-and-forget from the store. Unserialized, the archive + // runs before the create it is meant to undo, and the mirror then leaves a + // live record behind that later sessions rehydrate. + const b = block({ id: "raced", scope: "global" }) + createResult = [{ id: "mem-raced" }] + const mirroring = mirrorBlock(b) + const archiving = archiveBlock("global", "raced") + await Promise.all([mirroring, archiving]) + const patches = callsTo("/datamates/memory/mem-raced", "PATCH") + const archivePatch = patches.find((c) => c.body?.metadata?.archived === "true") + expect(archivePatch).toBeDefined() + }) + + test("an already-archived record is not archived again", async () => { + listResponse = [ + { + id: "mem-done", + memory: "x", + metadata: { source: MIRROR_SOURCE, block_id: "done", block_scope: "global", archived: "true" }, + }, + ] + await archiveBlock("global", "done") + expect(captured.filter((c) => c.method === "PATCH").length).toBe(0) + }) +}) + +describe("backfill", () => { + // Ids are unique per test: the index file lives in the sandbox and persists + // across tests, so reused ids would be skipped as already-synced rather than + // exercising the path under test. + const blocks = (prefix: string, n: number) => + Array.from({ length: n }, (_, i) => + block({ id: `${prefix}/${i}`, content: `Block ${prefix} ${i} describes a warehouse convention.` }), + ) + + test("reads the record set once for the whole sweep, not once per block", async () => { + // Every block in a first bind is an index miss, so resolving each through + // its own lookup made a bind cost one full fetch per block. + createResult = [{ id: "mem-x" }] + await backfill(blocks("once", 5), BINDING as any) + expect(callsTo("/datamates/memory/list").length).toBe(1) + }) + + test("counts declines separately from stored blocks", async () => { + // Reporting a decline as success made a sweep claim it had seeded blocks + // the service never kept. + createResult = [] + const result = await backfill(blocks("declined", 3), BINDING as any) + expect(result.declined).toBe(3) + expect(result.ok).toBe(0) + }) + + test("skips blocks already synced at their current payload", async () => { + createResult = [{ id: "mem-resume" }] + const set = blocks("resume", 2) + await backfill(set, BINDING as any) + captured = [] + const second = await backfill(set, BINDING as any) + expect(second.skipped).toBe(2) + expect(callsTo("/datamates/memory/", "POST").length).toBe(0) + }) + + test("does nothing when the workspace has memory disabled", async () => { + workspaces = [{ id: 42, name: "acme", memory_enabled: false }] + const result = await backfill(blocks("disabled", 2), BINDING as any) + expect(result.skipped).toBe(2) + expect(callsTo("/datamates/memory/", "POST").length).toBe(0) + }) +}) + +describe("truncated reads", () => { + test("refuses to create against a record set that hit the service limit", async () => { + // The record may exist just beyond the window, so creating would duplicate + // it. Leaving the block unindexed means a later save retries. + listResponse = Array.from({ length: 200 }, (_, i) => ({ + id: `r${i}`, + memory: "x", + metadata: { source: MIRROR_SOURCE, block_id: `other/${i}`, block_scope: "global" }, + })) + const result = await backfill([block({ id: "beyond/window" })], BINDING as any) + expect(callsTo("/datamates/memory/", "POST").length).toBe(0) + expect(result.skipped).toBeGreaterThan(0) + }) +}) + +describe("session isolation and turn behaviour", () => { + test("a session hydrates once, however many turns it takes", async () => { + // The caller's enclosing block runs on EVERY user turn, not once per + // session. Without idempotence a 20-turn conversation costs 40 requests, + // and the first injection of every turn blocks on them. + listResponse = [ + { id: "1", memory: "x", metadata: { source: MIRROR_SOURCE, block_id: "x", block_scope: "global" } }, + ] + await hydrate(SES) + await hydrate(SES) + await hydrate(SES) + expect(callsTo("/datamates/memory/list").length).toBe(1) + expect(overlayBlocks(SES).map((b) => b.id)).toEqual(["x"]) + }) + + test("a second turn never empties the overlay while refetching", async () => { + // Clearing before a refetch made workspace memory blink out of the prompt + // on any turn whose fetch ran long. + listResponse = [ + { id: "1", memory: "x", metadata: { source: MIRROR_SOURCE, block_id: "kept", block_scope: "global" } }, + ] + await hydrate(SES) + expect(overlayBlocks(SES).length).toBe(1) + await hydrate(SES) + expect(overlayBlocks(SES).length).toBe(1) + }) + + test("two concurrent sessions do not read each other's overlay", async () => { + // A shared module-level overlay let a session in one workspace be injected + // with another workspace's private memory. + listResponse = [ + { id: "a", memory: "alpha", metadata: { source: MIRROR_SOURCE, block_id: "alpha", block_scope: "global" } }, + ] + await hydrate("ses_a") + + listResponse = [ + { id: "b", memory: "beta", metadata: { source: MIRROR_SOURCE, block_id: "beta", block_scope: "global" } }, + ] + await hydrate("ses_b") + + expect(overlayBlocks("ses_a").map((b) => b.id)).toEqual(["alpha"]) + expect(overlayBlocks("ses_b").map((b) => b.id)).toEqual(["beta"]) + }) + + test("overlayBlocks returns a copy, so a caller cannot corrupt the cache", async () => { + listResponse = [ + { id: "1", memory: "x", metadata: { source: MIRROR_SOURCE, block_id: "x", block_scope: "global" } }, + ] + await hydrate(SES) + overlayBlocks(SES).length = 0 + expect(overlayBlocks(SES).length).toBe(1) + }) + + test("resetting one session leaves the others intact", async () => { + listResponse = [ + { id: "a", memory: "alpha", metadata: { source: MIRROR_SOURCE, block_id: "alpha", block_scope: "global" } }, + ] + await hydrate("ses_a") + await hydrate("ses_b") + resetOverlay("ses_a") + expect(overlayBlocks("ses_a")).toEqual([]) + expect(overlayBlocks("ses_b").map((b) => b.id)).toEqual(["alpha"]) + }) +}) + +describe("whenHydrated", () => { + test("returns immediately when no hydration is in flight", async () => { + resetOverlay() + const started = Date.now() + await whenHydrated(SES, 5_000) + expect(Date.now() - started).toBeLessThan(1_000) + }) + + test("gives up on a hydration that never settles rather than blocking the prompt", async () => { + globalThis.fetch = (() => new Promise(() => {})) as unknown as typeof fetch + void hydrate("ses_stalled") + const started = Date.now() + await whenHydrated("ses_stalled", 150) + const elapsed = Date.now() - started + expect(elapsed).toBeGreaterThanOrEqual(100) + expect(elapsed).toBeLessThan(2_000) + }) +}) diff --git a/packages/opencode/test/memory/overlay-merge.test.ts b/packages/opencode/test/memory/overlay-merge.test.ts new file mode 100644 index 0000000000..8aa825b461 --- /dev/null +++ b/packages/opencode/test/memory/overlay-merge.test.ts @@ -0,0 +1,207 @@ +// altimate_change - new file +// Covers the injection-side merge of workspace memory into local memory: +// which blocks survive, and the guard that keeps a cloud-sourced training +// block from writing to the local store. +import { afterAll, afterEach, beforeEach, describe, expect, test } from "bun:test" +import { mkdirSync, mkdtempSync, rmSync, writeFileSync } from "node:fs" +import path from "node:path" +import os from "node:os" + +const ORIGINAL_DATA = process.env.XDG_DATA_HOME +const ORIGINAL_STATE = process.env.XDG_STATE_HOME +const ORIGINAL_FLAG = process.env.ALTIMATE_WORKSPACE +const SANDBOX = mkdtempSync(path.join(os.tmpdir(), `altimate-overlay-${process.pid}-`)) +mkdirSync(path.join(SANDBOX, "data"), { recursive: true }) +mkdirSync(path.join(SANDBOX, "state"), { recursive: true }) +process.env.XDG_DATA_HOME = path.join(SANDBOX, "data") +process.env.XDG_STATE_HOME = path.join(SANDBOX, "state") +process.env.ALTIMATE_WORKSPACE = "1" + +afterAll(() => { + if (ORIGINAL_DATA === undefined) delete process.env.XDG_DATA_HOME + else process.env.XDG_DATA_HOME = ORIGINAL_DATA + if (ORIGINAL_STATE === undefined) delete process.env.XDG_STATE_HOME + else process.env.XDG_STATE_HOME = ORIGINAL_STATE + if (ORIGINAL_FLAG === undefined) delete process.env.ALTIMATE_WORKSPACE + else process.env.ALTIMATE_WORKSPACE = ORIGINAL_FLAG + try { + rmSync(SANDBOX, { recursive: true, force: true }) + } catch { + /* best effort */ + } +}) + +const { MemoryPrompt } = await import("../../src/memory/prompt") +const { MIRROR_SOURCE } = await import("../../src/altimate/workspace/memory-api") +const { hydrate, resetOverlay, syncInternals } = await import( + "../../src/altimate/workspace/memory-sync" +) +const { TrainingStore } = await import("../../src/altimate/training/store") +const { Global } = await import("../../src/global") + +/** Resolved from Global rather than assumed: whether the sandbox env is picked + * up depends on when Global.Path is evaluated relative to module loading. */ +const GLOBAL_MEMORY_DIR = path.join(Global.Path.data, "memory") + +import { AltimateApi } from "../../src/altimate/api/client" + +const originalIsConfigured = AltimateApi.isConfigured +const originalGetCreds = AltimateApi.getCredentials +const originalFetch = globalThis.fetch +const originalIncrement = TrainingStore.incrementApplied + +const SES = "ses_overlay" +const NOW = "2026-08-19T00:00:00.000Z" +const BINDING = { + datamateId: 7, + datamateName: "acme", + repoRemote: "ssh://git@github.com/acme/analytics.git", + projectPath: "/work/analytics", + linkedAt: 1, +} + +let listResponse: any[] = [] +let incrementCalls: string[] = [] + +/** Writes a real block to the sandboxed global memory directory. */ +function writeLocalBlock(id: string, content: string, tags: string[] = []) { + const dir = path.join(GLOBAL_MEMORY_DIR, ...id.split("/").slice(0, -1)) + mkdirSync(dir, { recursive: true }) + const frontmatter = [ + "---", + `id: ${id}`, + "scope: global", + `created: ${NOW}`, + `updated: ${NOW}${tags.length ? `\ntags: ${JSON.stringify(tags)}` : ""}`, + "---", + "", + content, + "", + ].join("\n") + writeFileSync(path.join(GLOBAL_MEMORY_DIR, `${id}.md`), frontmatter) +} + +beforeEach(() => { + listResponse = [] + incrementCalls = [] + ;(AltimateApi as any).isConfigured = async () => true + ;(AltimateApi as any).getCredentials = async () => ({ + altimateInstanceName: "acme", + altimateUrl: "https://api.example.com", + altimateApiKey: "key", + }) + globalThis.fetch = (async (input: any) => { + const url = String(input) + const payload = url.includes("/datamates/memory/list") + ? listResponse + : url.includes("/datamates/") + ? { datamates: [{ id: 7, name: "acme", memory_enabled: true }] } + : { message: "ok" } + return new Response(JSON.stringify(payload), { + status: 200, + headers: { "Content-Type": "application/json" }, + }) + }) as typeof fetch + ;(TrainingStore as any).incrementApplied = async (_s: string, _k: string, name: string) => { + incrementCalls.push(name) + } + syncInternals.resolveBinding = async () => BINDING as any + resetOverlay() + MemoryPrompt.resetSession() +}) + +afterEach(() => { + globalThis.fetch = originalFetch + ;(AltimateApi as any).isConfigured = originalIsConfigured + ;(AltimateApi as any).getCredentials = originalGetCreds + ;(TrainingStore as any).incrementApplied = originalIncrement + delete syncInternals.resolveBinding + rmSync(GLOBAL_MEMORY_DIR, { recursive: true, force: true }) +}) + +const remote = (id: string, content: string, extra: Record = {}) => ({ + id: `rec-${id}`, + memory: content, + created_at: NOW, + updated_at: NOW, + metadata: { source: MIRROR_SOURCE, block_id: id, block_scope: "global", ...extra }, +}) + +describe("workspace memory in the injected prompt", () => { + test("a cloud block with no local counterpart is injected", async () => { + listResponse = [remote("warehouse/sizing", "ANALYTICS_WH must not be resized without asking.")] + await hydrate(SES) + const injected = await MemoryPrompt.inject(20000, { sessionID: SES }) + expect(injected).toContain("ANALYTICS_WH must not be resized") + }) + + test("the local copy wins when both hold the same block", async () => { + // A cloud record may have been edited by a client that does not preserve + // this CLI's metadata, so the file on disk is authoritative. + writeLocalBlock("warehouse/sizing", "LOCAL: ask the data team before resizing.") + listResponse = [remote("warehouse/sizing", "REMOTE: stale copy of the same block.")] + await hydrate(SES) + const injected = await MemoryPrompt.inject(20000, { sessionID: SES }) + expect(injected).toContain("LOCAL: ask the data team") + expect(injected).not.toContain("REMOTE: stale copy") + }) + + test("a sibling project's block survives an id collision and is labelled", async () => { + // Block ids are unique only within a project directory, so the same id in + // two projects means two different blocks — not a duplicate. + writeLocalBlock("warehouse/sizing", "LOCAL: this project's own note.") + listResponse = [ + remote("warehouse/sizing", "SIBLING: another project's note.", { + block_scope: "project", + datamate_id: "7", + repo_remote: "ssh://git@github.com/acme/other-repo.git", + }), + ] + await hydrate(SES) + const injected = await MemoryPrompt.inject(20000, { sessionID: SES }) + expect(injected).toContain("LOCAL: this project's own note") + expect(injected).toContain("SIBLING: another project's note") + expect(injected).toContain("other-repo") + }) + + test("no overlay is merged when the caller has no session", async () => { + listResponse = [remote("warehouse/sizing", "REMOTE ONLY")] + await hydrate(SES) + const injected = await MemoryPrompt.inject(20000, {}) + expect(injected).not.toContain("REMOTE ONLY") + }) + + test("a sibling project's training entry is labelled with its origin", async () => { + // mergeOverlay deliberately keeps both when a sibling shares an id with + // this project's block, so an unlabelled entry leaves the model with two + // identical headings it cannot tell apart. formatBlock labels these; + // formatTrainingEntry has to as well. + listResponse = [ + remote("training/rule/shared", "Partition by event_date.", { + block_scope: "project", + datamate_id: "7", + repo_remote: "ssh://git@github.com/acme/other.git", + block_tags: '["training","rule"]', + }), + ] + await hydrate(SES) + const injected = await MemoryPrompt.inject(20000, { sessionID: SES }) + expect(injected).toContain("Partition by event_date.") + expect(injected).toContain("from workspace project") + }) + + test("a remote training block never drives the local applied counter", async () => { + // incrementApplied writes to the LOCAL store; firing it for a block this + // machine never had would fabricate a file from the cloud copy. + writeLocalBlock("training/rule/local-one", "Always alias CTEs.", ["training", "rule"]) + listResponse = [ + remote("training/rule/remote-one", "Never use FLOAT for money.", { + block_tags: "training,rule", + }), + ] + await hydrate(SES) + await MemoryPrompt.inject(20000, { sessionID: SES }) + expect(incrementCalls).toContain("local-one") + expect(incrementCalls).not.toContain("remote-one") + }) +}) diff --git a/packages/opencode/test/memory/store-directory.test.ts b/packages/opencode/test/memory/store-directory.test.ts new file mode 100644 index 0000000000..25b18f074f --- /dev/null +++ b/packages/opencode/test/memory/store-directory.test.ts @@ -0,0 +1,54 @@ +import { describe, test, expect, beforeEach, afterEach } from "bun:test" +import fs from "fs/promises" +import path from "path" +import os from "os" +import { MemoryStore } from "@/memory/store" + +// Exercises the REAL store, unlike store.test.ts which re-implements its logic +// and so cannot catch path-resolution bugs. Callers outside an Instance context +// -- the `link` subcommand is one -- pass `directory` explicitly; every step of +// the read path has to honour it, not just the directory scan. +describe("MemoryStore project scope with an explicit directory", () => { + let proj: string + + beforeEach(async () => { + proj = await fs.mkdtemp(path.join(os.tmpdir(), "store-dir-")) + const dir = path.join(proj, ".altimate-code", "memory") + await fs.mkdir(dir, { recursive: true }) + const now = new Date().toISOString() + await fs.writeFile( + path.join(dir, "proj-block.md"), + ["---", "id: proj-block", "scope: project", `created: ${now}`, `updated: ${now}`, "---", "", "A project fact.", ""].join("\n"), + ) + }) + + afterEach(async () => { + await fs.rm(proj, { recursive: true, force: true }) + }) + + test("list reads the blocks it scanned, not the ambient directory's", async () => { + // Regression: `list` scanned `directory` but `read` re-resolved the path + // from the ambient instance, so every block it found came back undefined. + const blocks = await MemoryStore.list("project", { directory: proj }) + expect(blocks.map((b) => b.id)).toEqual(["proj-block"]) + expect(blocks[0].content).toBe("A project fact.") + }) + + test("read honours an explicit directory", async () => { + const block = await MemoryStore.read("project", "proj-block", proj) + expect(block?.content).toBe("A project fact.") + }) + + test("listAll surfaces project blocks with no instance context", async () => { + const blocks = await MemoryStore.listAll({ directory: proj }) + expect(blocks.some((b) => b.id === "proj-block")).toBe(true) + }) + + test("listAll still returns global blocks when project scope cannot resolve", async () => { + // No directory and no instance: project scope throws. It must not take + // global memory down with it. + const blocks = await MemoryStore.listAll() + expect(Array.isArray(blocks)).toBe(true) + expect(blocks.every((b) => b.scope === "global")).toBe(true) + }) +})