Skip to content

Commit fee2a0c

Browse files
ralphstodomingoralphstodomingo
committed
fix(workspace): announce late attachments, re-attribute cached successes, close the stale-config class
Codex round 10. Three findings, all verified; each fix confirmed to fail only its own test when reverted. The config-cache class fix is folded in as agreed. **A late attachment now announces itself.** MCP.add stores the client but publishes nothing, so an attach landing after the turn's bounded wait — or on a repair retry, which never waits — produced tools the session had no way to learn about until the user sent another message. This branch has been documenting a tools-changed fallback as the justification for the wait being safe; that fallback depended on an event nobody published. It is published now. **A cached success is re-attributed, not merely re-connected.** Link A to B and back to A, with another session attaching B in between, and this session's key matches its original memo while the instance-wide client is serving B. The round-8 re-probe only checked that something was connected, so every later turn would expose B's tools under binding A. The live entry's pin is now checked too, which is what makes it ours. **The optional catalog lookup can no longer block a local spawn.** declared() is reporting only, but it runs before the engine is launched and its HTTP layer has no abort timeout, so an API that accepts a connection and then stalls stopped a good cached binding and an installed engine from ever attaching — and later turns kept returning the same pending task, since a pending attach has no settled repairable outcome. It is bounded now; reporting degrades, attaching does not wait on it. **Config reads are fresh by construction.** Three separate bugs in this module came from reading a per-instance cache after someone else wrote: our own addMcpToConfig, MCP.disconnect writing enabled:false, and an IDE rewriting the entry — which never goes through Config at all. Two of them defeated a fix from an earlier round. The writers cannot be enumerated, so freshness belongs at the point of read: freshConfig() is now the module's only path to config, and the single reader goes through it, which makes all three call sites correct without touching them. The ad-hoc invalidation added last round is removed rather than left as a second answer to the same question. The cost is named rather than hidden: invalidating drops the shared per-instance cache, so other Config consumers re-read. Also fixed in the test harness: the mocked config now reflects an entry after it is persisted, as production does. Without that the pin re-check saw no entry and a legitimate memo looked like a workspace change.
1 parent d6f5b5b commit fee2a0c

2 files changed

Lines changed: 164 additions & 25 deletions

File tree

‎packages/opencode/src/altimate/workspace/engine-sync.ts‎

Lines changed: 79 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -61,7 +61,7 @@ import { Flag as CoreFlag } from "@opencode-ai/core/flag/flag"
6161
import { which as whichBinary } from "@opencode-ai/core/util/which"
6262
import { Instance } from "@/project/instance"
6363
import { Log } from "@/altimate/util/log"
64-
import { MCP } from "@/mcp"
64+
import { MCP, ToolsChanged } from "@/mcp"
6565
import { addMcpToConfig, resolveConfigPath } from "@/mcp/config"
6666
import { Config } from "@/config/config"
6767
import { AltimateApi } from "@/altimate/api/client"
@@ -73,6 +73,9 @@ import { readLocalBinding, type CachedBinding } from "./state"
7373

7474
const log = Log.create({ service: "workspace-engine" })
7575

76+
/** How long the optional allowlist lookup may delay a local spawn. */
77+
const DECLARED_TIMEOUT_MS = 4_000
78+
7679
/** Oldest engine this client is known to work against.
7780
*
7881
* 0.7.0 is the first engine that LOCKS the `--datamate` pin, so a settings
@@ -132,7 +135,8 @@ export const syncInternals: {
132135
persist?: (name: string, cfg: LocalMcpConfig) => Promise<void>
133136
/** The configured (merged) MCP entry under `name`, or null if none. */
134137
existingEntry?: (name: string) => Promise<ExistingEntry | null>
135-
refreshConfig?: () => Promise<void>
138+
freshConfig?: () => Promise<{ mcp?: Record<string, ExistingEntry | undefined> }>
139+
toolsChanged?: () => Promise<void>
136140
declared?: (datamateId: string) => Promise<Declared | null>
137141
notify?: (toast: Toast) => Promise<void>
138142
} = {}
@@ -290,22 +294,30 @@ async function persist(name: string, cfg: LocalMcpConfig): Promise<void> {
290294
})
291295
}
292296

