InfinityPacer · Nov 8, 2026 · Nov 8, 2026 · Nov 8, 2026 · Nov 8, 2026
diff --git a/package.json b/package.json
@@ -1,6 +1,6 @@
 {
   "name": "meterleaf",
-  "version": "0.16.2",
+  "version": "0.16.3",
   "private": true,
   "type": "module",
   "license": "Apache-2.0",

diff --git a/src/collector/agy.ts b/src/collector/agy.ts
@@ -1084,7 +1084,7 @@ export function readAgy(
     if (cache.plan) plans[label] = cache.plan;
     const effectivePlan = plans[label] ?? null;
     const accounts = agyAccountsForLabel(label, effectivePlan);
-    state.db.transaction(() => {
+    state.write(() => {
       state.upsertAccount(accounts.account);
       for (const legacy of accounts.legacy) state.upsertAccount(legacy);
       if (cache.entries.length > 0) {
@@ -1094,7 +1094,7 @@ export function readAgy(
         }
       }
       state.setMeta("agy_plans", JSON.stringify(plans));
-    })();
+    });
   }
 
   const activeAccount = activeAgyLabel(sources.statuslineDir);
@@ -1204,7 +1204,7 @@ export function readAgy(
       const steps = buildStepTimeIndex(opened.db);
       const conversationId = basename(path, ".db");
 
-      state.db.transaction(() => {
+      state.write(() => {
         for (const row of genRows) {
           if (
             (typeof row.data !== "string" &&
@@ -1299,7 +1299,7 @@ export function readAgy(
             account: activeAccount,
           });
         }
-      })();
+      });
     } finally {
       opened.close();
     }

diff --git a/src/collector/codex.ts b/src/collector/codex.ts
@@ -573,7 +573,7 @@ export async function readCodex(
       report.filesRead += 1;
       report.bytes += text.length;
       const context = emptyContext();
-      state.db.transaction(() => {
+      state.write(() => {
         const lines = text.toString("utf8");
         let from = 0;
         while (from < lines.length) {
@@ -585,7 +585,7 @@ export async function readCodex(
         }
         commitObservations(sinks, accounts);
         save(stats.size, "", "");
-      })();
+      });
       return true;
     }
     // 未压缩文件仍会被追加：从上次完整行边界继续，未写完的行留到下一轮。
@@ -614,7 +614,7 @@ export async function readCodex(
       const block = Buffer.alloc(chunkBytes);
       // 进度只推进到最后一个完整行之后；末尾未写完的行下一轮重读。
       let consumed = offset;
-      state.db.transaction(() => {
+      state.write(() => {
         while (offset < length) {
           const chunkStart = offset;
           const read = readSync(
@@ -657,7 +657,7 @@ export async function readCodex(
         context.discarding = discarding;
         commitObservations(sinks, accounts);
         save(consumed, JSON.stringify(context), tailFingerprint(fd, consumed));
-      })();
+      });
       return true;
     } finally {
       closeSync(fd);

diff --git a/src/collector/omp.ts b/src/collector/omp.ts
@@ -441,7 +441,7 @@ export async function readOmp(
       const block = Buffer.alloc(chunkBytes);
       // 进度只推进到最后一个完整行之后；末尾未写完的行下一轮重读。
       let consumed = offset;
-      state.db.transaction(() => {
+      state.write(() => {
         while (offset < length) {
           const chunkStart = offset;
           const read = readSync(
@@ -494,7 +494,7 @@ export async function readOmp(
             size: stats.size,
             modified: stats.mtimeMs,
           });
-      })();
+      });
       return true;
     } finally {
       closeSync(fd);

diff --git a/src/collector/opencode.ts b/src/collector/opencode.ts
@@ -102,7 +102,7 @@ export function readOpenCode(
     }
     if (rows.length === 0) return report;
     let cursor = lastCursor ?? 0;
-    state.db.transaction(() => {
+    state.write(() => {
       for (const row of rows) {
         // 无论行内容结果如何，游标都推进到本轮读到的最新行。
         cursor = Math.max(cursor, row.time_updated);
@@ -181,7 +181,7 @@ export function readOpenCode(
       }
       // 游标与入账在同一事务提交；中途失败整体回滚，下一轮从原进度重读。
       state.saveOpencodeCursor(cursor);
-    })();
+    });
     return report;
   } finally {
     db.close();

diff --git a/src/collector/state.ts b/src/collector/state.ts
@@ -154,6 +154,16 @@ export class CollectorState {
     this.db.close();
   }
 
+  /**
+   * 同步与 OTEL 中继是两个进程，各自写这个库。所有写事务都在开始时拿写锁：
+   * 默认的延迟事务先读后写，读取后若另一个进程已经提交，升级写锁时 SQLite
+   * 直接返回“database is locked”，不经过 busy_timeout 等待，整轮同步因此失败。
+   * 嵌套调用按保存点执行，只有最外层决定加锁方式。
+   */
+  write<T>(work: () => T): T {
+    return this.db.transaction(work).immediate();
+  }
+
   /** 设置模型账户映射；改变映射后由调用方重新归属全部事件。 */
   setModelAccounts(rules: readonly ModelAccountRule[]) {
     this.modelAccounts = rules;
@@ -548,7 +558,7 @@ export class CollectorState {
     observations: readonly FileObservation[],
   ): number {
     let changed = 0;
-    this.db.transaction(() => {
+    this.write(() => {
       for (const observation of observations) {
         if (this.mergeObservation(observation)) changed += 1;
       }
@@ -565,7 +575,7 @@ export class CollectorState {
           cursor.fingerprint,
           cursor.mtimeMs,
         );
-    })();
+    });
     return changed;
   }
 
@@ -657,7 +667,7 @@ export class CollectorState {
   }
 
   private enqueue(kind: OutboxKind, itemId: string, payload: unknown) {
-    this.db.transaction(() => this.enqueueRow(kind, itemId, payload))();
+    this.write(() => this.enqueueRow(kind, itemId, payload));
   }
 
   /**
@@ -669,7 +679,7 @@ export class CollectorState {
     dropped: number;
   } {
     if (records.length === 0) return { queued: 0, dropped: 0 };
-    return this.db.transaction(() => {
+    return this.write(() => {
       for (const record of records)
         this.enqueueRow("otel", record.requestId, record);
       // 服务端长期收不到时队列不能无限增长：只保留最新的一批，更早的请求
@@ -682,7 +692,7 @@ export class CollectorState {
         )
         .run(OTEL_QUEUE_LIMIT).changes;
       return { queued: records.length, dropped };
-    })();
+    });
   }
 
   /** 账户资料变化才进入待发送；返回是否有变化。 */
@@ -803,7 +813,7 @@ export class CollectorState {
             )
             .all(since);
     let changed = 0;
-    this.db.transaction(() => {
+    this.write(() => {
       for (const row of rows) {
         const fact = JSON.parse(row.observation) as UsageFact;
         const { account, attribution } = this.accountFor(
@@ -819,7 +829,7 @@ export class CollectorState {
           changed += 1;
         }
       }
-    })();
+    });
     return changed;
   }
 
@@ -829,7 +839,7 @@ export class CollectorState {
    */
   requeueUsage(): number {
     let queued = 0;
-    this.db.transaction(() => {
+    this.write(() => {
       for (const row of this.db
         .query<EventRow, []>(
           "SELECT key, occurred_at, total, account, observation FROM events WHERE total > 0",
@@ -844,7 +854,7 @@ export class CollectorState {
         )
10BC8
;
         queued += 1;
       }
-    })();
+    });
     return queued;
   }
 
@@ -856,7 +866,7 @@ export class CollectorState {
         ? "last_quota_hash"
         : `last_quota_hash:${source}`;
     if (this.getMeta(metaKey) === snapshot.hash) return false;
-    this.db.transaction(() => {
+    this.write(() => {
       for (const quota of snapshot.quotas) {
         this.enqueue(
           "quota",
@@ -865,7 +875,7 @@ export class CollectorState {
         );
       }
       this.setMeta(metaKey, snapshot.hash);
-    })();
+    });
     return true;
   }
 
@@ -898,9 +908,9 @@ export class CollectorState {
   /** 只确认实际发送的版本；同一条目若已有更新版本，其 seq 不同，不会被删除。 */
   acknowledge(seqs: readonly number[]) {
     const remove = this.db.query("DELETE FROM outbox WHERE seq = ?");
-    this.db.transaction(() => {
+    this.write(() => {
       for (const seq of seqs) remove.run(seq);
-    })();
+    });
   }
 
   /** 非零事件按模型汇总，供 scan 与 status 对账。 */

diff --git a/tests/collector-agy.test.ts b/tests/collector-agy.test.ts
@@ -798,7 +798,7 @@ const usageJson = (fractions: Record<string, number>) =>
             buckets: Object.entries(fractions).map(([id, remaining]) => ({
               id,
               remaining_fraction: remaining,
-              reset_time: "2026-11-06T10:58:15Z",
+              reset_time: "2100-01-01T00:00:00Z",
             })),
           },
         ],
@@ -904,10 +904,10 @@ test("maybeRefreshAgyQuota runs agy, keeps the plan row, and replaces only refre
     // 没刷到的 3p_5h 与 plan 行原样保留，刷到的窗口换成新值（已用百分比取整）。
     expect(readFileSync(w.quota, "utf8")).toBe(
       [
-        "gemini_5h\t12\t1793962695",
-        "gemini_weekly\t26\t1793962695",
+        "gemini_5h\t12\t4102444800",
+        "gemini_weekly\t26\t4102444800",
         "3p_5h\t10\t1793409600",
-        "3p_weekly\t1\t1793962695",
+        "3p_weekly\t1\t4102444800",
         "plan\tGoogle AI Ultra\t-",
       ].join("\n") + "\n",
     );
@@ -1219,7 +1219,7 @@ test("a first-time omp Antigravity request is held until the refresh reveals the
       "bob\n",
     );
     expect(readFileSync(join(w.statuslineDir, "quota-bob.tsv"), "utf8")).toBe(
-      "gemini_5h\t50\t1793962695\n3p_5h\t10\t1793962695\n",
+      "gemini_5h\t50\t4102444800\n3p_5h\t10\t4102444800\n",
     );
     const omp = reports.find((item) => item.source === "omp")!;
     expect(omp).toMatchObject({ read: 1, recorded: 1, skipped: {} });
@@ -1276,7 +1276,7 @@ function ompReport(email: string, used: Record<string, number>) {
       (bucket.startsWith("3p") ? ["anthropic", "openai"] : ["google"]).map(
         (family) => ({
           id: `google-antigravity:${family}:default:${bucket}`,
-          window: { resetsAt: 1793962695000 },
+          window: { resetsAt: 4102444800000 },
           amount: { remainingFraction: 1 - fraction },
         }),
       ),
@@ -1323,15 +1323,15 @@ test("maybeRefreshOmpAgyQuota writes every Google account omp holds and reads li
     expect(omp.calls()).toEqual(["--json"]);
     expect(readFileSync(w.quota, "utf8")).toBe(
       [
-        "gemini_5h\t12\t1793962695",
-        "gemini_weekly\t100\t1793962695",
+        "gemini_5h\t12\t4102444800",
+        "gemini_weekly\t100\t4102444800",
         "3p_5h\t10\t1793409600",
-        "3p_weekly\t1\t1793962695",
+        "3p_weekly\t1\t4102444800",
         "plan\tGoogle AI Ultra\t-",
       ].join("\n") + "\n",
     );
     expect(readFileSync(join(w.statuslineDir, "quota-bob.tsv"), "utf8")).toBe(
-      "gemini_5h\t2\t1793962695\n",
+      "gemini_5h\t2\t4102444800\n",
     );
     // omp 刷新不改 agy 的 current-account。
     expect(readdirSync(w.statuslineDir).sort()).toEqual([

diff --git a/tests/collector-state.test.ts b/tests/collector-state.test.ts
@@ -519,3 +519,36 @@ describe("切换账户时的登录线索", () => {
     expect(loginMoment('{"type":"user"')).toBeNull();
   });
 });
+
+describe("与 OTEL 中继并发写入", () => {
+  test("同步的写事务读取之后，中继的提交不会让同步失败", () => {
+    const { root } = workspace();
+    const path = join(root, "state.sqlite");
+    const sync = new CollectorState(path);
+    const relay = new CollectorState(path);
+    // 中继不等待，锁冲突立即暴露，测试不必耗满 5 秒。
+    relay.db.exec("PRAGMA busy_timeout = 0;");
+    const record = {
+      requestId: "req_1",
+      accountUuid: FIXTURE_ACCOUNT_UUID,
+      organizationUuid: null,
+      sessionId: null,
+      model: null,
+      occurredAt: null,
+    };
+    try {
+      sync.write(() => {
+        expect(sync.getMeta("probe")).toBeNull();
+        // 延迟事务下中继能在这里提交，随后同步的写入会被 SQLite 直接拒绝。
+        expect(() => relay.enqueueOtel([record])).toThrow("database is locked");
+        sync.setMeta("probe", "written");
+      });
+      expect(relay.getMeta("probe")).toBe("written");
+      expect(relay.enqueueOtel([record]).queued).toBe(1);
+      expect(sync.pendingCounts().otel).toBe(1);
+    } finally {
+      relay.close();
+      sync.close();
+    }
+  });
+});