feat(workspace): record workspace sync state and publish changes to clients

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,12 @@ export async function status(
     ...(memory === "unreadable" ? { memoryUnreadable: true as const } : {}),
     skillsEnabled: SkillSync.isEnabled(),
     skillsSyncedAt: await skillsSyncedAt(directory, binding),
+    // 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,
   }
 }
 
@@ -374,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<UnlinkReport> {
+  // 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
@@ -491,6 +506,12 @@ export async function unlink(directory: string): Promise<UnlinkReport> {
   } 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.

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<CachedBinding | null> {
   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<string, { label: string; version: string }>
+    }
+  | { status: "unlinked"; /** The server confirmed it, rather than a lookup failing. */ confirmed: boolean }
   | { status: "disabled" }
   | { status: "error" }
-) & { epoch?: string; dir?: string | null }
+) & {
+  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. */
@@ -1386,26 +1431,36 @@ async function loadWorkspaceMemory(directory?: string): Promise<LoadOutcome> {
   // 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
+  // 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 }
-    if (enabled === "disabled") return { status: "disabled", epoch, dir }
+    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)
@@ -1430,6 +1485,7 @@ async function loadWorkspaceMemory(directory?: string): Promise<LoadOutcome> {
     )
 
     const blocks: RemoteMemoryBlock[] = []
+    const items: Record<string, { label: string; version: string }> = {}
     for (const record of records) {
       if (!record?.id) continue
       if (!isMirrorRecord(record) || isArchived(record)) continue
@@ -1440,13 +1496,14 @@ async function loadWorkspaceMemory(directory?: string): Promise<LoadOutcome> {
       // 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 }
+    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 }
+    return { status: "error", epoch: vouched, dir, datamateId: resolved, ...stamp }
   }
 }
 
@@ -1461,6 +1518,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 +1529,51 @@ function commitLoad(sessionID: string, state: SessionMemory, outcome: LoadOutcom
   }
 }
 
+// altimate_change start — workspace sync state
+/** 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, 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 (retire) return SyncState.forget(dir, "memory", { isCurrent, before: outcome.observedAt })
+      if (outcome.datamateId === undefined) return
+      const result =
+        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) }))
+}
+
+/** 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)
+    .update("\0")
+    .update([...block.tags].sort().join("\0"))
+    .digest("hex")
+    .slice(0, 16)
+  return { label: blockTitle(block), version: `${block.updated ?? ""}:${digest}` }
+}
+// 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[] {
@@ -1510,6 +1615,9 @@ export async function refresh(sessionID: string, directory?: string): Promise<Re
     // read nor what the session had before belongs to the current binding.
     if (outcome.epoch !== epochFor(outcome.dir ?? null)) return { count: 0, ok: false, status: "error" }
     if (outcome.status === "error") {
+      // altimate_change start — workspace sync state
+      recordSyncState(outcome)
+      // altimate_change end
       // Keep what the session had. Emptying it because the network hiccuped is
       // strictly worse than not reloading, and the user asked for a reload.
       const state = sessionState(sessionID)