-
Notifications
You must be signed in to change notification settings - Fork 134
feat(workspace): load workspace memory on demand, and show it in the memory tool #1123
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Changes from all commits
637b97c
ef335a9
9e6f4b1
8e9e506
2c5dd93
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -107,7 +107,7 @@ export function isEnabled(): boolean { | |
| export const syncInternals: { | ||
| resolveBinding?: () => Promise<CachedBinding | null> | ||
| /** Test seam for the local-existence check. Production reads the store. */ | ||
| blockExists?: (block: MemoryBlock) => Promise<boolean> | ||
| blockExists?: (block: MemoryBlock, directory?: string) => Promise<boolean> | ||
| } = {} | ||
|
|
||
| /** Instance.directory throws synchronously with no instance context, so a | ||
|
|
@@ -124,9 +124,13 @@ export function projectKeyFor(binding: CachedBinding): string { | |
| return binding.repoRemote ?? binding.projectPath ?? "unknown" | ||
| } | ||
|
|
||
| async function currentBinding(): Promise<CachedBinding | null> { | ||
| /** ``directory`` is the tree that owned the operation. Fire-and-forget mirrors | ||
| * 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. */ | ||
| async function currentBinding(directory?: string): Promise<CachedBinding | null> { | ||
| if (syncInternals.resolveBinding) return syncInternals.resolveBinding() | ||
| const directory = currentDirectory() | ||
| directory = directory ?? currentDirectory() ?? undefined | ||
| if (!directory) return null | ||
| try { | ||
| return await readLocalBinding(directory) | ||
|
|
@@ -147,7 +151,8 @@ async function currentBinding(): Promise<CachedBinding | null> { | |
| * anyway. */ | ||
| const MEMORY_ENABLED_TTL_MS = 60_000 | ||
|
|
||
| const memoryEnabledCache = new Map<number, { checkedAt: number }>() | ||
| /** Exported for tests: the positive TTL is why a failing read can look fine. */ | ||
| export const memoryEnabledCache = new Map<number, { checkedAt: number }>() | ||
|
|
||
| /** Warn once per workspace, not once per write. */ | ||
| const missingFieldWarned = new Set<number>() | ||
|
|
@@ -157,9 +162,19 @@ const missingFieldWarned = new Set<number>() | |
| * The workspace app exposes this as a user-facing toggle, so mirroring into a | ||
| * workspace with memory disabled would contradict what the user is shown. | ||
| * Fails closed: if the check cannot be made, nothing is mirrored. */ | ||
| /** Fail-closed answer for the WRITE path: an unreachable service must not let a | ||
| * mirror through. Reads need to tell "off" from "unreachable" — see | ||
| * {@link memoryStatus}. */ | ||
| async function memoryEnabled(binding: CachedBinding): Promise<boolean> { | ||
| return (await memoryStatus(binding)) === "enabled" | ||
| } | ||
|
|
||
| /** Three-way, because a read that cannot reach the service must not be reported | ||
| * as "this workspace has no memory" — that reads as success while destroying | ||
| * whatever the session already had. */ | ||
| async function memoryStatus(binding: CachedBinding): Promise<"enabled" | "disabled" | "error"> { | ||
| const cached = memoryEnabledCache.get(binding.datamateId) | ||
| if (cached && Date.now() - cached.checkedAt < MEMORY_ENABLED_TTL_MS) return true | ||
| if (cached && Date.now() - cached.checkedAt < MEMORY_ENABLED_TTL_MS) return "enabled" | ||
| try { | ||
| const workspaces = await WorkspaceApi.listDatamates() | ||
| const match = workspaces.find((w) => w.id === binding.datamateId) | ||
|
|
@@ -174,13 +189,12 @@ async function memoryEnabled(binding: CachedBinding): Promise<boolean> { | |
| const value = match?.memoryEnabled === true | ||
| if (value) memoryEnabledCache.set(binding.datamateId, { checkedAt: Date.now() }) | ||
| else memoryEnabledCache.delete(binding.datamateId) | ||
| return value | ||
| return value ? "enabled" : "disabled" | ||
| } catch (err) { | ||
| log.warn("could not confirm workspace memory setting, skipping mirror", { err: String(err) }) | ||
| log.warn("could not confirm workspace memory setting", { err: String(err) }) | ||
| // Not cached either way: a transient failure should neither disable the | ||
| // mirror for a minute nor keep it enabled. Failing closed already prevents | ||
| // this particular write. | ||
| return false | ||
| // mirror for a minute nor keep it enabled. | ||
| return "error" | ||
| } | ||
| } | ||
|
|
||
|
|
@@ -304,11 +318,14 @@ type PushOutcome = "stored" | "unchanged" | "declined" | "skipped" | |
| * a static import would close an eval-order cycle (see ./memory-backfill.ts). | ||
| * A read failure answers "yes" — refusing to mirror on a transient read error | ||
| * would silently drop a memory the user does have. */ | ||
| async function existsLocally(block: MemoryBlock): Promise<boolean> { | ||
| if (syncInternals.blockExists) return syncInternals.blockExists(block) | ||
| async function existsLocally(block: MemoryBlock, directory?: string): Promise<boolean> { | ||
| if (syncInternals.blockExists) return syncInternals.blockExists(block, directory) | ||
| try { | ||
| const { MemoryStore } = await import("@/memory/store") | ||
| return !!(await MemoryStore.read(block.scope, block.id)) | ||
| // ``directory`` is the tree that owned the write. Without it this resolves | ||
| // project scope from the ambient instance, which for a fire-and-forget | ||
| // mirror can be a different project entirely. | ||
| return !!(await MemoryStore.read(block.scope, block.id, directory)) | ||
| } catch (err) { | ||
| log.warn("could not confirm a block still exists locally; mirroring anyway", { | ||
| id: block.id, | ||
|
|
@@ -322,6 +339,7 @@ async function push( | |
| block: MemoryBlock, | ||
| binding: CachedBinding | null, | ||
| known?: KnownRecords, | ||
| directory?: string, | ||
| ): Promise<PushOutcome> { | ||
| const key = indexKey({ | ||
| scope: block.scope, | ||
|
|
@@ -358,7 +376,7 @@ async function push( | |
| // dequeues it, so a delete issued mid-sweep runs first and this push would | ||
| // otherwise undo it -- recreating a record the user deleted, or reviving an | ||
| // archived one, and then marking it synced so no later sweep re-archives it. | ||
| if (!(await existsLocally(block))) { | ||
| if (!(await existsLocally(block, directory))) { | ||
| log.warn("skipping mirror for a block that no longer exists locally", { | ||
| id: block.id, | ||
| scope: block.scope, | ||
|
|
@@ -457,7 +475,7 @@ function serialize<T>(scope: "global" | "project", blockId: string, op: () => Pr | |
| /** Mirror one block. Safe to call unconditionally — returns immediately when | ||
| * the pilot flag is off, the project is unbound, or the workspace has memory | ||
| * disabled. */ | ||
| export async function mirrorBlock(block: MemoryBlock): Promise<void> { | ||
| export async function mirrorBlock(block: MemoryBlock, directory?: string): Promise<void> { | ||
| if (!isEnabled()) return | ||
| // Queued BEFORE the binding lookup, not after. Both are async, so resolving | ||
| // them first let two operations on one block reach `serialize` in the | ||
|
|
@@ -469,23 +487,29 @@ export async function mirrorBlock(block: MemoryBlock): Promise<void> { | |
| // with nothing to consult and nothing to attribute it to. Global blocks still | ||
| // carry no workspace themselves, so they apply everywhere on read; the | ||
| // binding governs only whether we upload at all. | ||
| const binding = await currentBinding() | ||
| const binding = await currentBinding(directory) | ||
| if (!binding) return | ||
| if (!(await memoryEnabled(binding))) return | ||
| await push(block, binding) | ||
| await push(block, binding, undefined, directory) | ||
| }) | ||
| } | ||
|
|
||
| /** Archive a block's cloud record rather than deleting it, so the workspace | ||
| * keeps the history. Only this client filters the marker — other readers do | ||
| * not — so an archived record stays visible elsewhere. */ | ||
| export async function archiveBlock(scope: "global" | "project", blockId: string): Promise<void> { | ||
| export async function archiveBlock( | ||
| scope: "global" | "project", | ||
| blockId: string, | ||
| directory?: string, | ||
| ): Promise<void> { | ||
| if (!isEnabled()) return | ||
| // Queued behind any in-flight mirror for the same block, so a delete cannot | ||
| // run before the create it is meant to undo. The binding lookup happens | ||
| // inside the queued op for the same reason as in `mirrorBlock`. | ||
| return serialize(scope, blockId, async () => { | ||
| const binding = await currentBinding() | ||
| // Same capture as the mirror: the delete's own project decides which | ||
| // workspace record is archived, not whichever instance is current now. | ||
| const binding = await currentBinding(directory) | ||
| if (!binding) return | ||
| if (!(await memoryEnabled(binding))) return | ||
| await archiveNow(scope, blockId, binding) | ||
|
|
@@ -592,6 +616,7 @@ async function runQueue<T>( | |
| export async function backfill( | ||
| blocks: MemoryBlock[], | ||
| explicitBinding?: CachedBinding, | ||
| sweepDirectory?: string, | ||
| ): Promise<{ ok: number; failed: number; skipped: number; declined: number; gated: boolean }> { | ||
| // ``gated`` says the sweep never ran, as opposed to running and storing | ||
| // nothing. A caller recording "this binding is seeded" must be able to tell | ||
|
|
@@ -642,7 +667,10 @@ export async function backfill( | |
| log.info("workspace memory backfill starting", { pending: pending.length, skipped }) | ||
| const result = await runQueue( | ||
| pending, | ||
| (item) => serialize(item.block.scope, item.block.id, () => push(item.block, item.binding, known)), | ||
| (item) => | ||
| serialize(item.block.scope, item.block.id, () => | ||
| push(item.block, item.binding, known, sweepDirectory), | ||
| ), | ||
| BACKFILL_CONCURRENCY, | ||
| ) | ||
| // `skipped` combines blocks filtered before the queue (already synced, or | ||
|
|
@@ -729,7 +757,9 @@ export async function hydrate(sessionID: string): Promise<void> { | |
| if (!isEnabled()) return | ||
| const state = sessionState(sessionID) | ||
| if (state.hydration) return state.hydration | ||
| state.hydration = doHydrate(sessionID) | ||
| // The overlay is deliberately NOT cleared before loading: clearing first made | ||
| // workspace memory blink out of the prompt whenever a fetch ran long. | ||
| state.hydration = loadWorkspaceMemory().then((outcome) => commitLoad(sessionID, state, outcome)) | ||
| return state.hydration | ||
| } | ||
|
|
||
|
|
@@ -765,16 +795,29 @@ export async function whenHydrated( | |
| } | ||
| } | ||
|
|
||
| async function doHydrate(sessionID: string): Promise<void> { | ||
| /** What a load attempt actually concluded. | ||
| * | ||
| * "nothing to load" and "could not load" must stay distinguishable: collapsing | ||
| * 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: "disabled" } | ||
| | { status: "error" } | ||
|
|
||
| /** Read this project's workspace memory. Pure: it publishes nothing, so a slow | ||
| * load that has been superseded cannot write over a newer result. */ | ||
| async function loadWorkspaceMemory(): Promise<LoadOutcome> { | ||
| try { | ||
| const binding = await currentBinding() | ||
| if (!binding || !(await memoryEnabled(binding))) { | ||
| sessionState(sessionID).overlay = [] | ||
| return | ||
| } | ||
| const ownProjectKey = binding ? projectKeyFor(binding) : undefined | ||
| const ownWorkspace = binding ? String(binding.datamateId) : undefined | ||
| if (!binding) return { status: "unlinked" } | ||
| const enabled = await memoryStatus(binding) | ||
| if (enabled === "error") return { status: "error" } | ||
| if (enabled === "disabled") return { status: "disabled" } | ||
|
|
||
| const ownProjectKey = projectKeyFor(binding) | ||
| const ownWorkspace = String(binding.datamateId) | ||
| const records = await MemoryApi.list() | ||
|
|
||
| const blocks: RemoteMemoryBlock[] = [] | ||
|
|
@@ -789,14 +832,23 @@ async function doHydrate(sessionID: string): Promise<void> { | |
| if (block.expires && new Date(block.expires) <= new Date()) continue | ||
| blocks.push(block) | ||
| } | ||
|
|
||
| sessionState(sessionID).overlay = blocks | ||
| if (blocks.length > 0) { | ||
| log.info("workspace memory hydrated", { blocks: blocks.length, workspace: binding?.datamateName }) | ||
| } | ||
| return { status: "loaded", blocks } | ||
| } catch (err) { | ||
| log.warn("workspace memory hydration failed", { err: String(err) }) | ||
| sessionState(sessionID).overlay = [] | ||
| log.warn("workspace memory load failed", { err: String(err) }) | ||
| return { status: "error" } | ||
| } | ||
| } | ||
|
|
||
| /** Publish a load result into the state that launched it. | ||
| * | ||
| * The generation check is the point: ``refresh`` replaces a session's state, and | ||
| * an older in-flight load must not write into the newer one. */ | ||
| function commitLoad(sessionID: string, state: SessionMemory, outcome: LoadOutcome): void { | ||
| if (sessions.get(sessionID) !== state) return | ||
| if (outcome.status === "error") return | ||
| state.overlay = outcome.status === "loaded" ? outcome.blocks : [] | ||
| if (outcome.status === "loaded" && outcome.blocks.length > 0) { | ||
| log.info("workspace memory hydrated", { blocks: outcome.blocks.length }) | ||
| } | ||
| } | ||
|
|
||
|
|
@@ -806,6 +858,48 @@ export function overlayBlocks(sessionID: string): RemoteMemoryBlock[] { | |
| return [...(sessions.get(sessionID)?.overlay ?? [])] | ||
| } | ||
|
|
||
| export type RefreshResult = { | ||
| count: number | ||
| ok: boolean | ||
| status: "loaded" | "unlinked" | "disabled" | "error" | "off" | ||
| } | ||
|
|
||
| /** Re-read this session's workspace memory, discarding what it already holds. | ||
| * | ||
| * ``hydrate`` is idempotent for the life of a session, which keeps the per-turn | ||
| * call cheap -- but it also means a session started before a teammate (or this | ||
| * user on another machine) wrote a block never sees it. This is the on-demand | ||
| * path. | ||
| * | ||
| * Serialized per session: two refreshes racing would otherwise let the second | ||
| * capture the first's not-yet-filled state as "previous" and, on failure, | ||
| * restore emptiness over real memory. */ | ||
| export async function refresh(sessionID: string): Promise<RefreshResult> { | ||
| if (!isEnabled()) return { count: 0, ok: false, status: "off" } | ||
| return serialize("global", `refresh:${sessionID}`, async () => { | ||
| const previous = overlayBlocks(sessionID) | ||
| const outcome = await loadWorkspaceMemory() | ||
| if (outcome.status === "error") { | ||
| // 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) | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. thought: error path restores
In normal usage this doesn't fire: |
||
| state.overlay = previous | ||
|
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. WARNING:
It is also not merely dead code: Drop the Reply with |
||
| return { count: previous.length, ok: false, status: "error" } | ||
| } | ||
| // Replace the session's state so any older in-flight hydration is orphaned | ||
| // by commitLoad's generation check rather than overwriting this result. | ||
| sessions.delete(sessionID) | ||
| const state = sessionState(sessionID) | ||
| state.hydration = Promise.resolve() | ||
| commitLoad(sessionID, state, outcome) | ||
| return { | ||
| count: state.overlay.length, | ||
| ok: outcome.status === "loaded", | ||
| status: outcome.status, | ||
| } | ||
| }) | ||
| } | ||
|
|
||
| /** Forget a session's hydration, or all of them. | ||
| * | ||
| * Not called per turn: doing so defeated ``hydrate``'s idempotence and made | ||
|
|
||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
issue:
loadWorkspaceMemorycallsMemoryApi.list()directly — same truncation gap flagged in PR #1116.PR #1116 introduced
fetchKnownRecords()as a wrapper aroundMemoryApi.list()that detects when the response is exactlyLIST_LIMIT = 200and emits alog.warn. Every other call site in this file (archiveNow,push,backfill) goes through it;doHydratewas the sole exception, and a review comment on #1116 flagged it.This PR replaces
doHydratewithloadWorkspaceMemorybut doesn't pick up that fix. For a user whose workspace has ≥200 records, some blocks silently won't appear in session injection after ahydrateorrefresh— no log line, no signal.One-line fix, following the pattern everywhere else: