PavelRavvich · Nov 4, 2026 · Nov 4, 2026
diff --git a/.agboss-version.json b/.agboss-version.json
@@ -2,6 +2,6 @@
   "repository": "agbossjs/agboss",
   "branch": "main",
   "source": "commit",
-  "frameworkSha": "06c4bc1df2880b0480475309ce998490dc4d0477",
+  "frameworkSha": "53216599ef55ecb94cf0649d03bd6cac73cb8aab",
   "version": "0.1.0"
 }
diff --git a/README.md b/README.md
@@ -74,21 +74,21 @@ import { AppEnvironment } from "./environments/app.environment.js";
 export class DeskApplication {}
 ```
 
-| Store                                   | Core contract                        | Tables                                                              |
-| --------------------------------------- | ------------------------------------ | ------------------------------------------------------------------- |
-| `PostgresExecutionLeaseRepository`      | `ExecutionLeaseRepository`           | `agboss_execution_leases`                                           |
-| `PostgresExecutionRecordRepository`     | `ExecutionRecordRepository`          | `agboss_threads`, `agboss_execution_records`, … (`0002`)            |
-| `PostgresExecutionTreeRepository`       | `ExecutionTreeRepository`            | `agboss_execution_tree_keys`                                        |
-| `PostgresPausedExecutionRepository`     | `PausedExecutionRepository`          | `agboss_paused_executions`, `agboss_pauses`                         |
-| `PostgresChildExecutionQueueRepository` | `ChildExecutionQueueRepository`      | `agboss_child_execution_queue`                                      |
-| `PostgresCheckpointSaver`               | LangGraph `BaseCheckpointSaver`      | `checkpoints`, `checkpoint_blobs`, `checkpoint_writes`              |
-| `PostgresFencedEffectRepository`        | `FencedEffectRepository<PoolClient>` | `agboss_fenced_effect_marks`, `…_results`, `…_business_results`     |
-| `PostgresOutboundCallRepository`        | `OutboundCallRepository`             | `agboss_outbound_calls` (and the leases' `superseded`, `0014`)      |
-| `PostgresExecutionLeaseEventRepository` | `ExecutionLeaseEventRepository`      | `agboss_execution_lease_events`                                     |
-| `PostgresCostLedgerRepository`          | `CostLedgerRepository`               | `agboss_cost_counters`, `agboss_cost_max_spend`                     |
-| `PostgresExecutionHistoryRepository`    | `ExecutionHistoryRepository`         | `agboss_execution_history`                                          |
-| `PostgresRateLimitRepository`           | `RateLimitRepository`                | `agboss_rate_limit_keys`, `…_slots`, `…_window_buckets`, `…_pauses` |
-| `PostgresMeshRequestRepository`         | `MeshRequestRepository`              | `agboss_mesh_requests`                                              |
+| Store                                   | Core contract                        | Tables                                                                 |
+| --------------------------------------- | ------------------------------------ | ---------------------------------------------------------------------- |
+| `PostgresExecutionLeaseRepository`      | `ExecutionLeaseRepository`           | `agboss_execution_leases`                                              |
+| `PostgresExecutionRecordRepository`     | `ExecutionRecordRepository`          | `agboss_threads`, `agboss_execution_records`, … (`0002`)               |
+| `PostgresExecutionTreeRepository`       | `ExecutionTreeRepository`            | `agboss_execution_tree_keys`                                           |
+| `PostgresPausedExecutionRepository`     | `PausedExecutionRepository`          | `agboss_paused_executions`, `agboss_pauses`                            |
+| `PostgresChildExecutionQueueRepository` | `ChildExecutionQueueRepository`      | `agboss_child_execution_queue`                                         |
+| `PostgresCheckpointSaver`               | LangGraph `BaseCheckpointSaver`      | `checkpoints`, `checkpoint_blobs`, `checkpoint_writes`                 |
+| `PostgresFencedEffectRepository`        | `FencedEffectRepository<PoolClient>` | `agboss_fenced_effect_marks`, `…_results`, `…_business_results`        |
+| `PostgresOutboundCallRepository`        | `OutboundCallRepository`             | `agboss_outbound_calls` (and the leases' `superseded`, `0014`)         |
+| `PostgresExecutionLeaseEventRepository` | `ExecutionLeaseEventRepository`      | `agboss_execution_lease_events`                                        |
+| `PostgresCostLedgerRepository`          | `CostLedgerRepository`               | `agboss_cost_counters`, `agboss_cost_max_spend`, `agboss_cost_entries` |
+| `PostgresExecutionHistoryRepository`    | `ExecutionHistoryRepository`         | `agboss_execution_history`                                             |
+| `PostgresRateLimitRepository`           | `RateLimitRepository`                | `agboss_rate_limit_keys`, `…_slots`, `…_window_buckets`, `…_pauses`    |
+| `PostgresMeshRequestRepository`         | `MeshRequestRepository`              | `agboss_mesh_requests`                                                 |
 
 With all of them listed, nothing of the application is per pod: budgets sum every pod's spend, a
 provider's limits hold across pods, `agboss records export` on any pod lists every pod's
@@ -322,6 +322,7 @@ a pod never waits on another's migration at start (both are safe: they serialize
 | `0019_execution_ids_from_outbound_calls` | none: the ids of before of every execution that made an outbound call (agbossjs/agboss#638)                                                             |
 | `0020_pauses_by_tree`                    | `agboss_pauses` unique per tree and step key; the ids of before of every execution that paused, and its tree (agbossjs/agboss#663)                      |
 | `0021_pause_answers_by_tree`             | `agboss_pause_answers` keyed per tree and step key: another tree's answer under the same step key is kept (agbossjs/agboss#676)                         |
+| `0022_cost_ledger_entries`               | `agboss_cost_entries`: each spend's `ledgerId`, so a step run again counts its spend once (agbossjs/agboss#727)                                         |
 | LangGraph                                | `checkpoints`, `checkpoint_blobs`, `checkpoint_writes`, `checkpoint_migrations`                                                                         |
 
 ## Text with a NUL character

diff --git a/api/postgres.api.md b/api/postgres.api.md
@@ -506,11 +506,11 @@ export interface SchemaVersions {
 // src/stores/postgres-child-execution-queue.repository.ts:155:9 - (ae-undocumented) Missing documentation for "completeByTenantIdAndExecutionId".
 // src/stores/postgres-child-execution-queue.repository.ts:176:9 - (ae-undocumented) Missing documentation for "deleteAllByTenantIdAndRootExecutionId".
 // src/stores/postgres-child-execution-queue.repository.ts:187:9 - (ae-undocumented) Missing documentation for "findQueueDepth".
-// src/stores/postgres-cost-ledger.repository.ts:56:3 - (ae-undocumented) Missing documentation for "ping".
-// src/stores/postgres-cost-ledger.repository.ts:60:9 - (ae-undocumented) Missing documentation for "findSpentByCounter".
-// src/stores/postgres-cost-ledger.repository.ts:70:9 - (ae-undocumented) Missing documentation for "saveAll".
-// src/stores/postgres-cost-ledger.repository.ts:81:9 - (ae-undocumented) Missing documentation for "findMaxSpendByAccount".
-// src/stores/postgres-cost-ledger.repository.ts:91:9 - (ae-undocumented) Missing documentation for "saveMaxSpend".
+// src/stores/postgres-cost-ledger.repository.ts:81:3 - (ae-undocumented) Missing documentation for "ping".
+// src/stores/postgres-cost-ledger.repository.ts:85:9 - (ae-undocumented) Missing documentation for "findSpentByCounter".
+// src/stores/postgres-cost-ledger.repository.ts:95:9 - (ae-undocumented) Missing documentation for "saveAll".
+// src/stores/postgres-cost-ledger.repository.ts:109:9 - (ae-undocumented) Missing documentation for "findMaxSpendByAccount".
+// src/stores/postgres-cost-ledger.repository.ts:119:9 - (ae-undocumented) Missing documentation for "saveMaxSpend".
 // src/stores/postgres-execution-history.repository.ts:48:3 - (ae-undocumented) Missing documentation for "ping".
 // src/stores/postgres-execution-history.repository.ts:52:9 - (ae-undocumented) Missing documentation for "save".
 // src/stores/postgres-execution-history.repository.ts:63:9 - (ae-undocumented) Missing documentation for "findByTenantIdAndExecutionId".

diff --git a/migrations/0022_cost_ledger_entries.sql b/migrations/0022_cost_ledger_entries.sql
@@ -0,0 +1,7 @@
+-- agbossjs/agboss#727: spend is written to the ledger the moment it is made, each record under its
+-- ledgerId (the step's key, its call, its caller, its number there). A step run again after a crash
+-- or a takeover saves its spend again under the same ids: the ledger adds an entry once — its id
+-- inserted here in the statement that adds its cost to the counters.
+CREATE TABLE agboss_cost_entries (
+  ledger_id text PRIMARY KEY
+);
diff --git a/src/stores/postgres-cost-ledger.repository.ts b/src/stores/postgres-cost-ledger.repository.ts
@@ -12,36 +12,61 @@ import { PostgresConnection } from "../connection/postgres-connection.js";
 import { changesOf, firstRowOf } from "../connection/query.js";
 
 /**
- * Adds `$5` to every counter of the arrays (one counter per index) in one statement: a counter
- * given twice counts twice (grouped first — one upsert may not touch a row twice), and the rows
- * are upserted in key order, so two saves of overlapping counters lock them in one order and never
- * deadlock.
+ * Adds to every counter of the arrays (one counter per index), in one statement: `$5` — the cost of
+ * the records without a `ledgerId` — plus the cost of each entry of `$6` / `$7` (ids, costs) whose id
+ * this statement inserts (agbossjs/agboss#727: an id already there, or inserted by a save that
+ * commits first, adds nothing; the unique key makes the later save wait for the earlier one). A
+ * counter given twice counts twice (grouped first — one upsert may not touch a row twice), and the
+ * rows are upserted in key order, so two saves of overlapping counters lock them in one order and
+ * never deadlock.
  */
-const ADD = `INSERT INTO agboss_cost_counters AS counter
+const ADD = `WITH inserted AS (
+    INSERT INTO agboss_cost_entries (ledger_id) SELECT unnest($6::text[])
+    ON CONFLICT (ledger_id) DO NOTHING
+    RETURNING ledger_id
+  ), spent AS (
+    SELECT $5::double precision + COALESCE(sum(entry.cost), 0) AS usd
+      FROM unnest($6::text[], $7::double precision[]) AS entry (ledger_id, cost)
+      WHERE entry.ledger_id IN (SELECT ledger_id FROM inserted)
+  )
+  INSERT INTO agboss_cost_counters AS counter
     (budget, scope, scope_key, period, spent_usd)
-  SELECT budget, scope, scope_key, period, count(*) * $5::double precision
+  SELECT budget, scope, scope_key, period, count(*) * (SELECT usd FROM spent)
     FROM unnest($1::text[], $2::text[], $3::text[], $4::text[])
       AS given (budget, scope, scope_key, period)
     GROUP BY budget, scope, scope_key, period
     ORDER BY budget, scope, scope_key, period
   ON CONFLICT (budget, scope, scope_key, period)
     DO UPDATE SET spent_usd = counter.spent_usd + EXCLUDED.spent_usd`;
 
