diff --git a/packages/opencode/src/altimate/workspace/identity.ts b/packages/opencode/src/altimate/workspace/identity.ts index 4bc016ad6c..5075bd2f2e 100644 --- a/packages/opencode/src/altimate/workspace/identity.ts +++ b/packages/opencode/src/altimate/workspace/identity.ts @@ -29,15 +29,16 @@ import { workspaceLabel } from "./workspace-name" import { isEnabled } from "./engine-seams" import { Instance } from "../../project/instance" import { AltimateApi } from "../api/client" +import { memoryEnabledCached } from "./memory-sync" /** Independent of `awareness.ts`'s MAX_SECTION_CHARS (2,000) — this section is a short, * fixed-shape identity statement, not an open-ended list of served integrations, so a * much smaller ceiling is enough. The label is budgeted separately (`MAX_LABEL_CHARS` * in `workspace-name.ts`) so the cap here is defense in depth and never cuts the - * instruction itself: the longest fixed shape (pinned and stale) is ~1,080 characters - * before the label, and a label at its budget still leaves room. A test renders every - * shape with a budget-sized label and checks the name survives. */ -export const MAX_SECTION_CHARS = 1_500 + * instruction itself: the longest fixed shape (pinned, stale, with the team-memory line) + * is ~1,550 characters before the label, and a label at its budget still leaves room. A + * test renders every shape with a budget-sized label and checks the name survives. */ +export const MAX_SECTION_CHARS = 2_000 const HEADING = "## Altimate Workspace" @@ -65,20 +66,37 @@ const LINK_HINT = * leaving it to the caller) so the cap is part of the pure, testable surface — the * guard is against a pathological workspace name, and every branch below is built from * one, so it belongs where the name is rendered. */ -export function render(outcome: BindingOutcome, cap = MAX_SECTION_CHARS): string { - const body = renderBody(outcome) +export function render(outcome: BindingOutcome, cap = MAX_SECTION_CHARS, opts: RenderOptions = {}): string { + const body = renderBody(outcome, opts) if (body.length <= cap) return body // Fail closed rather than truncate: a cut instruction is worse than a missing // name. The name is the only variable field, so drop it and keep the id; if // even that does not fit, say nothing rather than something partial. if (outcome.status === "bound") { - const unnamed = renderBody({ ...outcome, binding: { ...outcome.binding, datamateName: "" } }) + const unnamed = renderBody({ ...outcome, binding: { ...outcome.binding, datamateName: "" } }, opts) if (unnamed.length <= cap) return unnamed } return "" } -function renderBody(outcome: BindingOutcome): string { +export type RenderOptions = { + /** The bound workspace has memory on (or has not said otherwise): tell the model + * which store is the team's. `systemSection` derives it from the enablement memo; + * the pure formatter takes it as an argument so tests stay deterministic. */ + teamMemory?: boolean +} + +/** Two stores answer "remember this". Only one is read by other linked checkouts, + * and nothing told the model which — so on a plain "save this for the team" it + * reached for the engine's hub and the decision never left the session (#1332). */ +const TEAM_MEMORY_LINE = + "Team memory: save decisions and conventions with `altimate_memory_write` (scope " + + '"project" for this project, "global" for everything); they sync to the workspace and to every ' + + "linked checkout. The `datamate_*` memory tools (`datamate_add_memories`, `datamate_search_memory`) " + + "are the engine's separate store and are not what teammates' sessions read. Before saving, check " + + "`altimate_memory_read` for an existing block on the same subject and update it rather than add a duplicate." + +function renderBody(outcome: BindingOutcome, opts: RenderOptions = {}): string { if (outcome.status === "bound") { const id = String(outcome.binding.datamateId) const name = workspaceLabel(outcome.binding.datamateName, undefined) @@ -105,6 +123,7 @@ function renderBody(outcome: BindingOutcome): string { ? " Skills and memory follow this workspace; warehouse tool routing still follows the " + "project's own link, which may name a different workspace." : ""), + ...(opts.teamMemory ? [TEAM_MEMORY_LINE] : []), `When ${TRIGGER}, the answer is this Altimate Workspace — never substitute ` + "another service's own \"workspace\" (a Databricks workspace, an IDE's " + "workspace folder, etc.) for it, and the reverse: a question about another " + @@ -287,6 +306,20 @@ async function accountScope(): Promise { return { tenant: c.altimateInstanceName, apiUrl: c.altimateUrl, account } } +/** The team-memory line is shown only for a bound workspace known to have memory + * enabled. "Unknown" is not enough: the write path uploads only on a confirmed + * "enabled", so advertising sync on an unconfirmed or failed check would tell the + * model teammates will read a block that stays on this machine. The memo is + * populated by the first enablement check of the session (the backfill sweep on + * bind, or the first mirror), so the line appears from the next turn on. */ +function renderOptions(outcome: BindingOutcome): RenderOptions { + if (outcome.status !== "bound") return {} + // A stale outcome is "last known … may since have changed": promising that a + // save syncs to that workspace would contradict the line above it. + if (outcome.stale) return {} + return { teamMemory: memoryEnabledCached(outcome.binding) === "enabled" } +} + function keyFor(scope: AccountScope, directory: string): string { return `${scope.tenant}|${scope.apiUrl}|${scope.account}|${directory}` } @@ -357,14 +390,15 @@ export async function systemSection(): Promise { if (!scope) return render({ status: "unknown" }) const key = keyFor(scope, directory) const hit = memo.get(key) - if (fresh(hit)) return render(hit!.outcome) + if (fresh(hit)) return render(hit!.outcome, MAX_SECTION_CHARS, renderOptions(hit!.outcome)) // The fallback is itself raced against a small budget, so the wait is // bounded by RESOLVE_DEADLINE_MS + FALLBACK_BUDGET_MS, not by the disk. const deadline = after(RESOLVE_DEADLINE_MS, () => Promise.race([lastKnown(key, directory), after(FALLBACK_BUDGET_MS, () => ({ status: "unknown" }))]), ) try { - return render(await Promise.race([resolve(key, directory), deadline])) + const outcome = await Promise.race([resolve(key, directory), deadline]) + return render(outcome, MAX_SECTION_CHARS, renderOptions(outcome)) } finally { for (const t of timers) clearTimeout(t) } diff --git a/packages/opencode/src/altimate/workspace/memory-sync.ts b/packages/opencode/src/altimate/workspace/memory-sync.ts index ca4240bbc3..8e7b9c0b3c 100644 --- a/packages/opencode/src/altimate/workspace/memory-sync.ts +++ b/packages/opencode/src/altimate/workspace/memory-sync.ts @@ -185,6 +185,11 @@ export function memoryEnabledCached(binding: CachedBinding): "enabled" | "disabl /** Test seam: both memos are process-global, and an earlier case's answer * would otherwise leak into a later one. */ +/** Test seam: record the workspace's "memory off" answer without a request. */ +export function noteMemoryDisabledForTests(datamateId: number): void { + memoryDisabledMemo.set(datamateId, Date.now()) +} + export function resetEnablementMemoForTests(): void { memoryEnabledCache.clear() memoryDisabledMemo.clear() @@ -515,7 +520,16 @@ async function push( * memory the user deleted. Two rapid saves race the same way and create * duplicates. Keyed by scope+id, so unrelated blocks still mirror in parallel. */ -const blockQueues = new Map>() +// Anchored on a process-global, as skill-sync's tables are: this module is reached +// through two module graphs in one process — `MemoryStore` via `@/…`, the `run` +// exit path via a relative specifier — and a runtime that keeps a record per +// specifier would fork plain module state. A flush that saw an empty set while +// the writer's copy held the upload would defeat the #1332 fix silently. +const SYNC_STATE = Symbol.for("altimate.memory-sync.state") +const syncState: { blockQueues: Map>; mirrorsInFlight: Set> } = (( + globalThis as unknown as Record +)[SYNC_STATE] ??= { blockQueues: new Map(), mirrorsInFlight: new Set() }) +const blockQueues = syncState.blockQueues function serialize(scope: "global" | "project", blockId: string, op: () => Promise): Promise { const key = `${scope}:${blockId}` @@ -530,6 +544,44 @@ function serialize(scope: "global" | "project", blockId: string, op: () => Pr return next } +/** Mirrors still in flight. `MemoryStore.write` fires the mirror and forgets + * it (the local file is already durable, and a cloud failure must not fail + * the write), which is right for the TUI and wrong for a one-shot `run`: the + * process exits the moment the turn ends, routinely before the upload lands, + * and the block a teammate was meant to see never leaves the machine (#1332). + * Tracked here so `flushPendingMirrors` can hold the exit for them, the way + * `skill-sync.flushPendingSyncs` holds it for a cold skill sync. */ +const mirrorsInFlight = syncState.mirrorsInFlight + +/** Hold a task in `mirrorsInFlight` for its lifetime. */ +async function tracked(task: Promise): Promise { + mirrorsInFlight.add(task) + try { + await task + } finally { + mirrorsInFlight.delete(task) + } +} + +/** Await every mirror and archive still in flight, bounded, so a short-lived + * process does not exit with an upload or an archive half-done. Failures are + * already logged by the caller; this only waits. */ +export async function flushPendingMirrors(timeoutMs = 30_000): Promise { + const pending = [...mirrorsInFlight] + if (pending.length === 0) return + let timer: ReturnType | undefined + try { + await Promise.race([ + Promise.allSettled(pending), + new Promise((resolve) => { + timer = setTimeout(resolve, timeoutMs) + }), + ]) + } finally { + if (timer) clearTimeout(timer) + } +} + /** 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. */ @@ -538,7 +590,7 @@ export async function mirrorBlock(block: MemoryBlock, directory?: string): Promi // Queued BEFORE the binding lookup, not after. Both are async, so resolving // them first let two operations on one block reach `serialize` in the // opposite order to the writes that triggered them. - await serialize(block.scope, block.id, async () => { + const task = serialize(block.scope, block.id, async () => { // A binding is required for EVERY scope, not just project. Memories are // associated with a workspace, and the workspace is what carries the // memory_enabled setting — mirroring from an unbound directory would upload @@ -550,6 +602,7 @@ export async function mirrorBlock(block: MemoryBlock, directory?: string): Promi if (!(await memoryEnabled(binding))) return await push(block, binding, undefined, directory) }) + return tracked(task) } /** Archive a block's cloud record rather than deleting it, so the workspace @@ -563,15 +616,19 @@ export async function archiveBlock( 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 () => { - // 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) - }) + // inside the queued op for the same reason as in `mirrorBlock`. Tracked like a + // mirror: a one-shot `run` that deletes a block must not exit before the + // workspace record is archived, or teammates keep a memory the author removed. + return tracked( + serialize(scope, blockId, async () => { + // 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) + }), + ) } async function archiveNow( diff --git a/packages/opencode/src/cli/cmd/run.ts b/packages/opencode/src/cli/cmd/run.ts index 3ea9ddb91b..3651824c3a 100644 --- a/packages/opencode/src/cli/cmd/run.ts +++ b/packages/opencode/src/cli/cmd/run.ts @@ -1075,14 +1075,32 @@ You are speaking to a non-technical business executive. Follow these rules stric // altimate_change end // Register crash handlers to flush the trace on unexpected exit + // altimate_change start — and hold a signal exit, briefly, for a memory + // mirror still on the wire: Ctrl-C while the last response streams used to + // kill the upload of a block saved that turn (#1332). Bounded well below + // the normal-exit flush, and a second signal is not delayed by the first. + let signalled = false + const exitAfterMirrors = (code: number) => { + if (signalled) return process.exit(code) + signalled = true + if (!CoreFlag.ALTIMATE_WORKSPACE) return process.exit(code) + // Stop the run first so no new mirror is enqueued behind the snapshot the + // flush takes; what is already on the wire is what gets the 2s. + eventAbort.abort() + void import("../../altimate/workspace/memory-sync") + .then((m) => m.flushPendingMirrors(2_000)) + .catch(() => {}) + .finally(() => process.exit(code)) + } const onSigint = () => { tracer?.flushSync("Process interrupted") - process.exit(130) + exitAfterMirrors(130) } const onSigterm = () => { tracer?.flushSync("Process interrupted") - process.exit(143) + exitAfterMirrors(143) } + // altimate_change end // altimate_change start — honest rc on fatal abort. beforeExit firing // before the run finishes means the event loop drained before the run // completed — the prompt/event stream was abandoned (observed: a @@ -1445,9 +1463,17 @@ You are speaking to a non-technical business executive. Follow these rules stric // received its skills at all. Imported lazily and only when the feature is // on, so an opted-out run does not load the module. if (CoreFlag.ALTIMATE_WORKSPACE) { - await import("../../altimate/workspace/skill-sync") - .then((m) => m.flushPendingSyncs()) - .catch(() => {}) + // And the memory mirrors: a block saved on the last turn was uploaded + // fire-and-forget and lost the same race (#1332). Both flushes run together + // under their own bounds, so two stalled backends cost one wait, not two. + await Promise.all([ + import("../../altimate/workspace/skill-sync") + .then((m) => m.flushPendingSyncs()) + .catch(() => {}), + import("../../altimate/workspace/memory-sync") + .then((m) => m.flushPendingMirrors()) + .catch(() => {}), + ]) } // altimate_change end diff --git a/packages/opencode/src/memory/tools/memory-write.ts b/packages/opencode/src/memory/tools/memory-write.ts index 3969f1cf45..4613d9756d 100644 --- a/packages/opencode/src/memory/tools/memory-write.ts +++ b/packages/opencode/src/memory/tools/memory-write.ts @@ -6,7 +6,7 @@ import { MEMORY_MAX_BLOCK_SIZE, MEMORY_MAX_BLOCKS_PER_SCOPE, CitationSchema, Mem const idSchema = MemoryBlockSchema.shape.id export const MemoryWriteTool = Tool.define("altimate_memory_write", { - description: `Save an Altimate Memory block for cross-session persistence. Use this to store information worth remembering across sessions — warehouse configurations, naming conventions, team preferences, data model notes, or past analysis decisions. Each block is a Markdown file persisted to disk. Max ${MEMORY_MAX_BLOCK_SIZE} chars per block, ${MEMORY_MAX_BLOCKS_PER_SCOPE} blocks per scope. Supports hierarchical IDs with slashes (e.g., 'warehouse/snowflake-config'), optional TTL expiration, and citation-backed memories.`, + description: `Save an Altimate Memory block for cross-session persistence. Use this to store information worth remembering across sessions — warehouse configurations, naming conventions, team preferences, data model notes, or past analysis decisions. When the project is linked to an Altimate Workspace with workspace sync on and workspace memory enabled this is the TEAM's memory: blocks sync to the workspace and to every linked checkout, so a decision saved here is what teammates' sessions read. Before saving, call \`altimate_memory_read\` for an existing block on the same subject and update it rather than add a duplicate. The datamate_* memory tools (datamate_add_memories, datamate_search_memory) are the engine's separate store and are not what linked checkouts read. Each block is a Markdown file persisted to disk. Max ${MEMORY_MAX_BLOCK_SIZE} chars per block, ${MEMORY_MAX_BLOCKS_PER_SCOPE} blocks per scope. Supports hierarchical IDs with slashes (e.g., 'warehouse/snowflake-config'), optional TTL expiration, and citation-backed memories.`, parameters: z.object({ id: idSchema .describe( diff --git a/packages/opencode/test/altimate/workspace/identity-section.test.ts b/packages/opencode/test/altimate/workspace/identity-section.test.ts index 31e770be75..4e77e3d300 100644 --- a/packages/opencode/test/altimate/workspace/identity-section.test.ts +++ b/packages/opencode/test/altimate/workspace/identity-section.test.ts @@ -104,6 +104,70 @@ describe("systemSection", () => { expect(out).toContain("never substitute") }) + test("a bound project gets the team-memory line only once the workspace's memory is confirmed on", async () => { + const { memoryEnabledCache, resetEnablementMemoForTests, noteMemoryDisabledForTests } = await import( + "../../../src/altimate/workspace/memory-sync" + ) + await recordApprovedBinding(projectDir, { + datamateId: 77, + datamateName: "Team", + repoRemote: null, + projectPath: projectDir, + linkedAt: Date.now(), + }) + resetEnablementMemoForTests() + memoryEnabledCache.set(77, { checkedAt: Date.now() }) + try { + expect(await inProject(systemSection)).toContain("Team memory:") + resetOutcomeMemoForTests() + resetEnablementMemoForTests() + // A remembered "no" from the workspace switches the line off. + noteMemoryDisabledForTests(77) + expect(await inProject(systemSection)).not.toContain("Team memory:") + expect(await inProject(systemSection)).toContain('is "Team"') + // Not yet checked is not "on": the write path uploads only on a confirmed + // yes, so the prompt must not promise a sync that would stay local. (bot review) + resetOutcomeMemoForTests() + resetEnablementMemoForTests() + expect(await inProject(systemSection)).not.toContain("Team memory:") + expect(await inProject(systemSection)).toContain('is "Team"') + } finally { + resetEnablementMemoForTests() + } + }) + + test("the production enablement check fills the memo the line reads (codex on #1344)", async () => { + // Not a hand-written cache entry: the sidebar poller's check (`memoryStatus`, the + // same one the write path and the backfill sweep run) asks GET /datamates/ and + // caches a yes; the section reads that. Deleting the cache write would hide the + // line for good, and the previous test would not notice. + const { memoryEnabledForPoller, resetEnablementMemoForTests, resetPollMemoForTests } = await import( + "../../../src/altimate/workspace/memory-sync" + ) + globalThis.fetch = (async (input: any) => + new Response( + JSON.stringify( + String(input).includes("/datamates/") ? { datamates: [{ id: 91, name: "Team", memory_enabled: true }] } : {}, + ), + { status: 200, headers: { "content-type": "application/json" } }, + )) as unknown as typeof fetch + const binding = { datamateId: 91, datamateName: "Team", repoRemote: null, projectPath: projectDir, linkedAt: Date.now() } + // The bind's own backfill sweep runs the same check; wait for it so it cannot + // refill the memo between the reset and the first assertion. (bot review) + await recordApprovedBinding(projectDir, binding, { awaitBackfill: true }) + resetEnablementMemoForTests() + resetPollMemoForTests() + try { + expect(await inProject(systemSection)).not.toContain("Team memory:") + expect(await memoryEnabledForPoller(binding as never)).toBe("enabled") + resetOutcomeMemoForTests() + expect(await inProject(systemSection)).toContain("Team memory:") + } finally { + resetEnablementMemoForTests() + resetPollMemoForTests() + } + }) + test("renders nothing when the workspace pilot is off", async () => { // A user outside the pilot has no Altimate Workspace to be linked to, and // must not be told every turn that none is linked and how to link one. @@ -262,11 +326,20 @@ describe("systemSection", () => { globalThis.fetch = (async () => { throw new Error("offline") }) as unknown as typeof fetch - const out = await inProject(systemSection) - expect(out).toContain("was last known to be linked to Altimate Workspace id 9") - expect(out).toContain('is "Finance"') - expect(out).toContain("could not be re-verified just now") - expect(out).not.toContain("This project is linked to Altimate Workspace id 9") + // Memory was confirmed on earlier; that does not make a sync promise true of a + // binding the server can no longer vouch for. (codex on #1344) + const { memoryEnabledCache, resetEnablementMemoForTests } = await import("../../../src/altimate/workspace/memory-sync") + memoryEnabledCache.set(9, { checkedAt: Date.now() }) + try { + const out = await inProject(systemSection) + expect(out).toContain("was last known to be linked to Altimate Workspace id 9") + expect(out).toContain('is "Finance"') + expect(out).toContain("could not be re-verified just now") + expect(out).not.toContain("This project is linked to Altimate Workspace id 9") + expect(out).not.toContain("Team memory:") + } finally { + resetEnablementMemoForTests() + } }) test("concurrent steps share one resolve (single-flight)", async () => { diff --git a/packages/opencode/test/altimate/workspace/identity.test.ts b/packages/opencode/test/altimate/workspace/identity.test.ts index b981fc0829..4d73db0453 100644 --- a/packages/opencode/test/altimate/workspace/identity.test.ts +++ b/packages/opencode/test/altimate/workspace/identity.test.ts @@ -25,6 +25,20 @@ describe("bound — a specific Altimate Workspace is linked", () => { } const boundOut = render(boundOutcome) + test("with team memory on, the section names the team's store and the engine's hub as separate (#1332)", () => { + const out = render(boundOutcome, undefined, { teamMemory: true }) + expect(out).toContain("Team memory: save decisions and conventions with `altimate_memory_write`") + expect(out).toContain("sync to the workspace and to every linked checkout") + expect(out).toContain("`datamate_add_memories`") + expect(out).toContain("are the engine's separate store") + expect(out).toContain("check `altimate_memory_read` for an existing block") + expect(out.split("\n")).toHaveLength(5) + // Off (workspace memory disabled) and by default (pure formatter): no line. + expect(render(boundOutcome, undefined, { teamMemory: false })).not.toContain("Team memory") + expect(boundOut).not.toContain("Team memory") + expect(render({ status: "unbound" }, undefined, { teamMemory: true })).not.toContain("Team memory") + }) + test("names the workspace and forbids substituting another service's 'workspace' for an identity question", () => { expect(boundOut).toContain("## Altimate Workspace") expect(boundOut).toContain('"Foo Corp Data Team"') @@ -267,10 +281,12 @@ describe("the section cap fails closed", () => { { status: "bound", binding: b(true), stale: true }, ] for (const shape of shapes) { - const out = render(shape) - expect(out.length).toBeLessThanOrEqual(MAX_SECTION_CHARS) - expect(out).toContain('\\"\\"\\"') // the name is there, not "(unnamed)" - expect(out).not.toContain("(unnamed)") + for (const teamMemory of [false, true]) { + const out = render(shape, MAX_SECTION_CHARS, { teamMemory }) + expect(out.length).toBeLessThanOrEqual(MAX_SECTION_CHARS) + expect(out).toContain('\\"\\"\\"') // the name is there, not "(unnamed)" + expect(out).not.toContain("(unnamed)") + } } }) diff --git a/packages/opencode/test/altimate/workspace/memory-sync.test.ts b/packages/opencode/test/altimate/workspace/memory-sync.test.ts index 84ba0844a0..a81510ae17 100644 --- a/packages/opencode/test/altimate/workspace/memory-sync.test.ts +++ b/packages/opencode/test/altimate/workspace/memory-sync.test.ts @@ -44,6 +44,7 @@ const { isEnabled, memoryEnabledCached, mirrorBlock, + flushPendingMirrors, overlayBlocks, resetOverlay, syncInternals, @@ -338,8 +339,134 @@ describe("buildMetadata", () => { }) }) +/** A fetch that parks every request behind a gate and reports when the first one + * has arrived — the readiness signal the gated tests wait on instead of a sleep. */ +function gatedFetch() { + let release!: () => void + const gate = new Promise((r) => (release = r)) + let entered!: () => void + const firstRequest = new Promise((r) => (entered = r)) + const original = globalThis.fetch + globalThis.fetch = (async (input: any, init?: any) => { + entered() + await gate + return original(input, init) + }) as unknown as typeof fetch + return { + release, + firstRequest, + restore: () => { + globalThis.fetch = original + }, + } +} + // ── write path ────────────────────────────────────────────────────────────── describe("mirrorBlock", () => { + test("flushPendingMirrors waits for a mirror a short-lived process would abandon (#1332)", async () => { + // `MemoryStore.write` fires the mirror and forgets it; a one-shot `run` exits + // when the turn ends, routinely before the upload lands. The flush holds the + // exit for it, the way `skill-sync.flushPendingSyncs` holds it for a skill sync. + const net = gatedFetch() + createResult = [{ id: "mem-slow" }] + let settled = false + const mirror = mirrorBlock(block({ id: "slow" })).then(() => (settled = true)) + try { + await net.firstRequest // it is genuinely on the wire + expect(settled).toBe(false) + const flush = flushPendingMirrors() + let flushed = false + void flush.then(() => (flushed = true)) + await Bun.sleep(20) + expect(flushed).toBe(false) // the flush is holding for the mirror + net.release() + await flush + expect(settled).toBe(true) + // Nothing in flight: an immediate return. + const started = Date.now() + await flushPendingMirrors() + expect(Date.now() - started).toBeLessThan(50) + } finally { + net.release() + await mirror + net.restore() + } + }) + + test("the in-flight set lives on globalThis, so every copy of this module shares it (codex on #1344)", async () => { + // `MemoryStore` reaches this module via `@/…`, the `run` exit path via a relative + // path. Bun hands both the same record today, so a two-specifier test proves + // nothing; what is asserted is the anchor itself: a tracked mirror is visible on + // the process-global state, which is what a second module record would read. + const net = gatedFetch() + createResult = [{ id: "mem-anchor" }] + const mirror = mirrorBlock(block({ id: "anchored" })) + try { + await net.firstRequest + const state = (globalThis as any)[Symbol.for("altimate.memory-sync.state")] + expect(state?.mirrorsInFlight?.size).toBe(1) + net.release() + await mirror + expect(state.mirrorsInFlight.size).toBe(0) + } finally { + net.release() + await mirror + net.restore() + } + }) + + test("flushPendingMirrors gives up after its bound rather than hanging exit forever", async () => { + // Gated, not hung forever: `mirrorsInFlight` is process-global, and a mirror that + // never settles would make every later default-bound flush in this process wait + // the full 30s. (bot review) + const net = gatedFetch() + createResult = [{ id: "mem-hung" }] + const mirror = mirrorBlock(block({ id: "hung" })) + try { + await net.firstRequest + const started = Date.now() + await flushPendingMirrors(100) + const waited = Date.now() - started + expect(waited).toBeGreaterThanOrEqual(90) + expect(waited).toBeLessThan(1000) + } finally { + net.release() + await mirror + net.restore() + } + }) + + test("flushPendingMirrors holds the exit for an archive too, not only for a mirror", async () => { + // A one-shot `run` that deletes a block fires `archiveBlock` and forgets it; + // without tracking, the process could exit with the workspace record still + // live and teammates keeping a memory the author removed. (bot review) + const b = block({ id: "to-archive-late" }) + createResult = [{ id: "mem-archive-late" }] + await mirrorBlock(b) + listResponse = [ + { id: "mem-archive-late", memory: b.content, metadata: { source: MIRROR_SOURCE, block_id: "to-archive-late", block_scope: "global" } }, + ] + const net = gatedFetch() + let settled = false + const archive = archiveBlock("global", "to-archive-late").then(() => (settled = true)) + try { + await net.firstRequest + expect(settled).toBe(false) + const flush = flushPendingMirrors() + let flushed = false + void flush.then(() => (flushed = true)) + await Bun.sleep(20) + expect(flushed).toBe(false) // the flush is holding for the archive + net.release() + await flush + expect(settled).toBe(true) + } finally { + net.release() + await archive + net.restore() + } + }) + test("a create is repaired with a verbatim update", async () => { // A create runs an extractor that rewrites the text; update() is verbatim, // so every create is followed by one. diff --git a/packages/opencode/test/cli/run-accounting.test.ts b/packages/opencode/test/cli/run-accounting.test.ts index eee8b8888b..10893a0e99 100644 --- a/packages/opencode/test/cli/run-accounting.test.ts +++ b/packages/opencode/test/cli/run-accounting.test.ts @@ -509,6 +509,15 @@ describe("run command request/stream lifecycle contracts", () => { expect(source).toContain('"IdleDoneContinuationUnconfirmed"') }) + test("the exit path flushes pending skill syncs AND memory mirrors when the workspace pilot is on (#1332)", async () => { + const source = await Bun.file(new URL("../../src/cli/cmd/run.ts", import.meta.url).pathname).text() + expect(source).toMatch( + // Each flush must be AWAITED: a `void import(...)` keeps the same tokens and + // brings back the lost-upload race. (bot review) + /if \(CoreFlag\.ALTIMATE_WORKSPACE\) \{[\s\S]{0,400}?await Promise\.all\(\[\s*import\("\.\.\/\.\.\/altimate\/workspace\/skill-sync"\)\s*\.then\(\(m\) => m\.flushPendingSyncs\(\)\)[\s\S]{0,200}?import\("\.\.\/\.\.\/altimate\/workspace\/memory-sync"\)\s*\.then\(\(m\) => m\.flushPendingMirrors\(\)\)[\s\S]{0,100}?\]\)/, + ) + }) + test("an SSE-triggered request abort preserves the original stream failure", async () => { const source = await Bun.file(new URL("../../src/cli/cmd/run.ts", import.meta.url).pathname).text() expect(source).toMatch(