From 620fef10a20f4352d948758c1489c73fc8bed381 Mon Sep 17 00:00:00 2001 From: Cameron Pfiffer Date: Fri, 31 Jul 2026 23:07:44 -0700 Subject: [PATCH] Harden Coil accounting and Auto session setup. MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Recover inference budget rows across Jazz visibility boundaries and let Letta Auto choose its own supported reasoning tier. 👾 Generated with [Letta Code](https://letta.com) Co-Authored-By: Letta Code --- agents/coil-public-knowledge.yaml | 1 - src/jazz/store.ts | 65 ++++++++++++++++++-------- test/public-knowledge-consumer.test.ts | 2 - 3 files changed, 45 insertions(+), 23 deletions(-) diff --git a/agents/coil-public-knowledge.yaml b/agents/coil-public-knowledge.yaml index adce16f..ebc9bd7 100644 --- a/agents/coil-public-knowledge.yaml +++ b/agents/coil-public-knowledge.yaml @@ -39,7 +39,6 @@ runner: dreaming: trigger: off model: letta/auto - reasoningEffort: low maxOutputTokens: 2000 timeoutMs: 180000 accounting: diff --git a/src/jazz/store.ts b/src/jazz/store.ts index ba36e18..468252b 100644 --- a/src/jazz/store.ts +++ b/src/jazz/store.ts @@ -67,6 +67,7 @@ export class JazzThoughtStore { private readonly runtimeRevision: string; private producerStorageRevision = 3; private consumerStorageRevision = 5; + private inferenceStorageRevision = 2; constructor(private readonly options: JazzThoughtStoreOptions) { this.registry = options.registry ?? createDefaultRegistry(); @@ -440,13 +441,19 @@ export class JazzThoughtStore { private async reserveInferenceUnlocked(request: InferenceReservationRequest): Promise { assertReservationRequest(request); const accountId = inferenceAccountId(request.scopeType, request.scopeKey); - const accountRowId = jazzRowId("inference-budget-account", accountId); const recordRowId = jazzRowId("inference-accounting", request.reservationId); - await this.ensureInferenceBudgetAccount(request.scopeType, request.scopeKey, accountId, accountRowId, request.reservedAt); + const accountSnapshot = await this.ensureInferenceBudgetAccount( + request.scopeType, + request.scopeKey, + accountId, + request.reservedAt, + ); await this.db.all(thoughtstreamApp.inferenceAccounting.where({ id: recordRowId }).limit(1), { tier: this.durabilityTier }); const result = await this.db.transaction(async (tx) => { - const accountRow = await tx.one(thoughtstreamApp.inferenceBudgetAccounts.where({ id: accountRowId }).limit(1)); - if (!accountRow) throw new Error(`Inference budget account disappeared during reservation: ${accountId}`); + // TransactionScope reads do not share Db's readiness barrier. Use the + // durable pre-transaction snapshot instead of asking a fresh transaction + // to rediscover an account created immediately before reservation. + const accountRow = accountSnapshot; const account = budgetAccountFromJazz(accountRow); await expireStaleReservations(tx, account, request.reservedAt); @@ -527,16 +534,21 @@ export class JazzThoughtStore { const snapshot = await this.getInferenceAccountingRecord(reservationId); if (!snapshot) throw new Error(`Cannot settle missing inference reservation ${reservationId}`); const accountId = inferenceAccountId(snapshot.scopeType, snapshot.scopeKey); - const accountRowId = jazzRowId("inference-budget-account", accountId); - const recordRowId = jazzRowId("inference-accounting", reservationId); - await this.db.all(thoughtstreamApp.inferenceBudgetAccounts.where({ id: accountRowId }).limit(1), { tier: this.durabilityTier }); + const [recordSnapshot] = await this.db.all(thoughtstreamApp.inferenceAccounting.where({ + key: reservationId, + }).limit(1), { tier: this.durabilityTier }); + if (!recordSnapshot) throw new Error(`Cannot settle missing inference reservation ${reservationId}`); + const [accountSnapshot] = await this.db.all(thoughtstreamApp.inferenceBudgetAccounts.where({ + key: accountId, + }).limit(1), { tier: this.durabilityTier }); + if (!accountSnapshot) throw new Error(`Cannot settle inference reservation without account ${accountId}`); const result = await this.db.transaction(async (tx) => { - const recordRow = await tx.one(thoughtstreamApp.inferenceAccounting.where({ id: recordRowId }).limit(1)); - if (!recordRow) throw new Error(`Cannot settle missing inference reservation ${reservationId}`); + // Use durable snapshots rather than visibility-sensitive TransactionScope + // reads for rows created by the immediately preceding reservation. + const recordRow = recordSnapshot; const record = inferenceAccountingFromJazz(recordRow); if (record.status !== "reserved") return record; - const accountRow = await tx.one(thoughtstreamApp.inferenceBudgetAccounts.where({ id: accountRowId }).limit(1)); - if (!accountRow) throw new Error(`Cannot settle inference reservation without account ${accountId}`); + const accountRow = accountSnapshot; const account = budgetAccountFromJazz(accountRow); const normalizedUsage = normalizeInferenceUsage(usage); const charged = chargedInferenceUsage(record.estimate, normalizedUsage); @@ -942,11 +954,12 @@ export class JazzThoughtStore { scopeType: InferenceReservationRequest["scopeType"], scopeKey: string, accountId: string, - accountRowId: string, at: string, - ): Promise { - const rows = await this.db.all(thoughtstreamApp.inferenceBudgetAccounts.where({ id: accountRowId }).limit(1), { tier: this.durabilityTier }); - if (rows.length > 0) return; + ): Promise> { + const existing = await this.db.all(thoughtstreamApp.inferenceBudgetAccounts.where({ + key: accountId, + }).limit(1), { tier: this.durabilityTier }); + if (existing[0]) return existing[0]; const account = emptyBudgetAccount(scopeType, scopeKey); const data = { key: accountId, @@ -956,12 +969,24 @@ export class JazzThoughtStore { activeLeasesJson: canonicalJson(asJsonValue(account.activeLeases)), updatedAt: at, }; - try { - await this.wait(this.db.insert(thoughtstreamApp.inferenceBudgetAccounts, data, { id: accountRowId })); - } catch (error) { - if (!(error instanceof Error) || !error.message.includes("object already exists")) throw error; - await this.db.all(thoughtstreamApp.inferenceBudgetAccounts.where({ id: accountRowId }).limit(1), { tier: this.durabilityTier }); + for (let attempt = 0; attempt < 8; attempt += 1) { + const storageRevision = this.inferenceStorageRevision; + try { + await this.wait(this.db.insert(thoughtstreamApp.inferenceBudgetAccounts, data, { + id: jazzRowId(`inference-budget-account-v${storageRevision}`, accountId), + })); + } catch (error) { + if (!isRecoverableStorageError(error) || attempt === 7) throw error; + this.inferenceStorageRevision = storageRevision + 1; + } + const rows = await this.db.all(thoughtstreamApp.inferenceBudgetAccounts.where({ + key: accountId, + }).limit(1), { tier: this.durabilityTier }); + if (rows[0]) return rows[0]; + this.inferenceStorageRevision = Math.max(this.inferenceStorageRevision, storageRevision + 1); + await new Promise((resolve) => setTimeout(resolve, Math.min(500, 20 * (2 ** attempt)))); } + throw new Error(`Inference budget account exhausted storage recovery attempts: ${accountId}`); } private async upsert(table: JazzTable, query: Record, data: Record): Promise { diff --git a/test/public-knowledge-consumer.test.ts b/test/public-knowledge-consumer.test.ts index 2118fd4..ae20633 100644 --- a/test/public-knowledge-consumer.test.ts +++ b/test/public-knowledge-consumer.test.ts @@ -161,7 +161,6 @@ describe("Coil Public Knowledge consumer", () => { env: { MEMORY_DIR: "/tmp/co-memory" }, filesystemConfinement: "memory", model: "letta/auto", - reasoningEffort: "low", }); const [binding] = await fixture.store.listLettaConversationBindings(); expect(binding).toMatchObject({ @@ -387,7 +386,6 @@ function publicKnowledgeDeclaration(): ThoughtAgentDeclaration { permissionMode: "strict", skillSources: [], memoryDirEnv: "THOUGHTSTREAM_LETTA_CO_MEMORY_DIR", - reasoningEffort: "low", dreaming: { trigger: "off" }, sandbox: { ttlMinutes: 5, terminateOnClose: false }, }, -- 2.51.2