293-
/** Drop the per-instance config cache so the next read sees the file.
297+
/** The module's ONLY path to config, and it is always fresh.
298+
*
299+
* `Config.get()` is cached per instance, and this module has now been bitten
300+
* three times by reading it after someone else wrote: our own `addMcpToConfig`,
301+
* `MCP.disconnect` writing `enabled: false`, and an IDE rewriting the entry —
302+
* which never goes through `Config` at all. Two of those defeated a fix from an
303+
* earlier round.
294304
*
295-
* `MCP.disconnect` writes `enabled: false` straight to disk and does not
296-
* invalidate `Config`, so a cached entry can still report `enabled: true` right
297-
* after a user has disconnected. */
298-
async function refreshConfig(): Promise<void> {
299-
if (syncInternals.refreshConfig) return syncInternals.refreshConfig()
305+
* Enumerating the writers is therefore not possible, so freshness is structural
306+
* at the point of READ rather than remembered at each write site. The cost is
307+
* real and shared: invalidating drops the per-instance cache for every other
308+
* `Config` consumer too. That is the price of not having a fourth instance. */
309+
async function freshConfig(): Promise<{ mcp?: Record<string, ExistingEntry | undefined> }> {
310+
if (syncInternals.freshConfig) return syncInternals.freshConfig()
300311
await Config.invalidate().catch((err) => {
301312
log.warn("could not refresh the config cache", { err: String(err) })
302313
})
314+
return (await Config.get()) as { mcp?: Record<string, ExistingEntry | undefined> }
303315
}
304316

305317
async function existingEntry(name: string): Promise<ExistingEntry | null> {
306318
if (syncInternals.existingEntry) return syncInternals.existingEntry(name)
307319
try {
308-
const cfg = (await Config.get()) as { mcp?: Record<string, ExistingEntry | undefined> }
320+
const cfg = await freshConfig()
309321
return cfg.mcp?.[name] ?? null
310322
} catch (err) {
311323
log.warn("could not read merged MCP config", { name, err: String(err) })
@@ -378,6 +390,24 @@ async function declared(datamateId: string): Promise<Declared | null> {
378390
}
379391
}
380392

393+
/** Tell the session its tool list changed.
394+
*
395+
* `MCP.add` stores the client but publishes nothing, so an attach that lands
396+
* after the turn's bounded wait — or on a repair retry, which never waits —
397+
* produced tools the session had no way to learn about until the user sent
398+
* another message. The documented fallback for exceeding the cap depends on
399+
* this event existing, so it has to be published here. */
400+
async function announceToolsChanged(): Promise<void> {
401+
if (syncInternals.toolsChanged) return syncInternals.toolsChanged()
402+
try {
403+
await AppRuntime.runPromise(
404+
EventV2Bridge.Service.use((events) => events.publish(ToolsChanged, { server: DATAMATE_KEY })),
405+
)
406+
} catch (err) {
407+
log.warn("could not announce the workspace engine tool change", { err: String(err) })
408+
}
409+
}
410+
381411
async function notify(toast: Toast): Promise<void> {
382412
if (syncInternals.notify) return syncInternals.notify(toast)
383413
try {
@@ -524,8 +554,8 @@ async function run(): Promise<Outcome> {
524554
// user disconnects. Treating that as a synthesized status would reconnect
525555
// and persist it enabled again, undoing their disconnect — globally, if
526556
// the owning entry is global. Read the file before deciding.
527-
if (existing.status === "disabled") await refreshConfig()
528-
const owning = existing.status === "disabled" ? await existingEntry(DATAMATE_KEY) : entry
557+
// `existingEntry` is always fresh now, so `entry` already reflects disk.
558+
const owning = entry
529559
if (existing.status === "disabled" && owning?.enabled === false) {
530560
// The user turned this entry off deliberately. Do NOT call `MCP.connect`
531561
// to "retry" it: that persists `enabled: true` into whichever config
@@ -662,7 +692,23 @@ async function run(): Promise<Outcome> {
662692
}
663693
}
664694

665-
const declaredKeys = await declared(workspaceId)
695+
// Bounded: this lookup is reporting only, but it runs BEFORE the engine is
696+
// launched and its HTTP layer has no abort timeout — so an API that accepts a
697+
// connection and then stalls stopped a good cached binding and an installed
698+
// engine from ever attaching. Reporting degrades; attaching does not wait.
699+
const declaredKeys = await Promise.race([
700+
declared(workspaceId),
701+
new Promise<null>((resolve) => {
702+
const timer = setTimeout(() => {
703+
log.warn("workspace allowlist lookup timed out; attaching without the declared-vs-delivered report", {
704+
workspaceId,
705+
timeoutMs: DECLARED_TIMEOUT_MS,
706+
})
707+
resolve(null)
708+
}, DECLARED_TIMEOUT_MS)
709+
timer.unref?.()
710+
}),
711+
])
666712
const declaredCount = declaredKeys?.keys.length ?? 0
667713

668714
// Rule 2 / 3 — opportunistic use, or an offer. Never an install.
@@ -730,6 +776,10 @@ async function run(): Promise<Outcome> {
730776
return { kind: "connect-failed", error }
731777
}
732778

779+
// The add succeeded and is ours: announce it, so a turn that had already given
780+
// up waiting still learns the tools arrived.
781+
await announceToolsChanged()
782+
733783
// Rule 5 — report declared-but-missing.
734784
const present = engineToolKeys(await client.tools())
735785
const missing = declaredKeys ? declaredKeys.keys.filter((k) => !present.has(k)) : []
@@ -795,11 +845,16 @@ function wasServing(outcome: Outcome | undefined): boolean {
795845
* reconnected until a new session or a re-link.
796846
*
797847
* Fails OPEN: a status read that throws must not invalidate a good attach. */
798-
async function engineStillConnected(): Promise<boolean> {
848+
async function engineStillOurs(workspaceId: string): Promise<boolean> {
799849
try {
800-
return (await mcp().status())[DATAMATE_KEY]?.status === "connected"
850+
if ((await mcp().status())[DATAMATE_KEY]?.status !== "connected") return false
851+
// Connected is not enough. Link A -> B -> A with another session attaching B
852+
// in between, and this session's key matches its original memo while the
853+
// instance-wide client is serving B — so the cached success would expose B's
854+
// tools under binding A. The pin is what makes it ours.
855+
return pinnedWorkspace(await existingEntry(DATAMATE_KEY)) === workspaceId
801856
} catch (err) {
802-
log.warn("could not re-probe the engine connection; keeping the cached attach", { err: String(err) })
857+
log.warn("could not re-probe the engine attribution; keeping the cached attach", { err: String(err) })
803858
return true
804859
}
805860
}
@@ -840,6 +895,13 @@ export function trackedSessionsForTests(): number {
840895
* workspace would keep serving the old workspace's tools, both silently and for
841896
* the rest of the session. Keying on the bound workspace makes a re-link produce
842897
* a fresh attach on the next turn and leaves everything else memoised as before. */
898+
/** The bound workspace id, or null when unbound or disabled. */
899+
async function attachKeyWorkspace(): Promise<string | null> {
900+
if (!isEnabled()) return null
901+
const binding = await resolveBinding()
902+
return binding ? String(binding.datamateId) : null
903+
}
904+
843905
async function attachKey(): Promise<string> {
844906
if (!isEnabled()) return "disabled"
845907
const binding = await resolveBinding()
@@ -877,7 +939,8 @@ export function ensure(sessionID: string): Promise<Outcome> {
877939
// the hint it produced.
878940
if (sameWorkspace && !isRepairable(previous!.outcome)) {
879941
// Re-probe before trusting a cached success — see `engineStillConnected`.
880-
if (!wasServing(previous!.outcome) || (await engineStillConnected())) return previous!.task
942+
const boundTo = await attachKeyWorkspace()
943+
if (!wasServing(previous!.outcome) || !boundTo || (await engineStillOurs(boundTo))) return previous!.task
881944
log.info("cached attach is no longer connected; re-attaching", { sessionID })
882945
}
883946
entry.key = key

‎packages/opencode/test/altimate/workspace/engine-sync.test.ts‎

Lines changed: 85 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,7 @@ import {
2222
type LocalMcpConfig,
2323
} from "../../../src/altimate/workspace/engine-sync"
2424
import type { CachedBinding } from "../../../src/altimate/workspace/state"
25+
import type { ExistingEntry } from "../../../src/altimate/workspace/engine-sync"
2526

2627
const ORIGINAL_FLAG = process.env.ALTIMATE_WORKSPACE
2728

@@ -38,6 +39,7 @@ type Harness = {
3839
connects: string[]
3940
removes: string[]
4041
toasts: Array<{ title: string; message: string; variant: string }>
42+
toolsChanged: number
4143
statusQueue: Array<Record<string, { status: string; error?: string } | undefined>>
4244
tools: Record<string, unknown>
4345
}
@@ -57,6 +59,7 @@ function install(opts: {
5759
connects: [],
5860
removes: [],
5961
toasts: [],
62+
toolsChanged: 0,
6063
statusQueue: opts.statuses ?? [{}],
6164
tools: opts.tools ?? {},
6265
}
@@ -71,10 +74,20 @@ function install(opts: {
7174
syncInternals.persist = async (name, cfg) => {
7275
h.persisted.push({ name, cfg })
7376
}
74-
syncInternals.existingEntry = async () => (opts.existing === undefined ? null : opts.existing)
77+
syncInternals.existingEntry = async () => {
78+
if (opts.existing !== undefined) return opts.existing
79+
// Production persists the pinned entry before adding it, so a later read
80+
// sees it. Without this the harness under-reports and a legitimate memo
81+
// looks like a workspace change.
82+
const last = h.persisted[h.persisted.length - 1]
83+
return last ? ({ type: "local", command: last.cfg.command, enabled: true } as ExistingEntry) : null
84+
}
7585
syncInternals.notify = async (toast) => {
7686
h.toasts.push(toast)
7787
}
88+
syncInternals.toolsChanged = async () => {
89+
h.toolsChanged += 1
90+
}
7891
syncInternals.mcp = {
7992
status: async () => h.statusQueue.length > 1 ? h.statusQueue.shift()! : h.statusQueue[0]!,
8093
add: async (name, cfg) => {
@@ -1081,20 +1094,19 @@ describe("ensure — round 9", () => {
10811094
// so the cached entry still says enabled:true. Believing the cache would
10821095
// reconnect the entry and persist it enabled again — undoing the user's
10831096
// disconnect, globally if the owning entry is global.
1084-
let refreshed = false
1097+
let reads = 0
10851098
const h = install({
10861099
statuses: [{ datamate: { status: "disabled" } }],
10871100
})
1088-
syncInternals.refreshConfig = async () => {
1089-
refreshed = true
1101+
// Reads go through freshConfig now, so the disk value is what is seen.
1102+
syncInternals.existingEntry = undefined
1103+
syncInternals.freshConfig = async () => {
1104+
reads += 1
1105+
return { mcp: { datamate: { type: "local", command: ["datamate", "start-stdio"], enabled: false } } }
10901106
}
1091-
syncInternals.existingEntry = async () =>
1092-
refreshed
1093-
? { type: "local", command: ["datamate", "start-stdio"], enabled: false } // on disk
1094-
: { type: "local", command: ["datamate", "start-stdio"], enabled: true } // stale cache
10951107

10961108
expect(await ensure("s1")).toEqual({ kind: "entry-disabled" })
1097-
expect(refreshed).toBe(true)
1109+
expect(reads).toBeGreaterThan(0)
10981110
expect(h.connects).toHaveLength(0) // MCP.connect would persist enabled:true
10991111
expect(h.persisted).toHaveLength(0)
11001112
})
@@ -1113,3 +1125,67 @@ describe("ensure — round 9", () => {
11131125
expect(old.toasts[0].message).toContain("needs 0.7.0 or newer")
11141126
})
11151127
})
1128+
1129+
describe("ensure — round 10", () => {
1130+
test("a successful add announces the new tools", async () => {
1131+
// MCP.add stores the client but publishes nothing, so a late attach — after
1132+
// the bounded wait expired, or on a repair retry — left the session with
1133+
// tools it had no way to learn about until another user turn.
1134+
const h = install({
1135+
statuses: [{}, { datamate: { status: "connected" } }],
1136+
tools: { datamate_dbt_build_model: 1 },
1137+
})
1138+
expect(await ensure("s1")).toMatchObject({ kind: "attached" })
1139+
expect(h.toolsChanged).toBe(1)
1140+
})
1141+
1142+
test("a cached success is re-attributed, not merely re-connected", async () => {
1143+
// A -> B -> A while another session attached B: the key matches this
1144+
// session's original memo and the client is connected, but it is serving B.
1145+
let pinned = "42"
1146+
const h = install({
1147+
statuses: [{}, { datamate: { status: "connected" } }, { datamate: { status: "connected" } }, {}, { datamate: { status: "connected" } }],
1148+
tools: { datamate_dbt_build_model: 1 },
1149+
})
1150+
syncInternals.existingEntry = async () => ({
1151+
type: "local",
1152+
command: ["datamate", "start-stdio", "--datamate", pinned],
1153+
})
1154+
const first = await ensure("s1")
1155+
expect(first).toMatchObject({ kind: "attached" })
1156+
1157+
pinned = "99" // the live entry now serves another workspace
1158+
const second = await ensure("s1")
1159+
expect(second).not.toBe(first)
1160+
})
1161+
1162+
test("a stalled catalog lookup cannot block the local engine", async () => {
1163+
const h = install({
1164+
statuses: [{}, { datamate: { status: "connected" } }],
1165+
tools: { datamate_dbt_build_model: 1 },
1166+
})
1167+
syncInternals.declared = () => new Promise(() => {}) // API accepts then stalls
1168+
const outcome = await ensure("s1")
1169+
// The engine is on PATH and the binding is cached; reporting is optional.
1170+
expect(outcome).toMatchObject({ kind: "attached" })
1171+
expect(h.added).toHaveLength(1)
1172+
})
1173+
1174+
test("config is read fresh, so an external write is never missed", async () => {
1175+
// Nothing in this module can enumerate the writers — MCP writes raw, and an
1176+
// IDE rewriting the entry never touches Config at all — so freshness has to
1177+
// be structural at the point of read.
1178+
let onDisk: Record<string, unknown> = { type: "local", command: ["datamate", "start-stdio"], enabled: true }
1179+
let invalidations = 0
1180+
const h = install({ statuses: [{ datamate: { status: "disabled" } }] })
1181+
syncInternals.existingEntry = undefined // let the real reader go through freshConfig
1182+
syncInternals.freshConfig = async () => {
1183+
invalidations += 1
1184+
return { mcp: { datamate: onDisk as ExistingEntry } }
1185+
}
1186+
onDisk = { type: "local", command: ["datamate", "start-stdio"], enabled: false }
1187+
expect(await ensure("s1")).toEqual({ kind: "entry-disabled" })
1188+
expect(invalidations).toBeGreaterThan(0)
1189+
expect(h.connects).toHaveLength(0)
1190+
})
1191+
})

0 commit comments

Comments
 (0)