From 88f267c863fbff19871f76dcf7ad670a90351c92 Mon Sep 17 00:00:00 2001 From: Sarav Date: Thu, 8 Oct 2026 13:31:02 +0530 Subject: [PATCH 1/4] feat(workspace): record workspace sync state and publish changes to clients Every client needed to say when workspace skills and memory were last checked and what changed, but only the TUI sidebar could see part of it ("skills synced Xm ago"), and an IDE extension running `serve` saw nothing. - Add `altimate/workspace/sync-state.ts`: one record per synced entity kind per project (last checked, last changed, count, status, what changed), persisted under the state dir so the TUI main thread and the server worker share it. Diffs are generic over `{id: {label, version}}`, so a new kind records the same metadata. - Skill sync and memory loads record into it from their existing paths; no new timers. The first check is a baseline, not a change. - Publish `altimate.workspace.sync.changed` on the bus, only when a check found a difference or an error state changed. - Add `GET /altimate/workspace/status`, returning the recorded state for the request's workspace. - TUI sidebar shows "memory loaded Xm ago" from the same state. Co-Authored-By: Claude Opus 5.5 (1M context) --- .../opencode/src/altimate/workspace/manage.ts | 7 + .../src/altimate/workspace/memory-sync.ts | 35 ++- .../src/altimate/workspace/skill-sync.ts | 65 +++++- .../src/altimate/workspace/sync-state.ts | 212 ++++++++++++++++++ .../plugin/tui/altimate/workspace-sidebar.tsx | 5 + packages/opencode/src/server/server.ts | 28 +++ .../altimate/workspace/memory-sync.test.ts | 45 ++++ .../altimate/workspace/skill-sync.test.ts | 54 +++++ .../altimate/workspace/sync-state.test.ts | 152 +++++++++++++ .../server/altimate-workspace-routes.test.ts | 41 ++++ 10 files changed, 637 insertions(+), 7 deletions(-) create mode 100644 packages/opencode/src/altimate/workspace/sync-state.ts create mode 100644 packages/opencode/test/altimate/workspace/sync-state.test.ts diff --git a/packages/opencode/src/altimate/workspace/manage.ts b/packages/opencode/src/altimate/workspace/manage.ts index 7ebe13c9db..342c58c21e 100644 --- a/packages/opencode/src/altimate/workspace/manage.ts +++ b/packages/opencode/src/altimate/workspace/manage.ts @@ -27,6 +27,7 @@ import { WorkspaceApi, type ProjectIdentifier } from "./api-client" import { resolveProjectIdentifier } from "./detect" import * as MemorySync from "./memory-sync" import * as SkillSync from "./skill-sync" +import * as SyncState from "./sync-state" import { clearLocalBinding, currentScope, @@ -59,6 +60,11 @@ export interface StatusReport { * per-process, so a fresh session has not synced yet even for a project whose * snapshot is current on disk. Callers must not render it as "never synced". */ skillsSyncedAt: number | null + // altimate_change start — workspace sync state + /** When this project's workspace memory was last loaded, from the shared sync state, or + * null when no load has been recorded for this binding. */ + memoryLoadedAt?: number | null + // altimate_change end } export interface RefreshReport { @@ -156,6 +162,7 @@ export async function status( ...(memory === "unreadable" ? { memoryUnreadable: true as const } : {}), skillsEnabled: SkillSync.isEnabled(), skillsSyncedAt: await skillsSyncedAt(directory, binding), + memoryLoadedAt: binding ? (SyncState.read(directory, binding.datamateId)?.entities.memory?.lastCheckedAt ?? null) : null, } } diff --git a/packages/opencode/src/altimate/workspace/memory-sync.ts b/packages/opencode/src/altimate/workspace/memory-sync.ts index 85e115146b..a8742f4e75 100644 --- a/packages/opencode/src/altimate/workspace/memory-sync.ts +++ b/packages/opencode/src/altimate/workspace/memory-sync.ts @@ -1376,7 +1376,7 @@ type LoadOutcome = ( | { status: "unlinked" } | { status: "disabled" } | { status: "error" } -) & { epoch?: string; dir?: string | null } +) & { epoch?: string; dir?: string | null; /** The workspace the load read, once resolved. */ datamateId?: number } /** Read this project's workspace memory. Pure: it publishes nothing, so a slow * load that has been superseded cannot write over a newer result. */ @@ -1403,9 +1403,10 @@ async function loadWorkspaceMemory(directory?: string): Promise { vouched = epoch if (!stable) return { status: "error", epoch, dir } if (!binding) return { status: "unlinked", epoch, dir } + const datamateId = binding.datamateId const enabled = await memoryStatus(binding) - if (enabled === "error") return { status: "error", epoch, dir } - if (enabled === "disabled") return { status: "disabled", epoch, dir } + if (enabled === "error") return { status: "error", epoch, dir, datamateId } + if (enabled === "disabled") return { status: "disabled", epoch, dir, datamateId } const ownProjectKey = projectKeyFor(binding) const ownWorkspace = String(binding.datamateId) @@ -1441,7 +1442,7 @@ async function loadWorkspaceMemory(directory?: string): Promise { if (block.expires && new Date(block.expires) <= new Date()) continue blocks.push(block) } - return { status: "loaded", blocks, epoch, dir } + return { status: "loaded", blocks, epoch, dir, datamateId } } catch (err) { log.warn("workspace memory load failed", { err: String(err) }) // Stamped like any other outcome: without an epoch the session would reload (and make @@ -1461,6 +1462,9 @@ function commitLoad(sessionID: string, state: SessionMemory, outcome: LoadOutcom if (outcome.epoch !== epochFor(outcome.dir ?? null)) return state.loadedEpoch = outcome.epoch state.dir = outcome.dir ?? null + // altimate_change start — workspace sync state + recordSyncState(outcome) + // altimate_change end // An error keeps whatever the session had and is not retried every turn (as before). if (outcome.status === "error") return state.overlay = outcome.status === "loaded" ? outcome.blocks : [] @@ -1469,6 +1473,29 @@ function commitLoad(sessionID: string, state: SessionMemory, outcome: LoadOutcom } } +// altimate_change start — workspace sync state +/** Record a committed load in the workspace sync state. Memory is loaded per session but + * recorded per project, so a second session reading the same blocks is not a change. + * Detached: the load is already committed and nothing waits on status metadata. A load + * that could not resolve a workspace, or found memory off, records nothing. */ +function recordSyncState(outcome: LoadOutcome): void { + if (!outcome.dir || outcome.datamateId === undefined) return + if (outcome.status !== "loaded" && outcome.status !== "error") return + const { dir, datamateId } = outcome + const items = + outcome.status === "loaded" + ? Object.fromEntries( + outcome.blocks.map((block) => [block.id, { label: blockTitle(block), version: block.updated ?? "" }]), + ) + : undefined + void import("./sync-state") + .then((SyncState) => + SyncState.record(dir, datamateId, "memory", items ? { items } : { error: "could not load workspace memory" }), + ) + .catch((err) => log.warn("could not record the memory sync state", { err: String(err) })) +} +// altimate_change end + /** A session's cloud overlay. Returns a copy so a caller cannot mutate the * cached state in place. */ export function overlayBlocks(sessionID: string): RemoteMemoryBlock[] { diff --git a/packages/opencode/src/altimate/workspace/skill-sync.ts b/packages/opencode/src/altimate/workspace/skill-sync.ts index 5b50802040..7214947352 100644 --- a/packages/opencode/src/altimate/workspace/skill-sync.ts +++ b/packages/opencode/src/altimate/workspace/skill-sync.ts @@ -101,6 +101,8 @@ export interface Manifest { interface RemoteSummary { publicId: string updatedAt: string + /** Display name, for the sync-state change list. Optional: sync never depends on it. */ + name?: string } /** ``CustomSkillDetail.files`` — ``CustomSkillFileMeta`` is ``{path, size}``. */ @@ -244,6 +246,39 @@ export function describeSyncProblems(result: SyncResult): { title: string; messa return { title: `${n} workspace skill${n === 1 ? "" : "s"} skipped`, message: lines.join("\n") } } +// altimate_change start — workspace sync state +/** Record this run in the workspace sync state. `rows` is the list the published snapshot + * describes, or null when the run left the snapshot as it was — the items are then + * unknown to this run and the previous record stands. Imported lazily: the state module + * reaches the bus, which this module must not load on the opted-out path. */ +async function recordSyncState( + directory: string, + datamateId: number, + result: SyncResult, + rows: RemoteSummary[] | null, + dropped: Set, +): Promise { + const problem = describeSyncProblems(result) + const items = rows + ? Object.fromEntries( + rows + .filter((row) => !dropped.has(row.publicId)) + .map((row) => [row.publicId, { label: displayId(row.name ?? row.publicId), version: row.updatedAt }]), + ) + : undefined + try { + const SyncState = await import("./sync-state") + await SyncState.record(directory, datamateId, "skills", { + ...(items ? { items } : {}), + ...(problem ? { error: `${problem.title}: ${problem.message.split("\n").join("; ")}` } : {}), + }) + } catch (err) { + // Status metadata: never the reason a sync fails. + log.warn("could not record the skill sync state", { err: String(err) }) + } +} +// altimate_change end + /** A fixed, user-facing reason for a skill that failed to sync. The raw error * can carry request URLs, server text or local paths — diagnostics for the * log, not for a toast. */ @@ -746,10 +781,14 @@ function parsePage( const rows: RemoteSummary[] = [] for (const row of p.items) { if (!row || typeof row !== "object") return null - const r = row as { public_id?: unknown; updated_at?: unknown } + const r = row as { public_id?: unknown; updated_at?: unknown; name?: unknown } if (typeof r.public_id !== "string" || !r.public_id) return null if (typeof r.updated_at !== "string" || !r.updated_at) return null - rows.push({ publicId: r.public_id, updatedAt: r.updated_at }) + rows.push({ + publicId: r.public_id, + updatedAt: r.updated_at, + ...(typeof r.name === "string" && r.name ? { name: r.name } : {}), + }) } return { rows, pages: rawPages, total } } @@ -1109,6 +1148,15 @@ export async function syncSkills(directory: string): Promise { // Set once the workspace's list has actually been read. Only then has this // project been "checked", and only then should the poll interval start. let sawRemote = false + // altimate_change start — workspace sync state + // What the sync-state record needs: the workspace checked, the list it read, the skills + // that list named but this run could not install (and had no previous copy of), and + // whether the run left a snapshot describing that list. + let checkedDatamateId: number | undefined + let remoteRows: RemoteSummary[] | null = null + const dropped = new Set() + let published = false + // altimate_change end const run = (async () => { // Checked BEFORE the binding: `resolveBinding` needs credentials too, so a // disconnected client would otherwise return on a null binding and never @@ -1212,6 +1260,7 @@ export async function syncSkills(directory: string): Promise { return } const binding = outcome.binding + checkedDatamateId = binding.datamateId // Refuse to touch a directory we did not create. Everything below either // deletes this tree or replaces it wholesale, so without this a user's own @@ -1279,6 +1328,7 @@ export async function syncSkills(directory: string): Promise { return } sawRemote = true + remoteRows = remote syncedFor.set(canon, accountKeyOf(creds.altimateInstanceName, creds.altimateUrl, account)) // The workspace has skills now, even if installing them fails below. if (remote.length > 0) await clearEmptyRecordFor(canon, currentEmptyKey) @@ -1289,6 +1339,7 @@ export async function syncSkills(directory: string): Promise { // live by then — another process can swap a partial snapshot in // between, and the marker would vouch for a sync this run never made. validated = manifest + published = true return } @@ -1302,6 +1353,7 @@ export async function syncSkills(directory: string): Promise { if (written && (after?.status !== "bound" || after.binding.datamateId !== binding.datamateId)) await withdrawEmptyRecord(canon, written) changed = true + published = true log.info("workspace has no custom skills; removed the local snapshot") return } @@ -1354,6 +1406,7 @@ export async function syncSkills(directory: string): Promise { failed = true log.warn("skipping a workspace skill with an unusable id", { skill: summary.publicId }) skippedSkills.push({ skill: displayId(summary.publicId), reason: "its id is not usable as a folder name" }) + dropped.add(summary.publicId) continue } try { @@ -1435,6 +1488,7 @@ export async function syncSkills(directory: string): Promise { } if (!carried) { await fs.rm(path.join(staging, summary.publicId), { recursive: true, force: true }).catch(() => {}) + dropped.add(summary.publicId) } log.warn("skipping a workspace skill; the rest of the snapshot still publishes", { skill: summary.publicId, @@ -1516,6 +1570,7 @@ export async function syncSkills(directory: string): Promise { await fs.rm(retired, { recursive: true, force: true }).catch(() => {}) await clearEmptyRecordFor(canon, currentEmptyKey) changed = true + published = true log.info("workspace skills synced", { datamateId: binding.datamateId, skills: remote.length, @@ -1566,7 +1621,11 @@ export async function syncSkills(directory: string): Promise { // too — otherwise a problem it fixed would stay latched, and its return // would never be announced. if (skippedSkills.length === 0 && !syncError) store.announced.delete(canon) - return { changed, skipped: skippedSkills, error: syncError } + const result: SyncResult = { changed, skipped: skippedSkills, error: syncError } + // altimate_change start — workspace sync state + if (checkedDatamateId !== undefined) await recordSyncState(canon, checkedDatamateId, result, published ? remoteRows : null, dropped) + // altimate_change end + return result })() inFlight.set(canon, settled) try { diff --git a/packages/opencode/src/altimate/workspace/sync-state.ts b/packages/opencode/src/altimate/workspace/sync-state.ts new file mode 100644 index 0000000000..8b5c8fa7f9 --- /dev/null +++ b/packages/opencode/src/altimate/workspace/sync-state.ts @@ -0,0 +1,212 @@ +// altimate_change start — workspace sync state +// +// One record per synced entity kind (skills, memory, …), kept per project, so every client +// can say when each kind was last checked, when it last changed and what changed — the +// TUI sidebar, and the IDE extension through `GET /altimate/workspace/status` and the +// `altimate.workspace.sync.changed` event. +// +// Written by the existing sync paths only; nothing here polls. A new entity kind records +// its items through `record` and gets the same metadata and change detection. +// +// Persisted rather than held in memory: in the TUI the per-turn syncs run in the server +// worker while the sidebar renders on the main thread, and the two share no module state +// (see `SkillSync.lastSuccessfulSyncAt`). A file is what both sides see. +import { createHash } from "node:crypto" +import { readFileSync } from "node:fs" +import path from "node:path" +import z from "zod" +import { Bus } from "@/bus" +import { BusEvent } from "@/bus/bus-event" +import { Global } from "@/global" +import { Log } from "@/altimate/util/log" +import { Filesystem } from "@/util/filesystem" +import { canonicalDirectory } from "./state" + +const log = Log.create({ service: "altimate-workspace-sync-state" }) + +export const KINDS = ["skills", "memory"] as const +export type SyncKind = (typeof KINDS)[number] + +/** What a kind holds: a stable id mapped to a display label and a version that moves when + * the item's content does. Diffs are computed from this alone, so every kind compares the + * same way. */ +export type SyncItems = Record + +const Changes = z.object({ + added: z.array(z.string()), + removed: z.array(z.string()), + updated: z.array(z.string()), +}) +export type SyncChanges = z.infer + +const EntityState = z.object({ + kind: z.enum(KINDS), + status: z.enum(["ok", "error"]), + /** Items held after the last check; null when it has never been read successfully. */ + count: z.number().int().nonnegative().nullable(), + /** Epoch ms of the last completed check, successful or not. */ + lastCheckedAt: z.number(), + /** Epoch ms of the last check that found a difference; null when none has. */ + lastChangedAt: z.number().nullable(), + /** What that last change was, by label. Kept until the next change replaces it. */ + changes: Changes.nullable(), + /** User-facing reason, set while `status` is "error". */ + error: z.string().optional(), +}) +export type EntityState = z.infer + +export const Event = { + /** Published only when a check found a difference, or when a kind's error state changed + * (a new problem, a different problem, or recovery) — never for a check that found + * nothing, so a client can surface every event without being noisy. */ + Changed: BusEvent.define( + "altimate.workspace.sync.changed", + z.object({ + directory: z.string(), + datamateId: z.number(), + kind: z.enum(KINDS), + state: EntityState, + /** This check's own difference; null for an error-state-only event. */ + changes: Changes.nullable(), + }), + ), +} + +export interface WorkspaceSyncState { + datamateId: number + entities: Partial> +} + +interface StoredEntity extends EntityState { + /** Absent until a check has read the items, so the first read is a baseline. */ + items?: SyncItems +} + +interface StoredFile { + version: 1 + directory: string + datamateId: number + entities: Partial> +} + +function filePath(directory: string): string { + const key = createHash("sha256").update(canonicalDirectory(directory)).digest("hex").slice(0, 32) + return path.join(Global.Path.state, "altimate-workspace-sync", `${key}.json`) +} + +function readStored(directory: string): StoredFile | null { + try { + const parsed = JSON.parse(readFileSync(filePath(directory), "utf8")) as StoredFile + if (parsed?.version !== 1 || typeof parsed.datamateId !== "number" || !parsed.entities) return null + // A hash collision or a moved file must not attribute another tree's state to this one. + if (parsed.directory !== canonicalDirectory(directory)) return null + return parsed + } catch { + return null + } +} + +/** The project's sync state for the workspace it is bound to now. Null when nothing has + * been recorded for that workspace — after a rebind the previous workspace's state is not + * this one's. */ +export function read(directory: string, datamateId: number): WorkspaceSyncState | null { + const stored = readStored(directory) + if (!stored || stored.datamateId !== datamateId) return null + const entities: WorkspaceSyncState["entities"] = {} + for (const kind of KINDS) { + const entity = stored.entities[kind] + if (!entity) continue + const { items: _items, ...state } = entity + entities[kind] = state + } + return { datamateId, entities } +} + +/** Added, removed and updated labels between two item sets. Pure. */ +export function diff(previous: SyncItems, next: SyncItems): SyncChanges { + const added: string[] = [] + const updated: string[] = [] + for (const [id, item] of Object.entries(next)) { + const before = previous[id] + if (!before) added.push(item.label) + else if (before.version !== item.version) updated.push(item.label) + } + const removed = Object.entries(previous) + .filter(([id]) => !(id in next)) + .map(([, item]) => item.label) + return { added: added.sort(), removed: removed.sort(), updated: updated.sort() } +} + +function isEmpty(changes: SyncChanges): boolean { + return changes.added.length === 0 && changes.removed.length === 0 && changes.updated.length === 0 +} + +/** A completed check. `items` is what the check read, absent when it could not read them + * (the previous set is kept); `error` is a problem to report, which a partial read can + * carry alongside the items it did get. */ +export interface Outcome { + items?: SyncItems + error?: string +} + +/** Writes are chained per file so two kinds settling at once cannot drop each other's update. */ +const writes = new Map>() + +/** Record a completed check of one kind and publish when it changed something. + * + * Never throws: this is status metadata, and a failure here must not fail the sync that + * reported it. Resolves once the state is written and any event published. */ +export function record(directory: string, datamateId: number, kind: SyncKind, outcome: Outcome): Promise { + const file = filePath(directory) + const run = (writes.get(file) ?? Promise.resolve()).then(() => apply(directory, datamateId, kind, outcome)) + const settled = run.catch((err) => log.warn("could not record workspace sync state", { kind, err: String(err) })) + writes.set(file, settled) + void settled.finally(() => { + if (writes.get(file) === settled) writes.delete(file) + }) + return settled +} + +async function apply(directory: string, datamateId: number, kind: SyncKind, outcome: Outcome): Promise { + const canon = canonicalDirectory(directory) + const stored = readStored(directory) + // Another workspace's state describes nothing about this one: start over. + const file: StoredFile = + stored && stored.datamateId === datamateId ? stored : { version: 1, directory: canon, datamateId, entities: {} } + const previous = file.entities[kind] + const now = Date.now() + + let changes: SyncChanges | null = null + if (outcome.items) { + // The first check this project has ever recorded is a baseline, not a change: reporting + // every existing item as "added" would announce the whole workspace on first use. + const found = previous?.items ? diff(previous.items, outcome.items) : null + changes = found && !isEmpty(found) ? found : null + } + // A check that could not read the items keeps what was last known: a failure says + // nothing about them. + const items = outcome.items ?? previous?.items + const next: StoredEntity = { + kind, + status: outcome.error ? "error" : "ok", + count: items ? Object.keys(items).length : null, + lastCheckedAt: now, + lastChangedAt: changes ? now : (previous?.lastChangedAt ?? null), + changes: changes ?? previous?.changes ?? null, + ...(outcome.error ? { error: outcome.error } : {}), + ...(items ? { items } : {}), + } + file.entities[kind] = next + Filesystem.writeJsonAtomic(filePath(directory), file) + + const errorChanged = (previous?.status ?? "ok") !== next.status || previous?.error !== next.error + if (!changes && !errorChanged) return + const { items: _items, ...state } = next + // Needs an instance (a turn, a server route). A sync started outside one — a bind from + // the CLI — has no client listening anyway; the state above is what it leaves behind. + await Bus.publish(Event.Changed, { directory: canon, datamateId, kind, state, changes }).catch((err) => + log.info("workspace sync change not published", { kind, err: String(err) }), + ) +} + +// altimate_change end diff --git a/packages/opencode/src/plugin/tui/altimate/workspace-sidebar.tsx b/packages/opencode/src/plugin/tui/altimate/workspace-sidebar.tsx index 4e8400bf1f..7f06b74f6f 100644 --- a/packages/opencode/src/plugin/tui/altimate/workspace-sidebar.tsx +++ b/packages/opencode/src/plugin/tui/altimate/workspace-sidebar.tsx @@ -307,6 +307,11 @@ function View(props: { api: TuiPluginApi }) { {(at) => {`skills synced ${describeAge(at())}`}} + {/* altimate_change start — workspace sync state */} + + {(at) => {`memory loaded ${describeAge(at())}`}} + + {/* altimate_change end */} {/* altimate_change end */} {/* ``pinned via --workspace`` means "this SESSION was launched * with --workspace and it resolved to this id". It does NOT diff --git a/packages/opencode/src/server/server.ts b/packages/opencode/src/server/server.ts index 19d3fb696e..6882985419 100644 --- a/packages/opencode/src/server/server.ts +++ b/packages/opencode/src/server/server.ts @@ -1141,6 +1141,34 @@ export namespace Server { } }) // altimate_change end + // altimate_change start — GET /altimate/workspace/status + // The recorded sync state for the request's directory: per entity kind, when it was last + // checked and changed, and what changed. Read-only and network-free past the binding + // resolution (memoized), so a client may call it whenever it renders. Changes after this + // read arrive as `altimate.workspace.sync.changed` on `/event`. + .get("/altimate/workspace/status", async (c) => { + const refused = workspaceRouteRefusal( + c.req.header("origin"), + c.req.header("host"), + undefined, + c.req.header("sec-fetch-site"), + ) + if (refused) return c.json(refused.body, refused.status) + try { + const { resolveBindingOutcome } = await import("../altimate/workspace/state") + const SyncState = await import("../altimate/workspace/sync-state") + const outcome = await resolveBindingOutcome(Instance.directory) + if (outcome.status !== "bound") return c.json({ ok: true as const, datamateId: null, entities: {} }) + const datamateId = outcome.binding.datamateId + const state = SyncState.read(Instance.directory, datamateId) + return c.json({ ok: true as const, datamateId, entities: state?.entities ?? {} }) + } catch (err) { + const error = err instanceof Error ? err.message : String(err) + log.error("workspace status: failed", { error }) + return c.json({ ok: false, error }, 500) + } + }) + // altimate_change end // altimate_change start — GET /altimate/skill/publishable, POST /altimate/skill/publish // The CLI's `skill publish ` for the IDE extension. Only `serve` holds the extension's // pin, so publishing here targets the workspace selected in the panel — through the same diff --git a/packages/opencode/test/altimate/workspace/memory-sync.test.ts b/packages/opencode/test/altimate/workspace/memory-sync.test.ts index eb37404247..3b2b0d1a91 100644 --- a/packages/opencode/test/altimate/workspace/memory-sync.test.ts +++ b/packages/opencode/test/altimate/workspace/memory-sync.test.ts @@ -2053,3 +2053,48 @@ describe("binding resolution for the mirror", () => { expect(posts.length).toBe(0) }) }) + +// altimate_change start — workspace sync state +describe("sync state", () => { + const note = (blockId: string, title: string, updated = "2026-01-01T00:00:00.000Z") => ({ + id: `mem-${blockId}`, + memory: `# ${title}\nbody`, + metadata: { source: MIRROR_SOURCE, block_id: blockId, block_scope: "global", block_updated: updated }, + }) + + /** The record is written detached from the load, so poll for it. */ + async function memoryState(dir: string, predicate: (e: any) => boolean = () => true) { + const SyncState = await import("../../../src/altimate/workspace/sync-state") + for (let i = 0; i < 50; i++) { + const entity = SyncState.read(dir, BINDING.datamateId)?.entities.memory + if (entity && predicate(entity)) return entity + await new Promise((r) => setTimeout(r, 10)) + } + return SyncState.read(dir, BINDING.datamateId)?.entities.memory + } + + test("loads are recorded per project; a later load records what changed, by title", async () => { + const dir = mkdtempSync(path.join(os.tmpdir(), "memory-sync-state-")) + listResponse = [note("a", "Naming conventions"), note("b", "Warehouse sizing")] + await refresh(`${SES}-state-1`, dir) + expect(await memoryState(dir)).toMatchObject({ kind: "memory", status: "ok", count: 2, changes: null }) + + // Another session reading the same blocks is not a change. + await refresh(`${SES}-state-2`, dir) + expect((await memoryState(dir))?.changes).toBeNull() + + listResponse = [note("a", "Naming conventions", "2026-02-01T00:00:00.000Z"), note("c", "Cost alerts")] + await refresh(`${SES}-state-3`, dir) + const entity = await memoryState(dir, (e) => e.changes !== null) + expect(entity?.changes).toEqual({ added: ["Cost alerts"], removed: ["Warehouse sizing"], updated: ["Naming conventions"] }) + expect(entity?.count).toBe(2) + }) + + test("a load for an unbound project records nothing", async () => { + const dir = mkdtempSync(path.join(os.tmpdir(), "memory-sync-state-off-")) + syncInternals.resolveBinding = async () => null + await refresh(`${SES}-state-off`, dir) + expect(await memoryState(dir)).toBeUndefined() + }) +}) +// altimate_change end diff --git a/packages/opencode/test/altimate/workspace/skill-sync.test.ts b/packages/opencode/test/altimate/workspace/skill-sync.test.ts index a596681308..762751e24b 100644 --- a/packages/opencode/test/altimate/workspace/skill-sync.test.ts +++ b/packages/opencode/test/altimate/workspace/skill-sync.test.ts @@ -2545,3 +2545,57 @@ describe("shouldAnnounce", () => { expect(shouldAnnounce("/tmp/other", problem)).toBe(true) }) }) + +// altimate_change start — workspace sync state +describe("sync state", () => { + const read = async () => (await import("@/altimate/workspace/sync-state")).read(project, 1)?.entities.skills + + test("the first sync is a baseline and a later one records what changed, by name", async () => { + serve({ "pub-1": { "SKILL.md": "one" } }) + await syncSkills(project) + expect(await read()).toMatchObject({ status: "ok", count: 1, changes: null, lastChangedAt: null }) + + serve({ "pub-1": { "SKILL.md": "one" }, "pub-2": { "SKILL.md": "two" } }) + await syncSkills(project) + const entity = await read() + expect(entity).toMatchObject({ status: "ok", count: 2, changes: { added: ["pub-2"], removed: [], updated: [] } }) + expect(entity?.lastChangedAt).toBeNumber() + }) + + test("an edited skill is recorded as updated", async () => { + serve({ "pub-1": { "SKILL.md": "one" } }, "2026-01-01T00:00:00Z") + await syncSkills(project) + serve({ "pub-1": { "SKILL.md": "one, edited" } }, "2026-02-01T00:00:00Z") + await syncSkills(project) + expect((await read())?.changes).toEqual({ added: [], removed: [], updated: ["pub-1"] }) + }) + + test("a failed list is an error that keeps the last known count", async () => { + serve({ "pub-1": { "SKILL.md": "one" } }) + await syncSkills(project) + globalThis.fetch = (async () => { + throw new Error("offline") + }) as unknown as typeof fetch + await syncSkills(project) + expect(await read()).toMatchObject({ status: "error", count: 1, error: expect.stringContaining("skill list") }) + }) + + test("a skill that could not be installed is not counted, and the problem is recorded", async () => { + serve({ "pub-1": { "SKILL.md": "one" } }) + await syncSkills(project) + + serve({ "pub-1": { "SKILL.md": "one" }, "pub-2": { "SKILL.md": "two" } }, "2026-03-01T00:00:00Z") + const inner = globalThis.fetch + globalThis.fetch = (async (input: string | URL, init?: RequestInit) => { + if (String(input).includes("/pub-2")) throw new Error("offline") + return inner(input as never, init as never) + }) as unknown as typeof fetch + await syncSkills(project) + + const entity = await read() + expect(entity).toMatchObject({ status: "error", count: 1, error: expect.stringContaining("pub-2") }) + // pub-1 moved to a new version; pub-2 never arrived, so it is not reported as added. + expect(entity?.changes).toEqual({ added: [], removed: [], updated: ["pub-1"] }) + }) +}) +// altimate_change end diff --git a/packages/opencode/test/altimate/workspace/sync-state.test.ts b/packages/opencode/test/altimate/workspace/sync-state.test.ts new file mode 100644 index 0000000000..61476d03a5 --- /dev/null +++ b/packages/opencode/test/altimate/workspace/sync-state.test.ts @@ -0,0 +1,152 @@ +// altimate_change - new file +// +// Unit coverage for the workspace sync state (src/altimate/workspace/sync-state.ts): +// what a check records, what counts as a change, and when the change event is published. +// The state file is real, in a sandboxed state directory; only the bus is spied, because a +// publish needs an instance the unit under test does not own. +import { afterAll, afterEach, beforeEach, describe, expect, spyOn, test } from "bun:test" +import { mkdirSync, mkdtempSync, realpathSync, rmSync } from "node:fs" +import os from "node:os" +import path from "node:path" + +// Global.Path.state resolves at module load, so the sandbox must exist first. +const ORIGINAL_XDG_STATE_HOME = process.env.XDG_STATE_HOME +const SANDBOX = path.join(os.tmpdir(), `altimate-sync-state-${process.pid}-${Date.now()}`) +mkdirSync(path.join(SANDBOX, "state"), { recursive: true }) +process.env.XDG_STATE_HOME = path.join(SANDBOX, "state") + +afterAll(() => { + if (ORIGINAL_XDG_STATE_HOME === undefined) delete process.env.XDG_STATE_HOME + else process.env.XDG_STATE_HOME = ORIGINAL_XDG_STATE_HOME + rmSync(SANDBOX, { recursive: true, force: true }) +}) + +const { Bus } = await import("../../../src/bus") +const SyncState = await import("../../../src/altimate/workspace/sync-state") + +let project: string +let published: Array<{ type: string; properties: any }> +let publish: ReturnType + +beforeEach(() => { + project = realpathSync(mkdtempSync(path.join(SANDBOX, "project-"))) + published = [] + publish = spyOn(Bus, "publish").mockImplementation((async (def: { type: string }, properties: unknown) => { + published.push({ type: def.type, properties }) + }) as never) +}) + +afterEach(() => { + publish.mockRestore() +}) + +const skill = (label: string, version = "v1") => ({ label, version }) + +describe("diff", () => { + test("reports added, removed and updated labels, sorted", () => { + const changes = SyncState.diff( + { a: skill("Alpha"), b: skill("Beta"), c: skill("Gamma") }, + { a: skill("Alpha"), b: skill("Beta", "v2"), d: skill("Delta") }, + ) + expect(changes).toEqual({ added: ["Delta"], removed: ["Gamma"], updated: ["Beta"] }) + }) + + test("identical sets have no changes", () => { + expect(SyncState.diff({ a: skill("Alpha") }, { a: skill("Alpha") })).toEqual({ added: [], removed: [], updated: [] }) + }) +}) + +describe("record", () => { + test("the first check is a baseline: recorded, not announced", async () => { + await SyncState.record(project, 7, "skills", { items: { a: skill("Alpha"), b: skill("Beta") } }) + + const state = SyncState.read(project, 7) + expect(state?.entities.skills).toMatchObject({ kind: "skills", status: "ok", count: 2, lastChangedAt: null, changes: null }) + expect(state?.entities.skills?.lastCheckedAt).toBeNumber() + expect(published).toEqual([]) + }) + + test("a later check that finds a difference records and publishes it", async () => { + await SyncState.record(project, 7, "skills", { items: { a: skill("Alpha") } }) + await SyncState.record(project, 7, "skills", { items: { a: skill("Alpha", "v2"), b: skill("Beta") } }) + + const entity = SyncState.read(project, 7)?.entities.skills + expect(entity?.changes).toEqual({ added: ["Beta"], removed: [], updated: ["Alpha"] }) + expect(entity?.lastChangedAt).toBe(entity!.lastCheckedAt) + expect(published).toHaveLength(1) + expect(published[0]).toMatchObject({ + type: "altimate.workspace.sync.changed", + properties: { datamateId: 7, kind: "skills", changes: { added: ["Beta"], removed: [], updated: ["Alpha"] } }, + }) + // The event carries what the status route serves — never the raw item map. + expect(published[0].properties.state.items).toBeUndefined() + }) + + test("a check that finds nothing new publishes nothing and keeps the last change", async () => { + await SyncState.record(project, 7, "skills", { items: { a: skill("Alpha") } }) + await SyncState.record(project, 7, "skills", { items: { b: skill("Beta") } }) + published = [] + await SyncState.record(project, 7, "skills", { items: { b: skill("Beta") } }) + + expect(published).toEqual([]) + expect(SyncState.read(project, 7)?.entities.skills?.changes).toEqual({ added: ["Beta"], removed: ["Alpha"], updated: [] }) + }) + + test("an error is published once, keeps the last items, and recovery is published", async () => { + await SyncState.record(project, 7, "skills", { items: { a: skill("Alpha") } }) + await SyncState.record(project, 7, "skills", { error: "could not fetch the workspace's skill list" }) + await SyncState.record(project, 7, "skills", { error: "could not fetch the workspace's skill list" }) + + const failed = SyncState.read(project, 7)?.entities.skills + expect(failed).toMatchObject({ status: "error", count: 1, error: "could not fetch the workspace's skill list" }) + expect(published).toHaveLength(1) + expect(published[0].properties.changes).toBeNull() + + // Recovery with the same items: no difference, but the error state cleared. + await SyncState.record(project, 7, "skills", { items: { a: skill("Alpha") } }) + expect(published).toHaveLength(2) + expect(published[1].properties.state).toMatchObject({ status: "ok" }) + expect(published[1].properties.state.error).toBeUndefined() + }) + + test("a failure before anything was read leaves the next read as the baseline", async () => { + await SyncState.record(project, 7, "memory", { error: "could not load workspace memory" }) + published = [] + await SyncState.record(project, 7, "memory", { items: { m1: skill("Naming conventions") } }) + + expect(SyncState.read(project, 7)?.entities.memory?.changes).toBeNull() + // The recovery is still news; the existing block is not "added". + expect(published).toHaveLength(1) + expect(published[0].properties.changes).toBeNull() + }) + + test("kinds are recorded independently", async () => { + await Promise.all([ + SyncState.record(project, 7, "skills", { items: { a: skill("Alpha") } }), + SyncState.record(project, 7, "memory", { items: { m1: skill("One"), m2: skill("Two") } }), + ]) + const state = SyncState.read(project, 7) + expect(state?.entities.skills?.count).toBe(1) + expect(state?.entities.memory?.count).toBe(2) + }) + + test("another workspace's state is not this one's", async () => { + await SyncState.record(project, 7, "skills", { items: { a: skill("Alpha") } }) + expect(SyncState.read(project, 8)).toBeNull() + + // A rebind starts over: the new workspace's first check is again a baseline. + await SyncState.record(project, 8, "skills", { items: { z: skill("Zeta") } }) + expect(SyncState.read(project, 8)?.entities.skills?.changes).toBeNull() + expect(SyncState.read(project, 7)).toBeNull() + expect(published).toEqual([]) + }) + + test("a publish failure does not fail the record", async () => { + publish.mockImplementation((async () => { + throw new Error("no instance") + }) as never) + await SyncState.record(project, 7, "skills", { items: { a: skill("Alpha") } }) + await SyncState.record(project, 7, "skills", { items: { b: skill("Beta") } }) + expect(SyncState.read(project, 7)?.entities.skills?.changes).toEqual({ added: ["Beta"], removed: ["Alpha"], updated: [] }) + }) +}) diff --git a/packages/opencode/test/server/altimate-workspace-routes.test.ts b/packages/opencode/test/server/altimate-workspace-routes.test.ts index c5637eedea..30cfa3a386 100644 --- a/packages/opencode/test/server/altimate-workspace-routes.test.ts +++ b/packages/opencode/test/server/altimate-workspace-routes.test.ts @@ -6,6 +6,8 @@ import { afterEach, beforeEach, describe, expect, spyOn, test } from "bun:test" import { Server } from "../../src/server/server" import * as Manage from "../../src/altimate/workspace/manage" +import * as State from "../../src/altimate/workspace/state" +import * as SyncState from "../../src/altimate/workspace/sync-state" import { Session } from "../../src/session" import { NotFoundError } from "../../src/storage/db" import { resetDatabase } from "./db" @@ -230,6 +232,45 @@ describe("POST /altimate/workspace/sync", () => { }) }) +// altimate_change start — GET /altimate/workspace/status +describe("GET /altimate/workspace/status", () => { + const get = (headers: Record = {}) => Server.Default().request("/altimate/workspace/status", { headers }) + + test("returns the recorded state for the bound workspace", async () => { + spies.push( + spyOn(State, "resolveBindingOutcome").mockResolvedValue({ + status: "bound", + binding: { datamateId: 5, datamateName: "Analytics", repoRemote: null, projectPath: null, linkedAt: 1 }, + } as never), + ) + await SyncState.record(process.cwd(), 5, "skills", { items: { a: { label: "Alpha", version: "v1" } } }) + + const response = await get() + expect(response.status).toBe(200) + const body = (await response.json()) as { ok: boolean; datamateId: number; entities: Record } + expect(body.ok).toBe(true) + expect(body.datamateId).toBe(5) + expect(body.entities.skills).toMatchObject({ kind: "skills", status: "ok", count: 1 }) + // The item map is internal; the route serves the metadata only. + expect(body.entities.skills.items).toBeUndefined() + }) + + test("answers an unbound project with no workspace and no state", async () => { + spies.push(spyOn(State, "resolveBindingOutcome").mockResolvedValue({ status: "unbound" } as never)) + expect(await (await get()).json()).toEqual({ ok: true, datamateId: null, entities: {} }) + }) + + test("is refused under the kill switch", async () => { + process.env.ALTIMATE_DISABLE_WORKSPACE = "1" + expect((await get()).status).toBe(409) + }) + + test("refuses a browser origin on an unsecured server", async () => { + expect((await get({ origin: "https://evil.test" })).status).toBe(403) + }) +}) +// altimate_change end + describe("origin policy with a server password set", () => { // The password flag is read once at module load, so the policy is exercised directly with one. test("lets a native client (no Origin) and this server's own page through", () => { From 9d360873364736e1c1d76edabe277663088e59bd Mon Sep 17 00:00:00 2001 From: Sarav Date: Mon, 12 Oct 2026 07:54:58 +0530 Subject: [PATCH 2/4] fix(workspace): address sync-state review feedback - Write one state file per entity kind, so skill and memory writers in different threads or processes never overwrite each other - Create the state file owner-only (`0o600`) before renaming it into place - Store the account scope with each record; another account's state for the same workspace id is neither served nor diffed against - Add `isCurrent` to `SyncState.record`: memory fences on the binding epoch, skills re-resolve the binding, and a late result from a previous binding is dropped - Record memory load failures, both from the load's `catch` (now carrying the resolved `datamateId`) and from `refresh()`'s error branch - Clear the memory record when memory is turned off or the project is unlinked (`SyncState.forget`), so an earlier error does not stay latched - Key memory items by scope, origin and id - Record a skill kept at its previous copy at that copy's version - Add `lastSucceededAt`; "memory loaded" in the sidebar reads it and is gated on it alone - Use `Object.hasOwn` in `diff`; make `read` async - Document that events from a refresh outside the serving process reach no `/event` subscriber - Tests: failed memory loads, late records after a relink, file mode, carried skills, two writers, account scope, `forget`, `unknown` and empty-bound status routes, and a skill name that differs from its id Co-Authored-By: Claude Opus 5.5 (1M context) --- .../opencode/src/altimate/workspace/manage.ts | 4 +- .../src/altimate/workspace/memory-sync.ts | 55 ++++-- .../src/altimate/workspace/skill-sync.ts | 40 ++++- .../src/altimate/workspace/sync-state.ts | 156 +++++++++++++----- .../plugin/tui/altimate/workspace-sidebar.tsx | 2 +- .../altimate/workspace/memory-sync.test.ts | 97 ++++++++++- .../altimate/workspace/skill-sync.test.ts | 36 +++- .../altimate/workspace/sync-state.test.ts | 156 ++++++++++++++---- .../server/altimate-workspace-routes.test.ts | 30 +++- 9 files changed, 468 insertions(+), 108 deletions(-) diff --git a/packages/opencode/src/altimate/workspace/manage.ts b/packages/opencode/src/altimate/workspace/manage.ts index 342c58c21e..828c8f558e 100644 --- a/packages/opencode/src/altimate/workspace/manage.ts +++ b/packages/opencode/src/altimate/workspace/manage.ts @@ -162,7 +162,9 @@ export async function status( ...(memory === "unreadable" ? { memoryUnreadable: true as const } : {}), skillsEnabled: SkillSync.isEnabled(), skillsSyncedAt: await skillsSyncedAt(directory, binding), - memoryLoadedAt: binding ? (SyncState.read(directory, binding.datamateId)?.entities.memory?.lastCheckedAt ?? null) : null, + memoryLoadedAt: binding + ? ((await SyncState.read(directory, binding.datamateId).catch(() => null))?.entities.memory?.lastSucceededAt ?? null) + : null, } } diff --git a/packages/opencode/src/altimate/workspace/memory-sync.ts b/packages/opencode/src/altimate/workspace/memory-sync.ts index a8742f4e75..f189a03777 100644 --- a/packages/opencode/src/altimate/workspace/memory-sync.ts +++ b/packages/opencode/src/altimate/workspace/memory-sync.ts @@ -1386,6 +1386,8 @@ async function loadWorkspaceMemory(directory?: string): Promise { // Last epoch this load can vouch for. A failure after the binding resolved keeps that one, so // an error from the previous workspace is not stamped as the new binding's settled load. let vouched = epochFor(dir) + // The workspace this load resolved, so a failure past that point is recorded against it. + let resolved: number | undefined try { // The epoch must bracket the lookup: read only after it, a relink that lands while the // lookup is pending would stamp the old binding as current. Read only before it, a lookup @@ -1404,6 +1406,7 @@ async function loadWorkspaceMemory(directory?: string): Promise { if (!stable) return { status: "error", epoch, dir } if (!binding) return { status: "unlinked", epoch, dir } const datamateId = binding.datamateId + resolved = datamateId const enabled = await memoryStatus(binding) if (enabled === "error") return { status: "error", epoch, dir, datamateId } if (enabled === "disabled") return { status: "disabled", epoch, dir, datamateId } @@ -1447,7 +1450,7 @@ async function loadWorkspaceMemory(directory?: string): Promise { log.warn("workspace memory load failed", { err: String(err) }) // Stamped like any other outcome: without an epoch the session would reload (and make // the prompt wait) on every turn for as long as the service is down. - return { status: "error", epoch: vouched, dir } + return { status: "error", epoch: vouched, dir, datamateId: resolved } } } @@ -1474,26 +1477,41 @@ function commitLoad(sessionID: string, state: SessionMemory, outcome: LoadOutcom } // altimate_change start — workspace sync state -/** Record a committed load in the workspace sync state. Memory is loaded per session but - * recorded per project, so a second session reading the same blocks is not a change. - * Detached: the load is already committed and nothing waits on status metadata. A load - * that could not resolve a workspace, or found memory off, records nothing. */ +/** Record a load in the workspace sync state. Memory is loaded per session but recorded per + * project, so a second session reading the same blocks is not a change. + * + * Detached: nothing waits on status metadata. Fenced on the load's binding epoch at write + * time, so a record that settles after a relink cannot replace the new workspace's state. + * Memory found off, or the project unlinked, clears the record rather than leaving an + * earlier error latched. A load that never resolved a workspace records nothing. */ function recordSyncState(outcome: LoadOutcome): void { - if (!outcome.dir || outcome.datamateId === undefined) return - if (outcome.status !== "loaded" && outcome.status !== "error") return - const { dir, datamateId } = outcome - const items = - outcome.status === "loaded" - ? Object.fromEntries( - outcome.blocks.map((block) => [block.id, { label: blockTitle(block), version: block.updated ?? "" }]), - ) - : undefined + const dir = outcome.dir + if (!dir) return + const isCurrent = () => epochFor(dir) === outcome.epoch void import("./sync-state") - .then((SyncState) => - SyncState.record(dir, datamateId, "memory", items ? { items } : { error: "could not load workspace memory" }), - ) + .then((SyncState) => { + if (outcome.status === "unlinked" || outcome.status === "disabled") { + return isCurrent() ? SyncState.forget(dir, "memory") : undefined + } + if (outcome.datamateId === undefined) return + const result = + outcome.status === "loaded" + ? { items: Object.fromEntries(outcome.blocks.map((block) => [memoryItemKey(block), memoryItem(block)])) } + : { error: "could not load workspace memory" } + return SyncState.record(dir, outcome.datamateId, "memory", result, { isCurrent }) + }) .catch((err) => log.warn("could not record the memory sync state", { err: String(err) })) } + +/** Block ids are unique per file, not across scopes or across the teammates whose blocks a + * workspace serves, so the key carries both. */ +function memoryItemKey(block: RemoteMemoryBlock): string { + return `${block.scope}:${block.origin ?? "own"}:${block.id}` +} + +function memoryItem(block: RemoteMemoryBlock): { label: string; version: string } { + return { label: blockTitle(block), version: block.updated ?? "" } +} // altimate_change end /** A session's cloud overlay. Returns a copy so a caller cannot mutate the @@ -1537,6 +1555,9 @@ export async function refresh(sessionID: string, directory?: string): Promise, + carried: Map, ): Promise { const problem = describeSyncProblems(result) + // A carried skill is recorded at the version the snapshot still holds, not the one the list + // advertised: the update did not land, and the run that installs it reports it as updated. const items = rows ? Object.fromEntries( rows .filter((row) => !dropped.has(row.publicId)) - .map((row) => [row.publicId, { label: displayId(row.name ?? row.publicId), version: row.updatedAt }]), + .map((row) => [ + row.publicId, + { label: displayId(row.name ?? row.publicId), version: carried.get(row.publicId) ?? row.updatedAt }, + ]), ) : undefined + // A sync can outlast a relink. Its result describes the workspace it read, so it is dropped + // rather than written over the state of the one the project is bound to now. + const isCurrent = async () => { + const now = await resolveBindingOutcome(directory).catch(() => null) + return now?.status === "bound" && now.binding.datamateId === datamateId + } try { const SyncState = await import("./sync-state") - await SyncState.record(directory, datamateId, "skills", { - ...(items ? { items } : {}), - ...(problem ? { error: `${problem.title}: ${problem.message.split("\n").join("; ")}` } : {}), - }) + await SyncState.record( + directory, + datamateId, + "skills", + { + ...(items ? { items } : {}), + ...(problem ? { error: `${problem.title}: ${problem.message.split("\n").join("; ")}` } : {}), + }, + { isCurrent }, + ) } catch (err) { // Status metadata: never the reason a sync fails. log.warn("could not record the skill sync state", { err: String(err) }) @@ -1160,6 +1178,8 @@ export async function syncSkills(directory: string): Promise { let checkedDatamateId: number | undefined let remoteRows: RemoteSummary[] | null = null const dropped = new Set() + /** Skills kept at their previous copy, mapped to that copy's version. */ + const carriedVersions = new Map() let published = false // altimate_change end const run = (async () => { @@ -1487,6 +1507,7 @@ export async function syncSkills(directory: string): Promise { await fs.cp(priorDir, path.join(staging, summary.publicId), { recursive: true }) next.skills[summary.publicId] = prior carried = true + carriedVersions.set(summary.publicId, prior.updatedAt) } catch { carried = false } @@ -1628,7 +1649,14 @@ export async function syncSkills(directory: string): Promise { if (skippedSkills.length === 0 && !syncError) store.announced.delete(canon) const result: SyncResult = { changed, skipped: skippedSkills, error: syncError } // altimate_change start — workspace sync state - if (checkedDatamateId !== undefined) await recordSyncState(canon, checkedDatamateId, result, published ? remoteRows : null, dropped) + if (checkedDatamateId !== undefined) await recordSyncState( + canon, + checkedDatamateId, + result, + published ? remoteRows : null, + dropped, + carriedVersions, + ) // altimate_change end return result })() diff --git a/packages/opencode/src/altimate/workspace/sync-state.ts b/packages/opencode/src/altimate/workspace/sync-state.ts index 8b5c8fa7f9..de70659797 100644 --- a/packages/opencode/src/altimate/workspace/sync-state.ts +++ b/packages/opencode/src/altimate/workspace/sync-state.ts @@ -11,16 +11,27 @@ // Persisted rather than held in memory: in the TUI the per-turn syncs run in the server // worker while the sidebar renders on the main thread, and the two share no module state // (see `SkillSync.lastSuccessfulSyncAt`). A file is what both sides see. -import { createHash } from "node:crypto" -import { readFileSync } from "node:fs" +// +// One file per kind, so two writers of different kinds — the TUI main thread, the server +// worker and a separate CLI process can each hold one — never read-modify-write the same +// file. Writes of one kind within this module realm are chained; across realms the later +// rename wins, which loses at most the earlier check of that same kind, and the next check +// rewrites it. +// +// Event delivery: `Bus.publish` reaches the bus of the instance the record runs in. That is +// the `/event` stream for `serve` (the IDE extension) and for the TUI's per-turn syncs, which +// run in its server worker. A refresh run on the TUI main thread, or by a separate CLI +// process, publishes to no `/event` subscriber — the file is still written, and the TUI +// sidebar and the status route read it from there. +import { createHash, randomBytes } from "node:crypto" +import { chmodSync, mkdirSync, readFileSync, renameSync, rmSync, writeFileSync } from "node:fs" import path from "node:path" import z from "zod" import { Bus } from "@/bus" import { BusEvent } from "@/bus/bus-event" import { Global } from "@/global" import { Log } from "@/altimate/util/log" -import { Filesystem } from "@/util/filesystem" -import { canonicalDirectory } from "./state" +import { canonicalDirectory, currentScope } from "./state" const log = Log.create({ service: "altimate-workspace-sync-state" }) @@ -46,6 +57,9 @@ const EntityState = z.object({ count: z.number().int().nonnegative().nullable(), /** Epoch ms of the last completed check, successful or not. */ lastCheckedAt: z.number(), + /** Epoch ms of the last check that read the items successfully; null when none has. A + * "loaded Xm ago" line reads this, not `lastCheckedAt`, which also moves on failures. */ + lastSucceededAt: z.number().nullable(), /** Epoch ms of the last check that found a difference; null when none has. */ lastChangedAt: z.number().nullable(), /** What that last change was, by label. Kept until the next change replaces it. */ @@ -83,21 +97,34 @@ interface StoredEntity extends EntityState { } interface StoredFile { - version: 1 + version: 2 directory: string + /** The account the state was recorded under. Workspace ids are per tenant, so the same id + * under another account is another workspace. */ + scope: string datamateId: number - entities: Partial> + entity: StoredEntity } -function filePath(directory: string): string { +function syncDir(): string { + return path.join(Global.Path.state, "altimate-workspace-sync") +} + +function filePath(directory: string, kind: SyncKind): string { const key = createHash("sha256").update(canonicalDirectory(directory)).digest("hex").slice(0, 32) - return path.join(Global.Path.state, "altimate-workspace-sync", `${key}.json`) + return path.join(syncDir(), `${key}.${kind}.json`) } -function readStored(directory: string): StoredFile | null { +/** The account the caller's credentials name, as a string, or null when there is none. */ +async function scopeKey(): Promise { + const scope = await currentScope().catch(() => null) + return scope ? `${scope.tenant}|${scope.apiUrl}|${scope.account}` : null +} + +function readStored(directory: string, kind: SyncKind): StoredFile | null { try { - const parsed = JSON.parse(readFileSync(filePath(directory), "utf8")) as StoredFile - if (parsed?.version !== 1 || typeof parsed.datamateId !== "number" || !parsed.entities) return null + const parsed = JSON.parse(readFileSync(filePath(directory, kind), "utf8")) as StoredFile + if (parsed?.version !== 2 || typeof parsed.datamateId !== "number" || !parsed.entity) return null // A hash collision or a moved file must not attribute another tree's state to this one. if (parsed.directory !== canonicalDirectory(directory)) return null return parsed @@ -106,33 +133,50 @@ function readStored(directory: string): StoredFile | null { } } -/** The project's sync state for the workspace it is bound to now. Null when nothing has - * been recorded for that workspace — after a rebind the previous workspace's state is not - * this one's. */ -export function read(directory: string, datamateId: number): WorkspaceSyncState | null { - const stored = readStored(directory) - if (!stored || stored.datamateId !== datamateId) return null +/** Private from the first byte, like the binding cache and the memory index beside it: the + * file holds the checkout path, skill names, memory titles and error text. The temp file is + * created owner-only and restricted before it is renamed into place, and one that cannot be + * restricted is never published. */ +function writePrivate(target: string, value: unknown): void { + mkdirSync(path.dirname(target), { recursive: true }) + const tmp = `${target}.tmp-${randomBytes(6).toString("hex")}` + writeFileSync(tmp, JSON.stringify(value, null, 2) + "\n", { mode: 0o600 }) + try { + chmodSync(tmp, 0o600) + renameSync(tmp, target) + } catch (err) { + rmSync(tmp, { force: true }) + throw err + } +} + +/** The project's sync state for the workspace it is bound to now, under the account in use. + * Null when nothing has been recorded for that workspace — after a rebind or an account + * switch the previous workspace's state is not this one's. */ +export async function read(directory: string, datamateId: number): Promise { + const scope = await scopeKey() + if (!scope) return null const entities: WorkspaceSyncState["entities"] = {} for (const kind of KINDS) { - const entity = stored.entities[kind] - if (!entity) continue - const { items: _items, ...state } = entity + const stored = readStored(directory, kind) + if (!stored || stored.datamateId !== datamateId || stored.scope !== scope) continue + const { items: _items, ...state } = stored.entity entities[kind] = state } - return { datamateId, entities } + return Object.keys(entities).length > 0 ? { datamateId, entities } : null } -/** Added, removed and updated labels between two item sets. Pure. */ +/** Added, removed and updated labels between two item sets. Pure. Own keys only, so an id + * such as `constructor` is not mistaken for an inherited property. */ export function diff(previous: SyncItems, next: SyncItems): SyncChanges { const added: string[] = [] const updated: string[] = [] for (const [id, item] of Object.entries(next)) { - const before = previous[id] - if (!before) added.push(item.label) - else if (before.version !== item.version) updated.push(item.label) + if (!Object.hasOwn(previous, id)) added.push(item.label) + else if (previous[id].version !== item.version) updated.push(item.label) } const removed = Object.entries(previous) - .filter(([id]) => !(id in next)) + .filter(([id]) => !Object.hasOwn(next, id)) .map(([, item]) => item.label) return { added: added.sort(), removed: removed.sort(), updated: updated.sort() } } @@ -149,16 +193,39 @@ export interface Outcome { error?: string } -/** Writes are chained per file so two kinds settling at once cannot drop each other's update. */ +export interface RecordOptions { + /** Whether the check still describes the project's current binding. Checked just before + * the write: a long sync, or a detached memory record, can settle after the project was + * rebound, and its result must not replace the new workspace's state. */ + isCurrent?: () => boolean | Promise +} + +/** Writes are chained per file within this module realm; see the header for other writers. */ const writes = new Map>() /** Record a completed check of one kind and publish when it changed something. * * Never throws: this is status metadata, and a failure here must not fail the sync that * reported it. Resolves once the state is written and any event published. */ -export function record(directory: string, datamateId: number, kind: SyncKind, outcome: Outcome): Promise { - const file = filePath(directory) - const run = (writes.get(file) ?? Promise.resolve()).then(() => apply(directory, datamateId, kind, outcome)) +export function record( + directory: string, + datamateId: number, + kind: SyncKind, + outcome: Outcome, + options: RecordOptions = {}, +): Promise { + return chain(filePath(directory, kind), () => apply(directory, datamateId, kind, outcome, options), kind) +} + +/** Drop a kind's record, without an event: memory turned off, or the project unlinked. A + * problem recorded before would otherwise stay latched for as long as nothing checks it. */ +export function forget(directory: string, kind: SyncKind): Promise { + const file = filePath(directory, kind) + return chain(file, async () => rmSync(file, { force: true }), kind) +} + +function chain(file: string, work: () => Promise, kind: SyncKind): Promise { + const run = (writes.get(file) ?? Promise.resolve()).then(work) const settled = run.catch((err) => log.warn("could not record workspace sync state", { kind, err: String(err) })) writes.set(file, settled) void settled.finally(() => { @@ -167,13 +234,25 @@ export function record(directory: string, datamateId: number, kind: SyncKind, ou return settled } -async function apply(directory: string, datamateId: number, kind: SyncKind, outcome: Outcome): Promise { +async function apply( + directory: string, + datamateId: number, + kind: SyncKind, + outcome: Outcome, + options: RecordOptions, +): Promise { + if (options.isCurrent && !(await options.isCurrent())) { + log.info("dropped a sync result for a binding that is no longer current", { kind, datamateId }) + return + } + const scope = await scopeKey() + // No account, no attribution: a record that cannot say whose it is would be served to + // whichever account looks next. + if (!scope) return const canon = canonicalDirectory(directory) - const stored = readStored(directory) - // Another workspace's state describes nothing about this one: start over. - const file: StoredFile = - stored && stored.datamateId === datamateId ? stored : { version: 1, directory: canon, datamateId, entities: {} } - const previous = file.entities[kind] + const stored = readStored(directory, kind) + // Another workspace's (or account's) state describes nothing about this one: start over. + const previous = stored && stored.datamateId === datamateId && stored.scope === scope ? stored.entity : undefined const now = Date.now() let changes: SyncChanges | null = null @@ -191,13 +270,14 @@ async function apply(directory: string, datamateId: number, kind: SyncKind, outc status: outcome.error ? "error" : "ok", count: items ? Object.keys(items).length : null, lastCheckedAt: now, + lastSucceededAt: outcome.items ? now : (previous?.lastSucceededAt ?? null), lastChangedAt: changes ? now : (previous?.lastChangedAt ?? null), changes: changes ?? previous?.changes ?? null, ...(outcome.error ? { error: outcome.error } : {}), ...(items ? { items } : {}), } - file.entities[kind] = next - Filesystem.writeJsonAtomic(filePath(directory), file) + const file: StoredFile = { version: 2, directory: canon, scope, datamateId, entity: next } + writePrivate(filePath(directory, kind), file) const errorChanged = (previous?.status ?? "ok") !== next.status || previous?.error !== next.error if (!changes && !errorChanged) return diff --git a/packages/opencode/src/plugin/tui/altimate/workspace-sidebar.tsx b/packages/opencode/src/plugin/tui/altimate/workspace-sidebar.tsx index 7f06b74f6f..069bd54b61 100644 --- a/packages/opencode/src/plugin/tui/altimate/workspace-sidebar.tsx +++ b/packages/opencode/src/plugin/tui/altimate/workspace-sidebar.tsx @@ -308,7 +308,7 @@ function View(props: { api: TuiPluginApi }) { {(at) => {`skills synced ${describeAge(at())}`}} {/* altimate_change start — workspace sync state */} - + {(at) => {`memory loaded ${describeAge(at())}`}} {/* altimate_change end */} diff --git a/packages/opencode/test/altimate/workspace/memory-sync.test.ts b/packages/opencode/test/altimate/workspace/memory-sync.test.ts index 3b2b0d1a91..8866bf5edf 100644 --- a/packages/opencode/test/altimate/workspace/memory-sync.test.ts +++ b/packages/opencode/test/altimate/workspace/memory-sync.test.ts @@ -2066,15 +2066,22 @@ describe("sync state", () => { async function memoryState(dir: string, predicate: (e: any) => boolean = () => true) { const SyncState = await import("../../../src/altimate/workspace/sync-state") for (let i = 0; i < 50; i++) { - const entity = SyncState.read(dir, BINDING.datamateId)?.entities.memory + const entity = (await SyncState.read(dir, BINDING.datamateId))?.entities.memory if (entity && predicate(entity)) return entity await new Promise((r) => setTimeout(r, 10)) } - return SyncState.read(dir, BINDING.datamateId)?.entities.memory + return (await SyncState.read(dir, BINDING.datamateId))?.entities.memory + } + + /** Settles once every record queued for this project's memory so far has. */ + async function settled(dir: string) { + const SyncState = await import("../../../src/altimate/workspace/sync-state") + await new Promise((r) => setTimeout(r, 20)) + await SyncState.record(dir, BINDING.datamateId, "memory", {}, { isCurrent: () => false }) } test("loads are recorded per project; a later load records what changed, by title", async () => { - const dir = mkdtempSync(path.join(os.tmpdir(), "memory-sync-state-")) + const dir = mkdtempSync(path.join(SANDBOX, "memory-sync-state-")) listResponse = [note("a", "Naming conventions"), note("b", "Warehouse sizing")] await refresh(`${SES}-state-1`, dir) expect(await memoryState(dir)).toMatchObject({ kind: "memory", status: "ok", count: 2, changes: null }) @@ -2091,10 +2098,92 @@ describe("sync state", () => { }) test("a load for an unbound project records nothing", async () => { - const dir = mkdtempSync(path.join(os.tmpdir(), "memory-sync-state-off-")) + const dir = mkdtempSync(path.join(SANDBOX, "memory-sync-state-off-")) syncInternals.resolveBinding = async () => null await refresh(`${SES}-state-off`, dir) expect(await memoryState(dir)).toBeUndefined() }) + + test("a load that fails is recorded as an error against the workspace it resolved", async () => { + const dir = mkdtempSync(path.join(SANDBOX, "memory-sync-state-fail-")) + listResponse = [note("a", "Naming conventions")] + await refresh(`${SES}-state-fail-1`, dir) + const loaded = await memoryState(dir) + expect(loaded).toMatchObject({ status: "ok", count: 1 }) + + listFails = true + try { + const result = await refresh(`${SES}-state-fail-2`, dir) + expect(result.status).toBe("error") + } finally { + listFails = false + } + const failed = await memoryState(dir, (e) => e.status === "error") + expect(failed).toMatchObject({ status: "error", count: 1, error: "could not load workspace memory" }) + // "memory loaded Xm ago" reads the last success, which the failure did not move. + expect(failed?.lastSucceededAt).toBe(loaded!.lastSucceededAt) + }) + + test("memory turned off clears an error recorded before", async () => { + const dir = mkdtempSync(path.join(SANDBOX, "memory-sync-state-disabled-")) + listFails = true + try { + await refresh(`${SES}-state-disabled-1`, dir) + } finally { + listFails = false + } + expect(await memoryState(dir, (e) => e.status === "error")).toMatchObject({ status: "error" }) + + workspaces = [{ id: 42, name: "acme", memory_enabled: false }] + resetOverlay() + await refresh(`${SES}-state-disabled-2`, dir) + await settled(dir) + expect(await memoryState(dir)).toBeUndefined() + }) + + test("a record that settles after a relink does not replace the new binding's state", async () => { + const SyncState = await import("../../../src/altimate/workspace/sync-state") + const dir = mkdtempSync(path.join(SANDBOX, "memory-sync-state-late-")) + listResponse = [note("a", "Naming conventions")] + await refresh(`${SES}-state-late-1`, dir) + expect(await memoryState(dir)).toMatchObject({ status: "ok", count: 1 }) + + // Hold this project's memory writes, so the next load's record is still queued when the + // project is relinked. + let release!: () => void + const gate = new Promise((r) => (release = r)) + const holder = SyncState.record(dir, BINDING.datamateId, "memory", {}, { + isCurrent: async () => { + await gate + return false + }, + }) + listResponse = [] + await refresh(`${SES}-state-late-2`, dir) + // Unlink: every binding epoch moves, and the queued "nothing left" record is now stale. + resetOverlay() + release() + await holder + await settled(dir) + + expect(await memoryState(dir)).toMatchObject({ status: "ok", count: 1, changes: null }) + }) + + test("blocks are keyed by scope and origin, so equal ids from two sources are two items", async () => { + const dir = mkdtempSync(path.join(SANDBOX, "memory-sync-state-keys-")) + const project = (id: string, title: string, repo: string) => ({ + ...note("same", title), + id, + metadata: { ...note("same", title).metadata, block_scope: "project", datamate_id: 42, repo_remote: repo }, + }) + listResponse = [ + note("same", "Global note"), + project("mem-own", "Own project note", BINDING.repoRemote), + project("mem-sibling", "Sibling project note", "ssh://git@github.com/acme/other.git"), + ] + await refresh(`${SES}-state-keys`, dir) + expect(overlayBlocks(`${SES}-state-keys`)).toHaveLength(3) + expect(await memoryState(dir)).toMatchObject({ status: "ok", count: 3 }) + }) }) // altimate_change end diff --git a/packages/opencode/test/altimate/workspace/skill-sync.test.ts b/packages/opencode/test/altimate/workspace/skill-sync.test.ts index 36794836a4..f8c5d8dd4b 100644 --- a/packages/opencode/test/altimate/workspace/skill-sync.test.ts +++ b/packages/opencode/test/altimate/workspace/skill-sync.test.ts @@ -142,13 +142,17 @@ afterAll(() => { /** Serve the real contract: a paginated summary page, a detail view carrying * files[{path,size}], and raw file bytes. */ -function serve(skills: Record>, updatedAt = "2026-01-01T00:00:00Z") { +function serve( + skills: Record>, + updatedAt = "2026-01-01T00:00:00Z", + names: Record = {}, +) { // Shapes verified against a local backend on `development`: the list is NOT // wrapped, the detail IS wrapped in `{skill: ...}`, and the file endpoint // answers `{path, content}` JSON rather than raw bytes. const items = Object.keys(skills).map((id) => ({ public_id: id, - name: id, + name: names[id] ?? id, file_count: Object.keys(skills[id]).length, updated_at: updatedAt, })) @@ -2568,17 +2572,19 @@ describe("shouldAnnounce", () => { // altimate_change start — workspace sync state describe("sync state", () => { - const read = async () => (await import("@/altimate/workspace/sync-state")).read(project, 1)?.entities.skills + const read = async () => + (await (await import("@/altimate/workspace/sync-state")).read(project, 1))?.entities.skills test("the first sync is a baseline and a later one records what changed, by name", async () => { - serve({ "pub-1": { "SKILL.md": "one" } }) + const names = { "pub-1": "revenue-model", "pub-2": "cost-alerts" } + serve({ "pub-1": { "SKILL.md": "one" } }, undefined, names) await syncSkills(project) expect(await read()).toMatchObject({ status: "ok", count: 1, changes: null, lastChangedAt: null }) - serve({ "pub-1": { "SKILL.md": "one" }, "pub-2": { "SKILL.md": "two" } }) + serve({ "pub-1": { "SKILL.md": "one" }, "pub-2": { "SKILL.md": "two" } }, undefined, names) await syncSkills(project) const entity = await read() - expect(entity).toMatchObject({ status: "ok", count: 2, changes: { added: ["pub-2"], removed: [], updated: [] } }) + expect(entity).toMatchObject({ status: "ok", count: 2, changes: { added: ["cost-alerts"], removed: [], updated: [] } }) expect(entity?.lastChangedAt).toBeNumber() }) @@ -2617,5 +2623,23 @@ describe("sync state", () => { // pub-1 moved to a new version; pub-2 never arrived, so it is not reported as added. expect(entity?.changes).toEqual({ added: [], removed: [], updated: ["pub-1"] }) }) + + test("a skill kept at its previous copy is not reported as updated until the update lands", async () => { + serve({ "pub-1": { "SKILL.md": "one" } }, "2026-01-01T00:00:00Z") + await syncSkills(project) + + serve({ "pub-1": { "SKILL.md": "one, edited" } }, "2026-02-01T00:00:00Z") + const inner = globalThis.fetch + globalThis.fetch = (async (input: string | URL, init?: RequestInit) => { + if (String(input).includes("/pub-1/files/")) throw new Error("offline") + return inner(input as never, init as never) + }) as unknown as typeof fetch + await syncSkills(project) + expect(await read()).toMatchObject({ status: "error", count: 1, changes: null }) + + serve({ "pub-1": { "SKILL.md": "one, edited" } }, "2026-02-01T00:00:00Z") + await syncSkills(project) + expect(await read()).toMatchObject({ status: "ok", changes: { added: [], removed: [], updated: ["pub-1"] } }) + }) }) // altimate_change end diff --git a/packages/opencode/test/altimate/workspace/sync-state.test.ts b/packages/opencode/test/altimate/workspace/sync-state.test.ts index 61476d03a5..52d9439491 100644 --- a/packages/opencode/test/altimate/workspace/sync-state.test.ts +++ b/packages/opencode/test/altimate/workspace/sync-state.test.ts @@ -2,45 +2,54 @@ // // Unit coverage for the workspace sync state (src/altimate/workspace/sync-state.ts): // what a check records, what counts as a change, and when the change event is published. -// The state file is real, in a sandboxed state directory; only the bus is spied, because a -// publish needs an instance the unit under test does not own. +// The state files are real, in a sandboxed state directory; the bus and the account in use +// are spied, because a publish needs an instance and a scope needs credentials, neither of +// which the unit under test owns. import { afterAll, afterEach, beforeEach, describe, expect, spyOn, test } from "bun:test" -import { mkdirSync, mkdtempSync, realpathSync, rmSync } from "node:fs" +import { mkdirSync, mkdtempSync, readdirSync, realpathSync, rmSync, statSync } from "node:fs" import os from "node:os" import path from "node:path" +import { Bus } from "../../../src/bus" +import * as State from "../../../src/altimate/workspace/state" +import * as SyncState from "../../../src/altimate/workspace/sync-state" -// Global.Path.state resolves at module load, so the sandbox must exist first. -const ORIGINAL_XDG_STATE_HOME = process.env.XDG_STATE_HOME +const ORIGINAL_STATE_HOME = process.env.OPENCODE_TEST_STATE_HOME const SANDBOX = path.join(os.tmpdir(), `altimate-sync-state-${process.pid}-${Date.now()}`) -mkdirSync(path.join(SANDBOX, "state"), { recursive: true }) -process.env.XDG_STATE_HOME = path.join(SANDBOX, "state") +const STATE_HOME = path.join(SANDBOX, "state") +mkdirSync(STATE_HOME, { recursive: true }) +process.env.OPENCODE_TEST_STATE_HOME = STATE_HOME 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_STATE_HOME === undefined) delete process.env.OPENCODE_TEST_STATE_HOME + else process.env.OPENCODE_TEST_STATE_HOME = ORIGINAL_STATE_HOME rmSync(SANDBOX, { recursive: true, force: true }) }) -const { Bus } = await import("../../../src/bus") -const SyncState = await import("../../../src/altimate/workspace/sync-state") +const ACCOUNT = { tenant: "acme", apiUrl: "https://api.example.com", account: "digest-a" } let project: string let published: Array<{ type: string; properties: any }> -let publish: ReturnType +let spies: Array<{ mockRestore: () => void }> +let scope: ReturnType beforeEach(() => { project = realpathSync(mkdtempSync(path.join(SANDBOX, "project-"))) published = [] - publish = spyOn(Bus, "publish").mockImplementation((async (def: { type: string }, properties: unknown) => { - published.push({ type: def.type, properties }) - }) as never) + spies = [ + spyOn(Bus, "publish").mockImplementation((async (def: { type: string }, properties: unknown) => { + published.push({ type: def.type, properties }) + }) as never), + ] + scope = spyOn(State, "currentScope").mockResolvedValue(ACCOUNT) + spies.push(scope) }) afterEach(() => { - publish.mockRestore() + for (const spy of spies) spy.mockRestore() }) const skill = (label: string, version = "v1") => ({ label, version }) +const syncFiles = () => readdirSync(path.join(STATE_HOME, "altimate-workspace-sync")) describe("diff", () => { test("reports added, removed and updated labels, sorted", () => { @@ -54,15 +63,21 @@ describe("diff", () => { test("identical sets have no changes", () => { expect(SyncState.diff({ a: skill("Alpha") }, { a: skill("Alpha") })).toEqual({ added: [], removed: [], updated: [] }) }) + + test("an id named like an inherited property is still an item", () => { + expect(SyncState.diff({}, { constructor: skill("Ctor") })).toEqual({ added: ["Ctor"], removed: [], updated: [] }) + expect(SyncState.diff({ toString: skill("Str") }, {})).toEqual({ added: [], removed: ["Str"], updated: [] }) + }) }) describe("record", () => { test("the first check is a baseline: recorded, not announced", async () => { await SyncState.record(project, 7, "skills", { items: { a: skill("Alpha"), b: skill("Beta") } }) - const state = SyncState.read(project, 7) + const state = await SyncState.read(project, 7) expect(state?.entities.skills).toMatchObject({ kind: "skills", status: "ok", count: 2, lastChangedAt: null, changes: null }) expect(state?.entities.skills?.lastCheckedAt).toBeNumber() + expect(state?.entities.skills?.lastSucceededAt).toBe(state!.entities.skills!.lastCheckedAt) expect(published).toEqual([]) }) @@ -70,7 +85,7 @@ describe("record", () => { await SyncState.record(project, 7, "skills", { items: { a: skill("Alpha") } }) await SyncState.record(project, 7, "skills", { items: { a: skill("Alpha", "v2"), b: skill("Beta") } }) - const entity = SyncState.read(project, 7)?.entities.skills + const entity = (await SyncState.read(project, 7))?.entities.skills expect(entity?.changes).toEqual({ added: ["Beta"], removed: [], updated: ["Alpha"] }) expect(entity?.lastChangedAt).toBe(entity!.lastCheckedAt) expect(published).toHaveLength(1) @@ -89,16 +104,25 @@ describe("record", () => { await SyncState.record(project, 7, "skills", { items: { b: skill("Beta") } }) expect(published).toEqual([]) - expect(SyncState.read(project, 7)?.entities.skills?.changes).toEqual({ added: ["Beta"], removed: ["Alpha"], updated: [] }) + expect((await SyncState.read(project, 7))?.entities.skills?.changes).toEqual({ + added: ["Beta"], + removed: ["Alpha"], + updated: [], + }) }) - test("an error is published once, keeps the last items, and recovery is published", async () => { + test("an error is published once, keeps the last items and success time, and recovery is published", async () => { await SyncState.record(project, 7, "skills", { items: { a: skill("Alpha") } }) + const succeededAt = (await SyncState.read(project, 7))!.entities.skills!.lastSucceededAt + await Bun.sleep(5) await SyncState.record(project, 7, "skills", { error: "could not fetch the workspace's skill list" }) await SyncState.record(project, 7, "skills", { error: "could not fetch the workspace's skill list" }) - const failed = SyncState.read(project, 7)?.entities.skills + const failed = (await SyncState.read(project, 7))?.entities.skills expect(failed).toMatchObject({ status: "error", count: 1, error: "could not fetch the workspace's skill list" }) + // A failure is a check, not a load: "loaded Xm ago" must not move. + expect(failed?.lastSucceededAt).toBe(succeededAt) + expect(failed!.lastCheckedAt).toBeGreaterThan(succeededAt!) expect(published).toHaveLength(1) expect(published[0].properties.changes).toBeNull() @@ -111,42 +135,116 @@ describe("record", () => { test("a failure before anything was read leaves the next read as the baseline", async () => { await SyncState.record(project, 7, "memory", { error: "could not load workspace memory" }) + expect((await SyncState.read(project, 7))?.entities.memory).toMatchObject({ count: null, lastSucceededAt: null }) published = [] await SyncState.record(project, 7, "memory", { items: { m1: skill("Naming conventions") } }) - expect(SyncState.read(project, 7)?.entities.memory?.changes).toBeNull() + expect((await SyncState.read(project, 7))?.entities.memory?.changes).toBeNull() // The recovery is still news; the existing block is not "added". expect(published).toHaveLength(1) expect(published[0].properties.changes).toBeNull() }) - test("kinds are recorded independently", async () => { + test("kinds are recorded independently, each in its own file", async () => { await Promise.all([ SyncState.record(project, 7, "skills", { items: { a: skill("Alpha") } }), SyncState.record(project, 7, "memory", { items: { m1: skill("One"), m2: skill("Two") } }), ]) - const state = SyncState.read(project, 7) + const state = await SyncState.read(project, 7) expect(state?.entities.skills?.count).toBe(1) expect(state?.entities.memory?.count).toBe(2) }) + test("two writers of different kinds never overwrite each other", async () => { + // Another realm's writer (the TUI main thread, a CLI process) has no share in this + // module's write chain. Writing each kind to its own file is what keeps both records. + const before = new Set(syncFiles()) + await SyncState.record(project, 7, "skills", { items: { a: skill("Alpha") } }) + await SyncState.record(project, 7, "memory", { items: { m1: skill("One") } }) + const written = syncFiles().filter((name) => !before.has(name)) + expect(written.map((name) => name.split(".").slice(1).join(".")).sort()).toEqual(["memory.json", "skills.json"]) + expect(new Set(written.map((name) => name.split(".")[0])).size).toBe(1) + await SyncState.record(project, 7, "skills", { items: { a: skill("Alpha"), b: skill("Beta") } }) + + const state = await SyncState.read(project, 7) + expect(state?.entities.memory?.count).toBe(1) + expect(state?.entities.skills?.count).toBe(2) + }) + + test("the state file is owner-only and no temp file is left behind", async () => { + await SyncState.record(project, 7, "skills", { items: { a: skill("Alpha") } }) + const files = syncFiles() + expect(files.some((name) => name.includes(".tmp-"))).toBe(false) + for (const name of files) { + expect(statSync(path.join(STATE_HOME, "altimate-workspace-sync", name)).mode & 0o777).toBe(0o600) + } + }) + test("another workspace's state is not this one's", async () => { await SyncState.record(project, 7, "skills", { items: { a: skill("Alpha") } }) - expect(SyncState.read(project, 8)).toBeNull() + expect(await SyncState.read(project, 8)).toBeNull() // A rebind starts over: the new workspace's first check is again a baseline. await SyncState.record(project, 8, "skills", { items: { z: skill("Zeta") } }) - expect(SyncState.read(project, 8)?.entities.skills?.changes).toBeNull() - expect(SyncState.read(project, 7)).toBeNull() + expect((await SyncState.read(project, 8))?.entities.skills?.changes).toBeNull() + expect(await SyncState.read(project, 7)).toBeNull() expect(published).toEqual([]) }) + test("a late result from a previous binding is dropped", async () => { + await SyncState.record(project, 8, "memory", { items: { m1: skill("New workspace block") } }) + await SyncState.record(project, 7, "memory", { error: "could not load workspace memory" }, { isCurrent: () => false }) + + const state = await SyncState.read(project, 8) + expect(state?.entities.memory).toMatchObject({ status: "ok", count: 1 }) + expect(published).toEqual([]) + }) + + test("the same workspace id under another account is another workspace", async () => { + await SyncState.record(project, 7, "skills", { items: { a: skill("Alpha") } }) + scope.mockResolvedValue({ ...ACCOUNT, tenant: "other", account: "digest-b" }) + expect(await SyncState.read(project, 7)).toBeNull() + + // And its first check is a baseline, not "Alpha removed". + await SyncState.record(project, 7, "skills", { items: { b: skill("Beta") } }) + expect((await SyncState.read(project, 7))?.entities.skills?.changes).toBeNull() + expect(published).toEqual([]) + }) + + test("without an account nothing is recorded or served", async () => { + scope.mockResolvedValue(null) + await SyncState.record(project, 7, "skills", { items: { a: skill("Alpha") } }) + expect(await SyncState.read(project, 7)).toBeNull() + scope.mockResolvedValue(ACCOUNT) + expect(await SyncState.read(project, 7)).toBeNull() + }) + test("a publish failure does not fail the record", async () => { - publish.mockImplementation((async () => { + spies[0].mockRestore() + spies[0] = spyOn(Bus, "publish").mockImplementation((async () => { throw new Error("no instance") }) as never) await SyncState.record(project, 7, "skills", { items: { a: skill("Alpha") } }) await SyncState.record(project, 7, "skills", { items: { b: skill("Beta") } }) - expect(SyncState.read(project, 7)?.entities.skills?.changes).toEqual({ added: ["Beta"], removed: ["Alpha"], updated: [] }) + expect((await SyncState.read(project, 7))?.entities.skills?.changes).toEqual({ + added: ["Beta"], + removed: ["Alpha"], + updated: [], + }) + }) +}) + +describe("forget", () => { + test("clears one kind's record, latched error included, and leaves the others", async () => { + await SyncState.record(project, 7, "skills", { items: { a: skill("Alpha") } }) + await SyncState.record(project, 7, "memory", { error: "could not load workspace memory" }) + published = [] + + await SyncState.forget(project, "memory") + + const state = await SyncState.read(project, 7) + expect(state?.entities.memory).toBeUndefined() + expect(state?.entities.skills?.count).toBe(1) + expect(published).toEqual([]) }) }) diff --git a/packages/opencode/test/server/altimate-workspace-routes.test.ts b/packages/opencode/test/server/altimate-workspace-routes.test.ts index e5d4a305c6..b667dc4e9e 100644 --- a/packages/opencode/test/server/altimate-workspace-routes.test.ts +++ b/packages/opencode/test/server/altimate-workspace-routes.test.ts @@ -236,20 +236,28 @@ describe("POST /altimate/workspace/sync", () => { // altimate_change start — GET /altimate/workspace/status describe("GET /altimate/workspace/status", () => { const get = (headers: Record = {}) => Server.Default().request("/altimate/workspace/status", { headers }) - - test("returns the recorded state for the bound workspace", async () => { + const bound = (datamateId: number) => + spyOn(State, "resolveBindingOutcome").mockResolvedValue({ + status: "bound", + binding: { datamateId, datamateName: "Analytics", repoRemote: null, projectPath: null, linkedAt: 1 }, + } as never) + + beforeEach(() => { + // State is recorded and served per account. spies.push( - spyOn(State, "resolveBindingOutcome").mockResolvedValue({ - status: "bound", - binding: { datamateId: 5, datamateName: "Analytics", repoRemote: null, projectPath: null, linkedAt: 1 }, - } as never), + spyOn(State, "currentScope").mockResolvedValue({ tenant: "acme", apiUrl: "https://api.example.com", account: "d" }), ) + }) + + test("returns the recorded state for the bound workspace", async () => { + spies.push(bound(5)) await SyncState.record(process.cwd(), 5, "skills", { items: { a: { label: "Alpha", version: "v1" } } }) const response = await get() expect(response.status).toBe(200) const body = (await response.json()) as { ok: boolean; datamateId: number; entities: Record } expect(body.ok).toBe(true) + expect((body as { binding?: string }).binding).toBe("bound") expect(body.datamateId).toBe(5) expect(body.entities.skills).toMatchObject({ kind: "skills", status: "ok", count: 1 }) // The item map is internal; the route serves the metadata only. @@ -261,6 +269,16 @@ describe("GET /altimate/workspace/status", () => { expect(await (await get()).json()).toEqual({ ok: true, binding: "unbound", datamateId: null, entities: {} }) }) + test("says when the binding could not be resolved, rather than reporting it unbound", async () => { + spies.push(spyOn(State, "resolveBindingOutcome").mockResolvedValue({ status: "unknown" } as never)) + expect(await (await get()).json()).toEqual({ ok: true, binding: "unknown", datamateId: null, entities: {} }) + }) + + test("a bound project with nothing recorded yet has no entities", async () => { + spies.push(bound(987_654)) + expect(await (await get()).json()).toEqual({ ok: true, binding: "bound", datamateId: 987_654, entities: {} }) + }) + test("is refused under the kill switch", async () => { process.env.ALTIMATE_DISABLE_WORKSPACE = "1" expect((await get()).status).toBe(409) From 45ffa61d55120625c63adb6451a7cd6a5fb4fbd2 Mon Sep 17 00:00:00 2001 From: Sarav Date: Mon, 12 Oct 2026 07:57:17 +0530 Subject: [PATCH 3/4] fix(workspace): version memory items by content and retire skill state on deactivate - Memory items are versioned by `block_updated` plus a content digest, so a web edit that keeps the mirror timestamp is still reported as updated - Taking the skill snapshot out of service clears its sync-state record, so a relink to the same workspace starts from a fresh baseline Co-Authored-By: Claude Opus 5.5 (1M context) --- .../src/altimate/workspace/memory-sync.ts | 5 +++- .../src/altimate/workspace/skill-sync.ts | 7 +++++ .../altimate/workspace/memory-sync.test.ts | 12 ++++++++ .../altimate/workspace/skill-sync.test.ts | 30 +++++++++++++++++++ 4 files changed, 53 insertions(+), 1 deletion(-) diff --git a/packages/opencode/src/altimate/workspace/memory-sync.ts b/packages/opencode/src/altimate/workspace/memory-sync.ts index f189a03777..eef8c139ac 100644 --- a/packages/opencode/src/altimate/workspace/memory-sync.ts +++ b/packages/opencode/src/altimate/workspace/memory-sync.ts @@ -1509,8 +1509,11 @@ function memoryItemKey(block: RemoteMemoryBlock): string { return `${block.scope}:${block.origin ?? "own"}:${block.id}` } +/** Versioned by content as well as the mirror's timestamp: an edit made in the web app can keep + * the block's original `block_updated`, and the session still loads the new text. */ function memoryItem(block: RemoteMemoryBlock): { label: string; version: string } { - return { label: blockTitle(block), version: block.updated ?? "" } + const digest = createHash("sha256").update(block.content).digest("hex").slice(0, 16) + return { label: blockTitle(block), version: `${block.updated ?? ""}:${digest}` } } // altimate_change end diff --git a/packages/opencode/src/altimate/workspace/skill-sync.ts b/packages/opencode/src/altimate/workspace/skill-sync.ts index 4c1221e9b2..0bb6aafe5b 100644 --- a/packages/opencode/src/altimate/workspace/skill-sync.ts +++ b/packages/opencode/src/altimate/workspace/skill-sync.ts @@ -1086,6 +1086,13 @@ async function deactivate(directory: string, why: string): Promise { // here, before the early return: relinking must sync again rather than repeat a stale "none". const canon = path.resolve(directory) await clearEmptyRecord(canon) + // altimate_change start — workspace sync state + // The snapshot is out of service, so its record no longer describes anything here: a relink to + // the same workspace starts from a fresh baseline instead of the old count. + await import("./sync-state") + .then((SyncState) => SyncState.forget(canon, "skills")) + .catch((err) => log.warn("could not clear the skill sync state", { err: String(err) })) + // altimate_change end try { await fs.stat(root) } catch { diff --git a/packages/opencode/test/altimate/workspace/memory-sync.test.ts b/packages/opencode/test/altimate/workspace/memory-sync.test.ts index 8866bf5edf..f6174f523c 100644 --- a/packages/opencode/test/altimate/workspace/memory-sync.test.ts +++ b/packages/opencode/test/altimate/workspace/memory-sync.test.ts @@ -2097,6 +2097,18 @@ describe("sync state", () => { expect(entity?.count).toBe(2) }) + test("an edit that keeps the mirror's timestamp is still recorded as updated", async () => { + const dir = mkdtempSync(path.join(SANDBOX, "memory-sync-state-edit-")) + listResponse = [note("a", "Naming conventions")] + await refresh(`${SES}-state-edit-1`, dir) + expect(await memoryState(dir)).toMatchObject({ status: "ok", changes: null }) + + listResponse = [{ ...note("a", "Naming conventions"), memory: "# Naming conventions\nnew body" }] + await refresh(`${SES}-state-edit-2`, dir) + const entity = await memoryState(dir, (e) => e.changes !== null) + expect(entity?.changes).toEqual({ added: [], removed: [], updated: ["Naming conventions"] }) + }) + test("a load for an unbound project records nothing", async () => { const dir = mkdtempSync(path.join(SANDBOX, "memory-sync-state-off-")) syncInternals.resolveBinding = async () => null diff --git a/packages/opencode/test/altimate/workspace/skill-sync.test.ts b/packages/opencode/test/altimate/workspace/skill-sync.test.ts index f8c5d8dd4b..8ee7f80bca 100644 --- a/packages/opencode/test/altimate/workspace/skill-sync.test.ts +++ b/packages/opencode/test/altimate/workspace/skill-sync.test.ts @@ -2624,6 +2624,36 @@ describe("sync state", () => { expect(entity?.changes).toEqual({ added: [], removed: [], updated: ["pub-1"] }) }) + test("taking the snapshot out of service retires its record", async () => { + serve({ "pub-1": { "SKILL.md": "one" } }) + await syncSkills(project) + expect(await read()).toMatchObject({ status: "ok", count: 1 }) + + // A pin that cannot be confirmed takes the snapshot out of service (see the pin tests above). + unbind() + globalThis.fetch = (async () => { + throw new Error("offline") + }) as unknown as typeof fetch + const pin = { + ALTIMATE_CODE_SERVE: "1", + ALTIMATE_PINNED_WORKSPACE_ID: "4243", + ALTIMATE_PINNED_WORKSPACE_NAME: "pinned", + ALTIMATE_PINNED_WORKSPACE_ROOT: project, + } + const saved = Object.fromEntries(Object.keys(pin).map((k) => [k, process.env[k]])) + Object.assign(process.env, pin) + try { + await syncSkills(project) + } finally { + for (const [k, v] of Object.entries(saved)) { + if (v === undefined) delete process.env[k] + else process.env[k] = v + } + } + expect(existsSync(skillFile("pub-1", "SKILL.md"))).toBe(false) + expect(await read()).toBeUndefined() + }) + test("a skill kept at its previous copy is not reported as updated until the update lands", async () => { serve({ "pub-1": { "SKILL.md": "one" } }, "2026-01-01T00:00:00Z") await syncSkills(project) From 04f1d55c82397ed6850f017993d2df89c3e59807 Mon Sep 17 00:00:00 2001 From: Sarav Date: Tue, 13 Oct 2026 07:51:00 +0530 Subject: [PATCH 4/4] fix(workspace): pin sync-state records to the account and time they were read - Capture the account and the fetch time before each sync reads, and drop a record whose account changed meanwhile or whose observation is older than the stored one (no rollback by a slow writer) - Run the binding fence after the account lookup, immediately before the synchronous write - Skills: drop a record only when the project is confirmed bound elsewhere or unbound; a failed lookup keeps it. Record detached from the turn (`syncStateRecorded` for tests) - Skills: retire the record only after the snapshot is actually removed, and after any pending record from an earlier run - Memory: retire the record only on a confirmed unlink or memory turned off; key items by cloud record id; version by timestamp, text and tags - `forget` takes the same fences (`isCurrent`, `before`); unlink clears the memory record - Status route returns `stale` for a last-known binding - Sidebar: no "memory loaded" line while memory is off on this machine; drop the nested change markers - Tests for each case; mutation-checked against the unfixed code Co-Authored-By: Claude Opus 5.5 (1M context) --- .../opencode/src/altimate/workspace/manage.ts | 18 ++- .../src/altimate/workspace/memory-sync.ts | 117 +++++++++++++----- .../src/altimate/workspace/skill-sync.ts | 73 +++++++++-- .../src/altimate/workspace/sync-state.ts | 94 ++++++++++---- .../plugin/tui/altimate/workspace-sidebar.tsx | 2 - packages/opencode/src/server/server.ts | 15 ++- .../test/altimate/workspace/manage.test.ts | 33 ++++- .../altimate/workspace/memory-sync.test.ts | 66 +++++++++- .../altimate/workspace/skill-sync.test.ts | 90 +++++++++++++- .../altimate/workspace/sync-state.test.ts | 54 +++++++- .../server/altimate-workspace-routes.test.ts | 41 +++++- 11 files changed, 522 insertions(+), 81 deletions(-) diff --git a/packages/opencode/src/altimate/workspace/manage.ts b/packages/opencode/src/altimate/workspace/manage.ts index 828c8f558e..da24513695 100644 --- a/packages/opencode/src/altimate/workspace/manage.ts +++ b/packages/opencode/src/altimate/workspace/manage.ts @@ -162,9 +162,12 @@ export async function status( ...(memory === "unreadable" ? { memoryUnreadable: true as const } : {}), skillsEnabled: SkillSync.isEnabled(), skillsSyncedAt: await skillsSyncedAt(directory, binding), - memoryLoadedAt: binding - ? ((await SyncState.read(directory, binding.datamateId).catch(() => null))?.entities.memory?.lastSucceededAt ?? null) - : null, + // Only while memory is on here: a load from before it was switched off is not news. + memoryLoadedAt: + binding && MemorySync.isEnabled() + ? ((await SyncState.read(directory, binding.datamateId).catch(() => null))?.entities.memory?.lastSucceededAt ?? + null) + : null, } } @@ -383,6 +386,9 @@ export interface UnlinkReport { * response means the binding is already gone server-side, which is exactly when * a stale local row most needs clearing. */ export async function unlink(directory: string): Promise { + // altimate_change start — workspace sync state + const startedAt = Date.now() + // altimate_change end // First, before any await: the row under whatever account the file belongs // to, so the cleanup can tell a file that was already another account's // from one another account wrote during the request. Taken any later and a @@ -500,6 +506,12 @@ export async function unlink(directory: string): Promise { } catch (err) { log.warn("could not reset the memory overlay after unlink", { err: String(err) }) } + // altimate_change start — workspace sync state + // The memory record described the binding just removed; a relink to the same workspace must + // not bring back its count and load time. Only a record from before this unlink began: a + // relink's own load since keeps its record. (The skill record goes with the snapshot below.) + await SyncState.forget(directory, "memory", { before: startedAt }) + // altimate_change end // Without this the workspace's skills keep loading into every session of a // project that is no longer bound to it — the snapshot lives under the // ordinary skill glob, so nothing else would stop it. diff --git a/packages/opencode/src/altimate/workspace/memory-sync.ts b/packages/opencode/src/altimate/workspace/memory-sync.ts index eef8c139ac..6302d8bdc7 100644 --- a/packages/opencode/src/altimate/workspace/memory-sync.ts +++ b/packages/opencode/src/altimate/workspace/memory-sync.ts @@ -24,7 +24,15 @@ import { Log } from "@/altimate/util/log" import type { MemoryBlock } from "@/memory/types" import { TRAINING_META_COMMENT } from "@/altimate/training/types" // Aliased: `syncInternals.resolveBinding` below is an unrelated test seam. -import { canonicalDirectory, onBindingChanged, resolveBinding as resolveProjectBinding, type CachedBinding } from "./state" +import { + canonicalDirectory, + currentScope, + onBindingChanged, + resolveBinding as resolveProjectBinding, + resolveBindingOutcome, + type AccountKey, + type CachedBinding, +} from "./state" import { indexKey, readIndex, readIndexEntry, recordIndexEntry } from "./memory-index" import { WorkspaceApi } from "./api-client" import { AltimateApi } from "@/altimate/api/client" @@ -146,6 +154,30 @@ export function projectKeyFor(binding: CachedBinding): string { * and archives run after their writing context has gone, so resolving the * binding from the ambient instance can attribute one project's memory to * another project's workspace. */ +// altimate_change start — workspace sync state +/** `currentBinding`, saying whether "no binding" was confirmed. The load reports both as + * unlinked, but only a confirmed unlink may retire the recorded sync state: a lookup that + * failed says nothing about the link. */ +async function currentBindingOutcome( + directory?: string, +): Promise<{ binding: CachedBinding | null; confirmed: boolean }> { + if (syncInternals.resolveBinding) { + const binding = await syncInternals.resolveBinding() + return { binding, confirmed: true } + } + directory = directory ?? currentDirectory() ?? undefined + if (!directory) return { binding: null, confirmed: false } + try { + const outcome = await resolveBindingOutcome(directory) + if (outcome.status === "bound") return { binding: outcome.binding, confirmed: true } + return { binding: null, confirmed: outcome.status === "unbound" && outcome.stale !== true } + } catch (err) { + log.warn("could not resolve binding for memory mirror", { err: String(err) }) + return { binding: null, confirmed: false } + } +} +// altimate_change end + async function currentBinding(directory?: string): Promise { if (syncInternals.resolveBinding) return syncInternals.resolveBinding() directory = directory ?? currentDirectory() ?? undefined @@ -1372,11 +1404,24 @@ export async function whenHydrated( * them is how a transient failure gets reported as a successful reload of an * empty workspace, taking the session's real memory with it. */ type LoadOutcome = ( - | { status: "loaded"; blocks: RemoteMemoryBlock[] } - | { status: "unlinked" } + | { + status: "loaded" + blocks: RemoteMemoryBlock[] + /** The sync-state items for `blocks`, keyed by cloud record id. */ + items: Record + } + | { status: "unlinked"; /** The server confirmed it, rather than a lookup failing. */ confirmed: boolean } | { status: "disabled" } | { status: "error" } -) & { epoch?: string; dir?: string | null; /** The workspace the load read, once resolved. */ datamateId?: number } +) & { + epoch?: string + dir?: string | null + /** The workspace the load read, once resolved. */ + datamateId?: number + /** The account and the time the load started reading, for the sync-state record. */ + scope?: AccountKey | null + observedAt?: number +} /** Read this project's workspace memory. Pure: it publishes nothing, so a slow * load that has been superseded cannot write over a newer result. */ @@ -1388,28 +1433,34 @@ async function loadWorkspaceMemory(directory?: string): Promise { let vouched = epochFor(dir) // The workspace this load resolved, so a failure past that point is recorded against it. let resolved: number | undefined + // Taken before anything is read: the record must name the account and time the items came + // from, not whichever is current when it is written. + const observedAt = Date.now() + const scope = await currentScope().catch(() => null) + const stamp = { scope, observedAt } try { // The epoch must bracket the lookup: read only after it, a relink that lands while the // lookup is pending would stamp the old binding as current. Read only before it, a lookup // that adopts this project's server binding (which notifies a change) would stamp itself // stale. So resolve until the epoch holds across one lookup; the retry is a cache read. let binding: CachedBinding | null = null + let confirmed = false let epoch = "" let stable = false for (let attempt = 0; attempt < 3 && !stable; attempt++) { const before = epochFor(dir) - binding = await currentBinding(raw ?? undefined) + ;({ binding, confirmed } = await currentBindingOutcome(raw ?? undefined)) epoch = epochFor(dir) stable = before === epoch } vouched = epoch - if (!stable) return { status: "error", epoch, dir } - if (!binding) return { status: "unlinked", epoch, dir } + if (!stable) return { status: "error", epoch, dir, ...stamp } + if (!binding) return { status: "unlinked", confirmed, epoch, dir, ...stamp } const datamateId = binding.datamateId resolved = datamateId const enabled = await memoryStatus(binding) - if (enabled === "error") return { status: "error", epoch, dir, datamateId } - if (enabled === "disabled") return { status: "disabled", epoch, dir, datamateId } + if (enabled === "error") return { status: "error", epoch, dir, datamateId, ...stamp } + if (enabled === "disabled") return { status: "disabled", epoch, dir, datamateId, ...stamp } const ownProjectKey = projectKeyFor(binding) const ownWorkspace = String(binding.datamateId) @@ -1434,6 +1485,7 @@ async function loadWorkspaceMemory(directory?: string): Promise { ) const blocks: RemoteMemoryBlock[] = [] + const items: Record = {} for (const record of records) { if (!record?.id) continue if (!isMirrorRecord(record) || isArchived(record)) continue @@ -1444,13 +1496,14 @@ async function loadWorkspaceMemory(directory?: string): Promise { // 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) + items[record.id] = memoryItem(block) } - return { status: "loaded", blocks, epoch, dir, datamateId } + return { status: "loaded", blocks, items, epoch, dir, datamateId, ...stamp } } catch (err) { log.warn("workspace memory load failed", { err: String(err) }) // Stamped like any other outcome: without an epoch the session would reload (and make // the prompt wait) on every turn for as long as the service is down. - return { status: "error", epoch: vouched, dir, datamateId: resolved } + return { status: "error", epoch: vouched, dir, datamateId: resolved, ...stamp } } } @@ -1481,38 +1534,42 @@ function commitLoad(sessionID: string, state: SessionMemory, outcome: LoadOutcom * project, so a second session reading the same blocks is not a change. * * Detached: nothing waits on status metadata. Fenced on the load's binding epoch at write - * time, so a record that settles after a relink cannot replace the new workspace's state. - * Memory found off, or the project unlinked, clears the record rather than leaving an - * earlier error latched. A load that never resolved a workspace records nothing. */ + * time, so a record that settles after a relink cannot replace the new workspace's state, and + * stamped with the account and time the load read under. Memory found off, or an unlink the + * server confirmed, clears the record rather than leaving an earlier error latched; a lookup + * that failed clears nothing. A load that never resolved a workspace records nothing. */ function recordSyncState(outcome: LoadOutcome): void { const dir = outcome.dir if (!dir) return const isCurrent = () => epochFor(dir) === outcome.epoch + const retire = outcome.status === "disabled" || (outcome.status === "unlinked" && outcome.confirmed) + if (outcome.status === "unlinked" && !retire) return void import("./sync-state") .then((SyncState) => { - if (outcome.status === "unlinked" || outcome.status === "disabled") { - return isCurrent() ? SyncState.forget(dir, "memory") : undefined - } + if (retire) return SyncState.forget(dir, "memory", { isCurrent, before: outcome.observedAt }) if (outcome.datamateId === undefined) return const result = - outcome.status === "loaded" - ? { items: Object.fromEntries(outcome.blocks.map((block) => [memoryItemKey(block), memoryItem(block)])) } - : { error: "could not load workspace memory" } - return SyncState.record(dir, outcome.datamateId, "memory", result, { isCurrent }) + outcome.status === "loaded" ? { items: outcome.items } : { error: "could not load workspace memory" } + return SyncState.record(dir, outcome.datamateId, "memory", result, { + isCurrent, + scope: outcome.scope, + observedAt: outcome.observedAt, + }) }) .catch((err) => log.warn("could not record the memory sync state", { err: String(err) })) } -/** Block ids are unique per file, not across scopes or across the teammates whose blocks a - * workspace serves, so the key carries both. */ -function memoryItemKey(block: RemoteMemoryBlock): string { - return `${block.scope}:${block.origin ?? "own"}:${block.id}` -} - -/** Versioned by content as well as the mirror's timestamp: an edit made in the web app can keep - * the block's original `block_updated`, and the session still loads the new text. */ +/** A block's sync-state item. Keyed by its cloud record id (see `loadWorkspaceMemory`): block + * ids repeat across teammates and sibling projects. Versioned by everything the session + * loads from it — the mirror's timestamp, the text and the tags — since a web edit can change + * the text or tags and keep the original `block_updated`. */ function memoryItem(block: RemoteMemoryBlock): { label: string; version: string } { - const digest = createHash("sha256").update(block.content).digest("hex").slice(0, 16) + const digest = createHash("sha256") + .update(block.content) + .update("\0") + .update([...block.tags].sort().join("\0")) + .digest("hex") + .slice(0, 16) return { label: blockTitle(block), version: `${block.updated ?? ""}:${digest}` } } // altimate_change end diff --git a/packages/opencode/src/altimate/workspace/skill-sync.ts b/packages/opencode/src/altimate/workspace/skill-sync.ts index 0bb6aafe5b..e7af76124a 100644 --- a/packages/opencode/src/altimate/workspace/skill-sync.ts +++ b/packages/opencode/src/altimate/workspace/skill-sync.ts @@ -45,7 +45,14 @@ import { Global } from "@/global" import { Flag as CoreFlag } from "@opencode-ai/core/flag/flag" import { Log } from "@/altimate/util/log" import { AltimateApi } from "@/altimate/api/client" -import { credentialDigest, readLocalBinding, resolveBindingOutcome, type CachedBinding } from "./state" +import { + credentialDigest, + currentScope, + readLocalBinding, + resolveBindingOutcome, + type AccountKey, + type CachedBinding, +} from "./state" import { altimateRequest, WorkspaceApiError } from "./api-client" import { readPin, resolveWithinRoot } from "./pin" import * as PendingTurns from "./pending-turns" @@ -247,6 +254,13 @@ export function describeSyncProblems(result: SyncResult): { title: string; messa } // altimate_change start — workspace sync state +const pendingRecords = new Map>() + +/** Settles once the sync-state record of the last run for `directory` has been written. */ +export function syncStateRecorded(directory: string): Promise { + return pendingRecords.get(path.resolve(directory)) ?? Promise.resolve() +} + /** Record this run in the workspace sync state. `rows` is the list the published snapshot * describes, or null when the run left the snapshot as it was — the items are then * unknown to this run and the previous record stands. Imported lazily: the state module @@ -258,6 +272,7 @@ async function recordSyncState( rows: RemoteSummary[] | null, dropped: Set, carried: Map, + stamp: { scope: AccountKey | null; observedAt: number }, ): Promise { const problem = describeSyncProblems(result) // A carried skill is recorded at the version the snapshot still holds, not the one the list @@ -273,10 +288,13 @@ async function recordSyncState( ) : undefined // A sync can outlast a relink. Its result describes the workspace it read, so it is dropped - // rather than written over the state of the one the project is bound to now. + // when the project is now known to be elsewhere: bound to another workspace, or unbound. A + // lookup that cannot answer keeps it — the run resolved this binding moments ago, and a blip + // is no evidence it changed. const isCurrent = async () => { const now = await resolveBindingOutcome(directory).catch(() => null) - return now?.status === "bound" && now.binding.datamateId === datamateId + if (now?.status === "bound") return now.binding.datamateId === datamateId + return now?.status !== "unbound" } try { const SyncState = await import("./sync-state") @@ -288,7 +306,7 @@ async function recordSyncState( ...(items ? { items } : {}), ...(problem ? { error: `${problem.title}: ${problem.message.split("\n").join("; ")}` } : {}), }, - { isCurrent }, + { isCurrent, scope: stamp.scope, observedAt: stamp.observedAt }, ) } catch (err) { // Status metadata: never the reason a sync fails. @@ -1085,24 +1103,26 @@ async function deactivate(directory: string, why: string): Promise { // An empty workspace leaves no tree, so its "known empty" record and sync stamps are cleared // here, before the early return: relinking must sync again rather than repeat a stale "none". const canon = path.resolve(directory) - await clearEmptyRecord(canon) // altimate_change start — workspace sync state - // The snapshot is out of service, so its record no longer describes anything here: a relink to - // the same workspace starts from a fresh baseline instead of the old count. - await import("./sync-state") - .then((SyncState) => SyncState.forget(canon, "skills")) - .catch((err) => log.warn("could not clear the skill sync state", { err: String(err) })) + const startedAt = Date.now() // altimate_change end + await clearEmptyRecord(canon) try { await fs.stat(root) } catch { lastSyncedAt.delete(canon) syncedFor.delete(canon) + // altimate_change start — workspace sync state + await retireSyncState(canon, startedAt) + // altimate_change end return false // nothing published here } if (!(await ownsManagedDir(directory))) return false await removeManaged(directory) await sweepStaging(directory) + // altimate_change start — workspace sync state + await retireSyncState(canon, startedAt) + // altimate_change end // The tree is gone, so the stamps that say "recently synced" no longer // describe anything. Left behind, a purge that is not followed by a // successful sync — an account switch to a project this user has not bound — @@ -1113,6 +1133,20 @@ async function deactivate(directory: string, why: string): Promise { return true } +// altimate_change start — workspace sync state +/** The snapshot is gone, so its record describes nothing here: a relink to the same workspace + * starts from a fresh baseline rather than the old count. Only after the removal succeeded (a + * refused or failed one leaves the snapshot, and its record, in place), and only a record + * observed before this deactivation began — a sync for a newer binding keeps its own. */ +async function retireSyncState(canon: string, before: number): Promise { + // A detached record from an earlier run lands first, so it cannot bring the record back. + await pendingRecords.get(canon) + await import("./sync-state") + .then((SyncState) => SyncState.forget(canon, "skills", { before })) + .catch((err) => log.warn("could not clear the skill sync state", { err: String(err) })) +} +// altimate_change end + async function removeManaged(directory: string): Promise { await fs.rm(managedRoot(directory), { recursive: true, force: true }) } @@ -1188,6 +1222,9 @@ export async function syncSkills(directory: string): Promise { /** Skills kept at their previous copy, mapped to that copy's version. */ const carriedVersions = new Map() let published = false + /** When this run started reading, and under which account (set once credentials are known). */ + const observedAt = Date.now() + let runScope: AccountKey | null = null // altimate_change end const run = (async () => { // Checked BEFORE the binding: `resolveBinding` needs credentials too, so a @@ -1223,6 +1260,9 @@ export async function syncSkills(directory: string): Promise { // to an account with no binding here left the previous tenant's skills on // disk and loading into prompts, with every retry hitting the same return. const credsForPurge = await AltimateApi.getCredentials().catch(() => null) + // altimate_change start — workspace sync state + runScope = await currentScope().catch(() => null) + // altimate_change end // A credential with no key names no account, so nothing here can be judged // foreign — and the sync below will fail for the same reason, so there is // no replacement coming. Purging on it deleted a good snapshot and put @@ -1656,14 +1696,25 @@ export async function syncSkills(directory: string): Promise { if (skippedSkills.length === 0 && !syncError) store.announced.delete(canon) const result: SyncResult = { changed, skipped: skippedSkills, error: syncError } // altimate_change start — workspace sync state - if (checkedDatamateId !== undefined) await recordSyncState( + // Detached: the turn waits on the sync, not on its status metadata. + if (checkedDatamateId !== undefined) { + const recorded = recordSyncState( canon, checkedDatamateId, result, published ? remoteRows : null, dropped, carriedVersions, + { + scope: runScope, + observedAt, + }, ) + pendingRecords.set(canon, recorded) + void recorded.finally(() => { + if (pendingRecords.get(canon) === recorded) pendingRecords.delete(canon) + }) + } // altimate_change end return result })() diff --git a/packages/opencode/src/altimate/workspace/sync-state.ts b/packages/opencode/src/altimate/workspace/sync-state.ts index de70659797..9e72ebc77b 100644 --- a/packages/opencode/src/altimate/workspace/sync-state.ts +++ b/packages/opencode/src/altimate/workspace/sync-state.ts @@ -14,9 +14,11 @@ // // One file per kind, so two writers of different kinds — the TUI main thread, the server // worker and a separate CLI process can each hold one — never read-modify-write the same -// file. Writes of one kind within this module realm are chained; across realms the later -// rename wins, which loses at most the earlier check of that same kind, and the next check -// rewrites it. +// file. Writes of one kind within this module realm are chained. Across realms, each record +// carries when its check read the items (`observedAt`, taken before the fetch), and a write +// whose observation is older than the stored one is dropped, so a slow sync cannot roll a +// newer record back. That compare is not atomic across processes: two writes landing in the +// same instant can still both pass it, and the later rename wins until the next check. // // Event delivery: `Bus.publish` reaches the bus of the instance the record runs in. That is // the `/event` stream for `serve` (the IDE extension) and for the TUI's per-turn syncs, which @@ -31,7 +33,7 @@ import { Bus } from "@/bus" import { BusEvent } from "@/bus/bus-event" import { Global } from "@/global" import { Log } from "@/altimate/util/log" -import { canonicalDirectory, currentScope } from "./state" +import { canonicalDirectory, currentScope, type AccountKey } from "./state" const log = Log.create({ service: "altimate-workspace-sync-state" }) @@ -55,10 +57,11 @@ const EntityState = z.object({ status: z.enum(["ok", "error"]), /** Items held after the last check; null when it has never been read successfully. */ count: z.number().int().nonnegative().nullable(), - /** Epoch ms of the last completed check, successful or not. */ + /** Epoch ms at which the last recorded check started reading, successful or not. */ lastCheckedAt: z.number(), - /** Epoch ms of the last check that read the items successfully; null when none has. A - * "loaded Xm ago" line reads this, not `lastCheckedAt`, which also moves on failures. */ + /** Epoch ms at which the last check that read the items successfully started; null when + * none has. A "loaded Xm ago" line reads this, not `lastCheckedAt`, which also moves on + * failures. */ lastSucceededAt: z.number().nullable(), /** Epoch ms of the last check that found a difference; null when none has. */ lastChangedAt: z.number().nullable(), @@ -94,6 +97,8 @@ export interface WorkspaceSyncState { interface StoredEntity extends EntityState { /** Absent until a check has read the items, so the first read is a baseline. */ items?: SyncItems + /** When the check that wrote this record started reading; see the header. */ + observedAt?: number } interface StoredFile { @@ -115,10 +120,13 @@ function filePath(directory: string, kind: SyncKind): string { return path.join(syncDir(), `${key}.${kind}.json`) } +function scopeKeyOf(scope: AccountKey | null): string | null { + return scope ? `${scope.tenant}|${scope.apiUrl}|${scope.account}` : null +} + /** The account the caller's credentials name, as a string, or null when there is none. */ async function scopeKey(): Promise { - const scope = await currentScope().catch(() => null) - return scope ? `${scope.tenant}|${scope.apiUrl}|${scope.account}` : null + return scopeKeyOf(await currentScope().catch(() => null)) } function readStored(directory: string, kind: SyncKind): StoredFile | null { @@ -160,8 +168,7 @@ export async function read(directory: string, datamateId: number): Promise 0 ? { datamateId, entities } : null } @@ -181,6 +188,11 @@ export function diff(previous: SyncItems, next: SyncItems): SyncChanges { return { added: added.sort(), removed: removed.sort(), updated: updated.sort() } } +function publicState(entity: StoredEntity): EntityState { + const { items: _items, observedAt: _observedAt, ...state } = entity + return state +} + function isEmpty(changes: SyncChanges): boolean { return changes.added.length === 0 && changes.removed.length === 0 && changes.updated.length === 0 } @@ -198,6 +210,22 @@ export interface RecordOptions { * the write: a long sync, or a detached memory record, can settle after the project was * rebound, and its result must not replace the new workspace's state. */ isCurrent?: () => boolean | Promise + /** The account the check read under, captured before it fetched. Workspace ids are per + * tenant, so a check that started under another account describes another workspace even + * when the id matches, and is dropped. Null (no account then) is never recorded. */ + scope?: AccountKey | null + /** When the check started reading (epoch ms); defaults to now. Also the record's + * `lastCheckedAt`, and its `lastSucceededAt` when it read the items. */ + observedAt?: number +} + +export interface ForgetOptions { + /** Whether the binding the caller is retiring is still the project's. */ + isCurrent?: () => boolean | Promise + /** Only a record observed no later than this time (epoch ms) is removed: one observed + * since describes a binding made after the caller decided to retire the old one. A tie is + * removed, since the caller decided after that observation. */ + before?: number } /** Writes are chained per file within this module realm; see the header for other writers. */ @@ -217,11 +245,24 @@ export function record( return chain(filePath(directory, kind), () => apply(directory, datamateId, kind, outcome, options), kind) } -/** Drop a kind's record, without an event: memory turned off, or the project unlinked. A - * problem recorded before would otherwise stay latched for as long as nothing checks it. */ -export function forget(directory: string, kind: SyncKind): Promise { +/** Drop a kind's record, without an event: memory turned off, the project unlinked, or the + * skill snapshot taken out of service. A problem recorded before would otherwise stay + * latched for as long as nothing checks it, and a relink to the same workspace would bring + * back a count and a load time that describe nothing on disk. */ +export function forget(directory: string, kind: SyncKind, options: ForgetOptions = {}): Promise { const file = filePath(directory, kind) - return chain(file, async () => rmSync(file, { force: true }), kind) + return chain( + file, + async () => { + if (options.isCurrent && !(await options.isCurrent())) return + const stored = readStored(directory, kind) + if (!stored) return + const observed = stored.entity.observedAt ?? stored.entity.lastCheckedAt + if (options.before !== undefined && observed > options.before) return + rmSync(file, { force: true }) + }, + kind, + ) } function chain(file: string, work: () => Promise, kind: SyncKind): Promise { @@ -241,19 +282,29 @@ async function apply( outcome: Outcome, options: RecordOptions, ): Promise { - if (options.isCurrent && !(await options.isCurrent())) { - log.info("dropped a sync result for a binding that is no longer current", { kind, datamateId }) - return - } const scope = await scopeKey() // No account, no attribution: a record that cannot say whose it is would be served to // whichever account looks next. if (!scope) return + if (options.scope !== undefined && scopeKeyOf(options.scope) !== scope) { + log.info("dropped a sync result read under another account", { kind, datamateId }) + return + } + // Last before the read-modify-write, which is synchronous from here on: nothing can rebind + // between this check and the rename. + if (options.isCurrent && !(await options.isCurrent())) { + log.info("dropped a sync result for a binding that is no longer current", { kind, datamateId }) + return + } const canon = canonicalDirectory(directory) const stored = readStored(directory, kind) // Another workspace's (or account's) state describes nothing about this one: start over. const previous = stored && stored.datamateId === datamateId && stored.scope === scope ? stored.entity : undefined - const now = Date.now() + const now = options.observedAt ?? Date.now() + if (previous?.observedAt !== undefined && previous.observedAt > now) { + log.info("dropped a sync result older than the recorded one", { kind, datamateId }) + return + } let changes: SyncChanges | null = null if (outcome.items) { @@ -275,13 +326,14 @@ async function apply( changes: changes ?? previous?.changes ?? null, ...(outcome.error ? { error: outcome.error } : {}), ...(items ? { items } : {}), + observedAt: now, } const file: StoredFile = { version: 2, directory: canon, scope, datamateId, entity: next } writePrivate(filePath(directory, kind), file) const errorChanged = (previous?.status ?? "ok") !== next.status || previous?.error !== next.error if (!changes && !errorChanged) return - const { items: _items, ...state } = next + const state = publicState(next) // Needs an instance (a turn, a server route). A sync started outside one — a bind from // the CLI — has no client listening anyway; the state above is what it leaves behind. await Bus.publish(Event.Changed, { directory: canon, datamateId, kind, state, changes }).catch((err) => diff --git a/packages/opencode/src/plugin/tui/altimate/workspace-sidebar.tsx b/packages/opencode/src/plugin/tui/altimate/workspace-sidebar.tsx index 069bd54b61..4ee9b859b3 100644 --- a/packages/opencode/src/plugin/tui/altimate/workspace-sidebar.tsx +++ b/packages/opencode/src/plugin/tui/altimate/workspace-sidebar.tsx @@ -307,12 +307,10 @@ function View(props: { api: TuiPluginApi }) { {(at) => {`skills synced ${describeAge(at())}`}} - {/* altimate_change start — workspace sync state */} {(at) => {`memory loaded ${describeAge(at())}`}} {/* altimate_change end */} - {/* altimate_change end */} {/* ``pinned via --workspace`` means "this SESSION was launched * with --workspace and it resolved to this id". It does NOT * mean "the current binding was set by --workspace" — if the diff --git a/packages/opencode/src/server/server.ts b/packages/opencode/src/server/server.ts index 01cd46e6d5..7ea0a48a54 100644 --- a/packages/opencode/src/server/server.ts +++ b/packages/opencode/src/server/server.ts @@ -1202,7 +1202,9 @@ export namespace Server { // // `binding` says why there may be no state: `unbound` is a project with no workspace, // `unknown` one whose link could not be confirmed (no credentials, an invalid pin, the - // service unreachable) — which a client must not show as "not linked". + // service unreachable) — which a client must not show as "not linked". `stale` marks a + // last-known answer (`bound` or `unbound`) the server could not confirm just now, for a + // client to show as last known. .get("/altimate/workspace/status", async (c) => { const refused = workspaceRouteRefusal( c.req.header("origin"), @@ -1215,12 +1217,19 @@ export namespace Server { const { resolveBindingOutcome } = await import("../altimate/workspace/state") const SyncState = await import("../altimate/workspace/sync-state") const outcome = await resolveBindingOutcome(Instance.directory) + const stale = "stale" in outcome && outcome.stale === true if (outcome.status !== "bound") { - return c.json({ ok: true as const, binding: outcome.status, datamateId: null, entities: {} }) + return c.json({ ok: true as const, binding: outcome.status, stale, datamateId: null, entities: {} }) } const datamateId = outcome.binding.datamateId const state = await SyncState.read(Instance.directory, datamateId) - return c.json({ ok: true as const, binding: "bound" as const, datamateId, entities: state?.entities ?? {} }) + return c.json({ + ok: true as const, + binding: "bound" as const, + stale, + datamateId, + entities: state?.entities ?? {}, + }) } catch (err) { const error = err instanceof Error ? err.message : String(err) log.error("workspace status: failed", { error }) diff --git a/packages/opencode/test/altimate/workspace/manage.test.ts b/packages/opencode/test/altimate/workspace/manage.test.ts index e0f8f6b527..4ce46e821d 100644 --- a/packages/opencode/test/altimate/workspace/manage.test.ts +++ b/packages/opencode/test/altimate/workspace/manage.test.ts @@ -13,7 +13,7 @@ // stubbed at `globalThis.fetch` so assertions are about the requests actually // issued — method, path, query — and the binding cache is a real file in a real // sandbox directory. -import { afterAll, afterEach, beforeEach, describe, expect, test } from "bun:test" +import { afterAll, afterEach, beforeEach, describe, expect, spyOn, test } from "bun:test" import { chmodSync, existsSync, mkdirSync, mkdtempSync, readFileSync, realpathSync, rmSync, writeFileSync } from "node:fs" import { execFileSync } from "node:child_process" import path from "node:path" @@ -202,6 +202,37 @@ describe("sync", () => { }) }) +describe("workspace sync state", () => { + const items = { "rec-1": { label: "Naming conventions", version: "v1" } } + + test("unlink clears the memory record, so a relink does not bring back its load time", async () => { + const SyncState = await import("../../../src/altimate/workspace/sync-state") + await bind(projectDir) + await SyncState.record(projectDir, 42, "memory", { items }, { observedAt: Date.now() - 1_000 }) + expect((await SyncState.read(projectDir, 42))?.entities.memory?.count).toBe(1) + + await unlink(projectDir) + + expect(await SyncState.read(projectDir, 42)).toBeNull() + }) + + test("status reports when memory last loaded, and nothing while memory is off here", async () => { + const SyncState = await import("../../../src/altimate/workspace/sync-state") + const MemorySync = await import("../../../src/altimate/workspace/memory-sync") + await bind(projectDir) + await SyncState.record(projectDir, 42, "memory", { items }, { observedAt: 1_000 }) + + expect((await status(projectDir)).memoryLoadedAt).toBe(1_000) + + const enabled = spyOn(MemorySync, "isEnabled").mockReturnValue(false) + try { + expect((await status(projectDir)).memoryLoadedAt).toBeNull() + } finally { + enabled.mockRestore() + } + }) +}) + describe("status", () => { test("reports the binding a project is linked to", async () => { await bind(projectDir) diff --git a/packages/opencode/test/altimate/workspace/memory-sync.test.ts b/packages/opencode/test/altimate/workspace/memory-sync.test.ts index f6174f523c..b32598d174 100644 --- a/packages/opencode/test/altimate/workspace/memory-sync.test.ts +++ b/packages/opencode/test/altimate/workspace/memory-sync.test.ts @@ -2109,6 +2109,65 @@ describe("sync state", () => { expect(entity?.changes).toEqual({ added: [], removed: [], updated: ["Naming conventions"] }) }) + test("a tags-only edit is recorded as updated", async () => { + const dir = mkdtempSync(path.join(SANDBOX, "memory-sync-state-tags-")) + listResponse = [note("a", "Naming conventions")] + await refresh(`${SES}-state-tags-1`, dir) + expect(await memoryState(dir)).toMatchObject({ status: "ok", changes: null }) + + const tagged = note("a", "Naming conventions") + listResponse = [{ ...tagged, metadata: { ...tagged.metadata, block_tags: JSON.stringify(["sql"]) } }] + await refresh(`${SES}-state-tags-2`, dir) + const entity = await memoryState(dir, (e) => e.changes !== null) + expect(entity?.changes).toEqual({ added: [], removed: [], updated: ["Naming conventions"] }) + }) + + test("two teammates' blocks with the same block id are two items", async () => { + const dir = mkdtempSync(path.join(SANDBOX, "memory-sync-state-teammates-")) + listResponse = [note("same", "Mine"), { ...note("same", "Theirs"), id: "mem-same-teammate" }] + await refresh(`${SES}-state-teammates`, dir) + expect(await memoryState(dir)).toMatchObject({ status: "ok", count: 2 }) + }) + + test("a binding lookup that fails does not clear the record", async () => { + const dir = mkdtempSync(path.join(SANDBOX, "memory-sync-state-blip-")) + listResponse = [note("a", "Naming conventions")] + await refresh(`${SES}-state-blip-1`, dir) + expect(await memoryState(dir)).toMatchObject({ status: "ok", count: 1 }) + + // No local row, and the server cannot be asked: the load reads as unlinked, unconfirmed. + delete syncInternals.resolveBinding + const inner = globalThis.fetch + globalThis.fetch = (async (input: any, init?: any) => { + if (String(input).includes("/datamate-project-bindings/")) throw new Error("offline") + return inner(input, init) + }) as typeof fetch + const result = await refresh(`${SES}-state-blip-2`, dir) + expect(result.status).toBe("unlinked") + await new Promise((r) => setTimeout(r, 50)) + + syncInternals.resolveBinding = async () => BINDING as any + expect(await memoryState(dir)).toMatchObject({ status: "ok", count: 1 }) + }) + + test("a load that started under another account is not recorded", async () => { + const SyncState = await import("../../../src/altimate/workspace/sync-state") + const dir = mkdtempSync(path.join(SANDBOX, "memory-sync-state-switch-")) + listResponse = [note("a", "Their block")] + const inner = globalThis.fetch + globalThis.fetch = (async (input: any, init?: any) => { + // The credentials change while the list is in flight. + if (String(input).includes("/datamates/memory/list")) stubCreds("other", "https://api.example.com", "key-b") + return inner(input, init) + }) as typeof fetch + await refresh(`${SES}-state-switch`, dir) + await new Promise((r) => setTimeout(r, 50)) + + expect(await SyncState.read(dir, BINDING.datamateId)).toBeNull() + stubCreds("acme", "https://api.example.com") + expect(await SyncState.read(dir, BINDING.datamateId)).toBeNull() + }) + test("a load for an unbound project records nothing", async () => { const dir = mkdtempSync(path.join(SANDBOX, "memory-sync-state-off-")) syncInternals.resolveBinding = async () => null @@ -2149,8 +2208,11 @@ describe("sync state", () => { workspaces = [{ id: 42, name: "acme", memory_enabled: false }] resetOverlay() await refresh(`${SES}-state-disabled-2`, dir) - await settled(dir) - expect(await memoryState(dir)).toBeUndefined() + const SyncState = await import("../../../src/altimate/workspace/sync-state") + for (let i = 0; i < 50 && (await SyncState.read(dir, BINDING.datamateId)); i++) { + await new Promise((r) => setTimeout(r, 10)) + } + expect(await SyncState.read(dir, BINDING.datamateId)).toBeNull() }) test("a record that settles after a relink does not replace the new binding's state", async () => { diff --git a/packages/opencode/test/altimate/workspace/skill-sync.test.ts b/packages/opencode/test/altimate/workspace/skill-sync.test.ts index 8ee7f80bca..ed73991ea9 100644 --- a/packages/opencode/test/altimate/workspace/skill-sync.test.ts +++ b/packages/opencode/test/altimate/workspace/skill-sync.test.ts @@ -67,6 +67,7 @@ const { purgeManagedSnapshot, snapshotKnownEmpty, snapshotWorkspaceId, + syncStateRecorded, } = await import("@/altimate/workspace/skill-sync") const { cachePath, recordApprovedBinding, credentialDigest } = await import("@/altimate/workspace/state") @@ -120,7 +121,10 @@ beforeEach(() => { bindTo(1) }) -afterEach(() => { +afterEach(async () => { + // A run's sync-state record is detached and can still be resolving the binding; let it land + // before the next test swaps `fetch`, or that test counts its requests. + await syncStateRecorded(project) globalThis.fetch = ORIGINAL_FETCH if (ORIGINAL_WORKSPACE_FLAG === undefined) delete process.env.ALTIMATE_DISABLE_WORKSPACE else process.env.ALTIMATE_DISABLE_WORKSPACE = ORIGINAL_WORKSPACE_FLAG @@ -2572,8 +2576,11 @@ describe("shouldAnnounce", () => { // altimate_change start — workspace sync state describe("sync state", () => { - const read = async () => - (await (await import("@/altimate/workspace/sync-state")).read(project, 1))?.entities.skills + // The record is written detached from the sync; wait for it before reading. + const read = async () => { + await (await import("@/altimate/workspace/skill-sync")).syncStateRecorded(project) + return (await (await import("@/altimate/workspace/sync-state")).read(project, 1))?.entities.skills + } test("the first sync is a baseline and a later one records what changed, by name", async () => { const names = { "pub-1": "revenue-model", "pub-2": "cost-alerts" } @@ -2654,6 +2661,83 @@ describe("sync state", () => { expect(await read()).toBeUndefined() }) + /** Runs `during` when the run fetches the workspace's skill list, then answers as before. */ + const onList = (during: () => void) => { + const inner = globalThis.fetch + globalThis.fetch = (async (input: string | URL, init?: RequestInit) => { + if (String(input).includes("datamate_id")) during() + return inner(input as never, init as never) + }) as unknown as typeof fetch + } + + test("a sync that started under another account is not recorded", async () => { + const credsFile = path.join(SANDBOX, "home", ".altimate", "altimate.json") + const original = readFileSync(credsFile, "utf8") + serve({ "pub-1": { "SKILL.md": "one" } }) + onList(() => + writeFileSync( + credsFile, + JSON.stringify({ altimateUrl: API_URL, altimateInstanceName: TENANT, altimateApiKey: "another-key" }), + ), + ) + try { + await syncSkills(project) + expect(await read()).toBeUndefined() + } finally { + writeFileSync(credsFile, original) + } + expect(await read()).toBeUndefined() + }) + + test("a sync whose project was rebound meanwhile is not recorded", async () => { + serve({ "pub-1": { "SKILL.md": "one" } }) + await syncSkills(project) + const before = await read() + + serve({ "pub-1": { "SKILL.md": "one" }, "pub-2": { "SKILL.md": "two" } }) + onList(() => bindTo(2)) + await syncSkills(project) + + // Neither workspace's record took the run's result. + expect(await read()).toEqual(before) + const SyncState = await import("@/altimate/workspace/sync-state") + expect(await SyncState.read(project, 2)).toBeNull() + }) + + test("a binding lookup that fails after the sync does not drop its record", async () => { + serve({ "pub-1": { "SKILL.md": "one" } }) + let offline = false + const inner = globalThis.fetch + globalThis.fetch = (async (input: string | URL, init?: RequestInit) => { + const url = String(input) + if (url.includes("datamate_id")) { + // By the time the record is written there is no local row, and the server cannot be + // asked: the binding resolves as unknown. + unbind() + offline = true + } else if (offline && url.includes("/datamate-project-bindings/")) throw new Error("offline") + return inner(input as never, init as never) + }) as unknown as typeof fetch + await syncSkills(project) + expect(await read()).toMatchObject({ status: "ok", count: 1 }) + }) + + test("a snapshot that could not be removed keeps its record", async () => { + serve({ "pub-1": { "SKILL.md": "one" } }) + await syncSkills(project) + // A manifest this client cannot read as its own: the snapshot is not removed. + const manifest = path.join(project, MANAGED, ".manifest.json") + const original = readFileSync(manifest, "utf8") + writeFileSync(manifest, "not a manifest") + + expect(await purgeManagedSnapshot(project, "test")).toBe("refused") + expect(await read()).toMatchObject({ status: "ok", count: 1 }) + + writeFileSync(manifest, original) + expect(await purgeManagedSnapshot(project, "test")).toBe("removed") + expect(await read()).toBeUndefined() + }) + test("a skill kept at its previous copy is not reported as updated until the update lands", async () => { serve({ "pub-1": { "SKILL.md": "one" } }, "2026-01-01T00:00:00Z") await syncSkills(project) diff --git a/packages/opencode/test/altimate/workspace/sync-state.test.ts b/packages/opencode/test/altimate/workspace/sync-state.test.ts index 52d9439491..c4e275e6d9 100644 --- a/packages/opencode/test/altimate/workspace/sync-state.test.ts +++ b/packages/opencode/test/altimate/workspace/sync-state.test.ts @@ -175,8 +175,11 @@ describe("record", () => { await SyncState.record(project, 7, "skills", { items: { a: skill("Alpha") } }) const files = syncFiles() expect(files.some((name) => name.includes(".tmp-"))).toBe(false) - for (const name of files) { - expect(statSync(path.join(STATE_HOME, "altimate-workspace-sync", name)).mode & 0o777).toBe(0o600) + // POSIX permission bits; Windows reports its own. + if (process.platform !== "win32") { + for (const name of files) { + expect(statSync(path.join(STATE_HOME, "altimate-workspace-sync", name)).mode & 0o777).toBe(0o600) + } } }) @@ -211,6 +214,37 @@ describe("record", () => { expect(published).toEqual([]) }) + test("a check that read under another account is dropped, even for the same workspace id", async () => { + await SyncState.record(project, 7, "skills", { items: { a: skill("Alpha") } }) + // Started under account B, written after the switch back to A. + const other = { ...ACCOUNT, tenant: "other", account: "digest-b" } + await SyncState.record(project, 7, "skills", { items: { b: skill("B's skill") } }, { scope: other }) + + expect((await SyncState.read(project, 7))?.entities.skills).toMatchObject({ count: 1, changes: null }) + expect(published).toEqual([]) + }) + + test("a check with no account when it started is not recorded", async () => { + await SyncState.record(project, 7, "skills", { items: { a: skill("Alpha") } }, { scope: null }) + expect(await SyncState.read(project, 7)).toBeNull() + }) + + test("an older observation never rolls a newer record back", async () => { + await SyncState.record( + project, + 7, + "skills", + { items: { a: skill("Alpha"), b: skill("Beta") } }, + { observedAt: 2_000 }, + ) + // A slow check that started earlier lands last. + await SyncState.record(project, 7, "skills", { items: { a: skill("Alpha") } }, { observedAt: 1_000 }) + + const entity = (await SyncState.read(project, 7))?.entities.skills + expect(entity).toMatchObject({ count: 2, lastCheckedAt: 2_000 }) + expect(published).toEqual([]) + }) + test("without an account nothing is recorded or served", async () => { scope.mockResolvedValue(null) await SyncState.record(project, 7, "skills", { items: { a: skill("Alpha") } }) @@ -247,4 +281,20 @@ describe("forget", () => { expect(state?.entities.skills?.count).toBe(1) expect(published).toEqual([]) }) + + test("keeps a record observed after the caller decided to retire the old one", async () => { + await SyncState.record(project, 7, "memory", { items: { m1: skill("One") } }, { observedAt: 5_000 }) + await SyncState.forget(project, "memory", { before: 4_000 }) + expect((await SyncState.read(project, 7))?.entities.memory?.count).toBe(1) + + // A tie is removed: the caller decided after that observation. + await SyncState.forget(project, "memory", { before: 5_000 }) + expect(await SyncState.read(project, 7)).toBeNull() + }) + + test("keeps the record when the binding being retired is no longer current", async () => { + await SyncState.record(project, 7, "memory", { items: { m1: skill("One") } }) + await SyncState.forget(project, "memory", { isCurrent: () => false }) + expect((await SyncState.read(project, 7))?.entities.memory?.count).toBe(1) + }) }) diff --git a/packages/opencode/test/server/altimate-workspace-routes.test.ts b/packages/opencode/test/server/altimate-workspace-routes.test.ts index b667dc4e9e..3a3def5401 100644 --- a/packages/opencode/test/server/altimate-workspace-routes.test.ts +++ b/packages/opencode/test/server/altimate-workspace-routes.test.ts @@ -266,17 +266,52 @@ describe("GET /altimate/workspace/status", () => { test("answers an unbound project with no workspace and no state", async () => { spies.push(spyOn(State, "resolveBindingOutcome").mockResolvedValue({ status: "unbound" } as never)) - expect(await (await get()).json()).toEqual({ ok: true, binding: "unbound", datamateId: null, entities: {} }) + expect(await (await get()).json()).toEqual({ + ok: true, + binding: "unbound", + stale: false, + datamateId: null, + entities: {}, + }) }) test("says when the binding could not be resolved, rather than reporting it unbound", async () => { spies.push(spyOn(State, "resolveBindingOutcome").mockResolvedValue({ status: "unknown" } as never)) - expect(await (await get()).json()).toEqual({ ok: true, binding: "unknown", datamateId: null, entities: {} }) + expect(await (await get()).json()).toEqual({ + ok: true, + binding: "unknown", + stale: false, + datamateId: null, + entities: {}, + }) + }) + + test("marks a binding the server could not confirm just now as last known", async () => { + spies.push( + spyOn(State, "resolveBindingOutcome").mockResolvedValue({ + status: "bound", + stale: true, + binding: { datamateId: 987_653, datamateName: "Analytics", repoRemote: null, projectPath: null, linkedAt: 1 }, + } as never), + ) + expect(await (await get()).json()).toEqual({ + ok: true, + binding: "bound", + stale: true, + datamateId: 987_653, + entities: {}, + }) }) test("a bound project with nothing recorded yet has no entities", async () => { spies.push(bound(987_654)) - expect(await (await get()).json()).toEqual({ ok: true, binding: "bound", datamateId: 987_654, entities: {} }) + expect(await (await get()).json()).toEqual({ + ok: true, + binding: "bound", + stale: false, + datamateId: 987_654, + entities: {}, + }) }) test("is refused under the kill switch", async () => {