+/** The entries of `records` (one per `ledgerId`, the first), and what the others cost. */
+function entriesOf(records: readonly UsageRecord[]): {
+  readonly ids: string[];
+  readonly costs: number[];
+  readonly unnamed: number;
+} {
+  const entries = new Map<string, number>();
+  let unnamed = 0;
+  for (const { ledgerId, costUsd } of records) {
+    if (ledgerId === undefined) unnamed += costUsd;
+    else if (!entries.has(ledgerId)) entries.set(ledgerId, costUsd);
+  }
+  return { ids: [...entries.keys()], costs: [...entries.values()], unnamed };
+}
+
 const SpentRow = z.object({ spent_usd: z.number() });
 const MaxSpendRow = z.object({ max_spend_usd: z.number() });
 
 /** The scope as stored: one of `BudgetScope`'s values (anything else is refused). */
 const scopeOf = (account: BudgetAccount): string => z.enum(BudgetScope).parse(account.scope);
 
-/** What the calls cost together, in USD. */
-const costOf = (records: readonly UsageRecord[]): number =>
-  records.reduce((sum, record) => sum + record.costUsd, 0);
-
 /**
  * `CostLedgerRepository` on PostgreSQL (agbossjs/agboss#483): one row per counter — budget, scope,
  * scope key and period — so the instances of an application share a tenant's day. `saveAll` is
- * one `INSERT … ON CONFLICT DO UPDATE SET spent_usd = spent_usd + …` of every counter it is given:
- * an addition under the row's own lock, so two instances' concurrent saves lose nothing. No other
+ * one statement: the entries' ids inserted into `agboss_cost_entries` (a repeated one adds nothing,
+ * agbossjs/agboss#727), then `INSERT … ON CONFLICT DO UPDATE SET spent_usd = spent_usd + …` of every
+ * counter it is given: an addition under the row's own lock, so two instances' concurrent saves
+ * lose nothing. No other
  * locking: as on SQLite, the core accepts a small overrun of executions racing past one check
  * (nothing is reserved). A read is one `SELECT` of the counter's row (`usd(0)` when none). The
  * period is the core's (the `Clock`'s UTC day, week or month), passed as text. Every statement runs
@@ -69,12 +94,15 @@ export class PostgresCostLedgerRepository extends CostLedgerRepository {
 
   async saveAll(counters: readonly CostCounter[], records: readonly UsageRecord[]): Promise<void> {
     if (counters.length === 0 || records.length === 0) return;
+    const { ids, costs, unnamed } = entriesOf(records);
     await changesOf(this.connection.pool, ADD, [
       counters.map((counter) => counter.budget),
       counters.map(scopeOf),
       counters.map((counter) => counter.key),
       counters.map((counter) => counter.period),
-      costOf(records),
+      unnamed,
+      ids,
+      costs,
     ]);
   }
 

diff --git a/tests/cli/bin.test.ts b/tests/cli/bin.test.ts
@@ -26,7 +26,7 @@ describe("the agboss-postgres bin (a Kubernetes Job)", () => {
     );
 
     expect(first.stdout).toContain("applied 0013_nul_escape");
-    expect(second.stdout).toMatch(/nothing to apply: t_\w+ is at migration 21/);
+    expect(second.stdout).toMatch(/nothing to apply: t_\w+ is at migration 22/);
     expect(usage).toMatchObject({
       code: 2,
       stderr: expect.stringContaining("usage: agboss-postgres"),

diff --git a/tests/cli/command.test.ts b/tests/cli/command.test.ts
@@ -23,7 +23,7 @@ describe("agboss-postgres migrate", () => {
     expect(await runCommand(["migrate", "--url", url], first)).toBe(ExitCode.Ok);
     expect(await runCommand(["migrate", "--url", url], second)).toBe(ExitCode.Ok);
 
-    expect(first.lines.slice(0, 21)).toEqual([
+    expect(first.lines.slice(0, 22)).toEqual([
       expect.stringMatching(/^applied 0001_execution_leases \(t_/),
       expect.stringMatching(/^applied 0002_execution_records/),
       expect.stringMatching(/^applied 0003_execution_trees/),
@@ -45,10 +45,11 @@ describe("agboss-postgres migrate", () => {
       expect.stringMatching(/^applied 0019_execution_ids_from_outbound_calls/),
       expect.stringMatching(/^applied 0020_pauses_by_tree/),
       expect.stringMatching(/^applied 0021_pause_answers_by_tree/),
+      expect.stringMatching(/^applied 0022_cost_ledger_entries/),
     ]);
-    expect(first.lines[21]).toMatch(/^applied \d+ LangGraph checkpoint migration\(s\)$/);
+    expect(first.lines[22]).toMatch(/^applied \d+ LangGraph checkpoint migration\(s\)$/);
     expect(second.lines).toEqual([
-      expect.stringMatching(/^nothing to apply: t_\w+ is at migration 21$/),
+      expect.stringMatching(/^nothing to apply: t_\w+ is at migration 22$/),
     ]);
     expect([...first.errors, ...second.errors]).toEqual([]);
   });
@@ -58,7 +59,7 @@ describe("agboss-postgres migrate", () => {
     const io = ioOf({ DATABASE_URL: url });
 
     expect(await runCommand(["migrate"], io)).toBe(ExitCode.Ok);
-    expect(io.lines).toHaveLength(22);
+    expect(io.lines).toHaveLength(23);
   });
 
   it("neither --url nor DATABASE_URL: a usage error, exit code 2", async () => {
@@ -102,13 +103,13 @@ describe("agboss-postgres migrate", () => {
     const url = await freshSchema();
     await migrateWith(
       { url },
-      migrationsIn(migrationsFolder({ "0022_later.sql": "CREATE TABLE later (id integer);" })),
+      migrationsIn(migrationsFolder({ "0023_later.sql": "CREATE TABLE later (id integer);" })),
     );
     const io = ioOf();
 
     expect(await runCommand(["migrate", "--url", url], io)).toBe(ExitCode.Failed);
     expect(io.errors[0]).toMatch(
-      /^agboss-postgres: the migration failed — the PostgreSQL schema is at migration 22/,
+      /^agboss-postgres: the migration failed — the PostgreSQL schema is at migration 23/,
     );
   });
 

diff --git a/tests/connection/postgres-connection.test.ts b/tests/connection/postgres-connection.test.ts
@@ -108,7 +108,7 @@ describe("PostgresConnection.onInit: the schema check at start", () => {
     );
 
     expect(error).toBeInstanceOf(PostgresSchemaMissingError);
-    expect(error).toMatchObject({ code: "postgres.schema-missing", applied: 0, expected: 21 });
+    expect(error).toMatchObject({ code: "postgres.schema-missing", applied: 0, expected: 22 });
     expect(String(error)).toContain('npx agboss-postgres migrate --url "$DATABASE_URL"');
     expect(String(error)).toContain("postgresMigrateOnStart");
   });
@@ -121,6 +121,7 @@ describe("PostgresConnection.onInit: the schema check at start", () => {
         migrationsFolder({
           "0020_pauses_by_tree.sql": null,
           "0021_pause_answers_by_tree.sql": null,
+          "0022_cost_ledger_entries.sql": null,
         }),
       ),
     );
@@ -129,7 +130,7 @@ describe("PostgresConnection.onInit: the schema check at start", () => {
       rejectionOf(connection.onInit()),
     );
 
-    expect(error).toMatchObject({ applied: 19, expected: 21 });
+    expect(error).toMatchObject({ applied: 19, expected: 22 });
   });
 
   it("postgresMigrateOnStart: a missing schema is migrated at start", async () => {
@@ -141,7 +142,7 @@ describe("PostgresConnection.onInit: the schema check at start", () => {
 
     await withConnection(new PostgresConnection(application), (connection) => connection.onInit());
 
-    expect(await historyOf(url)).toHaveLength(21);
+    expect(await historyOf(url)).toHaveLength(22);
   });
 
   it("postgresMigrateOnStart: two instances starting at once migrate it once", async () => {
@@ -153,14 +154,14 @@ describe("PostgresConnection.onInit: the schema check at start", () => {
     await Promise.all(instances.map((connection) => connection.onInit()));
     await Promise.all(instances.map((connection) => connection.onDestroy()));
 
-    expect(await historyOf(url)).toHaveLength(21);
+    expect(await historyOf(url)).toHaveLength(22);
   });
 
   it("a schema a newer package migrated fails with PostgresSchemaTooNewError, flag or not", async () => {
     const url = await freshSchema();
     await migrateWith(
       { url },
-      migrationsIn(migrationsFolder({ "0022_later.sql": "CREATE TABLE later (id integer);" })),
+      migrationsIn(migrationsFolder({ "0023_later.sql": "CREATE TABLE later (id integer);" })),
     );
 
     for (const migrateOnStart of [false, true]) {
@@ -169,7 +170,7 @@ describe("PostgresConnection.onInit: the schema check at start", () => {
         (connection) => rejectionOf(connection.onInit()),
       );
       expect(error).toBeInstanceOf(PostgresSchemaTooNewError);
-      expect(error).toMatchObject({ applied: 22, expected: 21 });
+      expect(error).toMatchObject({ applied: 23, expected: 22 });
       expect(String(error)).toContain("the migrations are not rolled back");
     }
   });

diff --git a/tests/fixtures/process.application.ts b/tests/fixtures/process.application.ts
@@ -46,6 +46,7 @@ import {
   SpendWorkflow,
 } from "./shared.workflow.js";
 import { PaidHoldWorkflow } from "./paid-hold.workflow.js";
+import { PayItemsWorkflow } from "./spend-in-flight.workflow.js";
 import { TakeoverWorkflow } from "./takeover.workflow.js";
 
 /**
@@ -111,6 +112,7 @@ export class ProcessObserver implements OnExecutionEnd, OnExecutionLeaseTakeover
     ChargeWorkflow,
     OrderChargeWorkflow,
     PaidHoldWorkflow,
+    PayItemsWorkflow,
   ],
   observers: [ProcessObserver],
   budgets: [
@@ -156,6 +158,7 @@ export {
   ItemWorkflow,
   LimitedWorkflow,
   PaidHoldWorkflow,
+  PayItemsWorkflow,
   SpendWorkflow,
   TakeoverWorkflow,
 };