Fixes a usage under-counting bug introduced by #22524 (progressive
assistant-message persistence), flagged by @cubic-dev-ai and confirmed
by @FelixMalfait.
## Root cause
#22524 gated the thread-totals update (tokens, credits,
`conversationSize`) on `assistantMessageExistedAtStart` — an existence
check on the deterministic `uuidv5(streamId)` message id captured at job
start. That was a sound idempotency signal *before* #22524, when the
message only ever existed if a prior run had completed and applied
totals.
Progressive checkpoints broke that assumption: a checkpoint creates the
message row ~2s into the stream, without applying totals. So if the
worker is SIGKILLed after a checkpoint but before `handleStreamFinish`,
and the job is re-delivered (BullMQ's stalled re-run — `aiStreamQueue`
has no `maxStalledCount: 0` yet, that's #22518 — or an admin
`retryJobs`), the re-run sees `assistantMessageExistedAtStart === true`
and returns before the totals update. The turn's usage is lost
permanently. cubic's P2 (the non-transactional `delete`+`insert` in
`upsertAssistantMessage`) is the same root cause: its partless window
only mattered because it tripped the same existence-based gate.
## Fix
Stop inferring "totals already applied" from message existence. Gate the
totals update on **still owning the stream** — a conditional `UPDATE ...
WHERE id = :threadId AND activeStreamId = :streamId`, and only
`notifyThreadUsageUpdated` when it affects a row. This is the same claim
pattern the stream already uses (#22481), and it's idempotent by
construction:
- The run that completes while holding the claim → `affected = 1` →
totals applied exactly once. This holds **even when a checkpoint already
created the message**, which is precisely the bug.
- A duplicate/zombie run after another run completed (and its `finally`
cleared `activeStreamId`) → `affected = 0` → skipped, no double-count.
- A superseded run whose thread has moved to a newer stream → `affected
= 0` → skipped (defense-in-depth, aligns with #22518's ownership
pre-check).
The `assistantMessageExistedAtStart` flag and its start-of-stream
`hasMessageById` query are removed entirely — the message write is
already idempotent via the deterministic id + `upsert`, so it needs no
gate.
This subsumes cubic's P2: the totals are no longer lost regardless of
the `delete`+`insert` window, so no transaction is required for
correctness (the residual window is a benign sub-millisecond transient
for an actively-streaming message; happy to add a workspace-datasource
transaction as separate hardening if you'd prefer).
## Validation
`stream-agent-chat.job.spec.ts` (9 green):
- New: totals update returns `affected: 0` → `notifyThreadUsageUpdated`
**not** called (prior completion not double-counted), message still
upserted.
- New: message already exists from a checkpoint but claim still held
(`affected: 1`) → totals **are** applied — the exact regression #22524
caused.
- Existing success/error/cancel/abort flows updated for the conditional
criteria and still green.
<!-- This is an auto-generated description by cubic. -->
<a
href="https://cubic.dev/pr/twentyhq/twenty/pull/22534?utm_source=github"
target="_blank" rel="noopener noreferrer"
data-no-image-dialog="true"><picture><source
media="(prefers-color-scheme: dark)"
srcset="https://www.cubic.dev/buttons/review-in-cubic-dark.svg"><source
media="(prefers-color-scheme: light)"
srcset="https://www.cubic.dev/buttons/review-in-cubic-light.svg"><img
alt="Review in cubic"
src="https://www.cubic.dev/buttons/review-in-cubic-dark.svg"></picture></a>
<!-- End of auto-generated description by cubic. -->
Félix Malfait
9be2b21e51
