[bug] Session records are lost permanently and silently on ingest, worker, and crash paths

[mmabrouk]

Summary

The session-records pipeline loses records permanently in three separate ways, and reports success while doing it. This is independent of #5486, but it matters most there: that PR makes the record log the source of truth the runner rebuilds a conversation from, so a lost record becomes an agent that silently forgets part of the conversation and still answers confidently.

Found during a design review of a proposed Redis buffer in front of the record read. The buffer does not fix any of these.

1. The API reports success for records it failed to make durable

publish_record returns False when the Redis publish fails:

# api/oss/src/core/sessions/records/streaming.py:155-157
except Exception as e:
    log.error(f"[RECORDS] Failed to publish: {e}", exc_info=True)
    return False

The ingest route discards that return value and answers {"ok": True}:

# api/oss/src/apis/fastapi/sessions/router.py:524
await publish_record(...)
return {"ok": True}

The runner only retries on a failed response, so it never retries. This defeats the AGENTA_RECORDS_DURABLE retry path that #5486 added, and it also means the runner's dropped-record counter under-reports, because the drop happens on the API side where the runner cannot see it.

Expected: a durable publish failure returns a non-2xx so the producer retries.

2. The worker acknowledges and deletes records it never wrote

In api/oss/src/tasks/asyncio/sessions/records_worker.py, message ids are collected during deserialization, before any database write:

# :87 and :95
group["events"].append(msg)
processed_ids.append(msg_id)

The database write swallows its failure:

# :147-155
try:
    results = await self.service.append_many(...)
    total_appended += len(results)
except Exception:
    log.error("[RECORDS] Failed to append event batch", ...)

and every id is returned regardless:

# :160
return total_appended, processed_ids

The shared consumer then acknowledges and deletes them from the stream (api/oss/src/tasks/asyncio/shared/consumer.py:145-157). A database failure therefore destroys those records outright: no retry, no dead letter, no pending entry left to reclaim. The only trace is one log line.

The EE quota path at :145 continues past a whole project's batch while its ids stay in processed_ids, so a quota rejection silently discards records the same way.

Expected: acknowledge only ids whose records were committed. Leave the rest pending, or route them to a dead letter.

3. A crashed worker strands records forever

The consumer reads only new entries:

# api/oss/src/tasks/asyncio/shared/consumer.py:100
streams={self.stream_name: ">"}

There is no XAUTOCLAIM or pending-entry reclaim anywhere. Records claimed by a worker that dies before acknowledging are never redelivered to another worker.

Expected: a reclaim path for entries left pending beyond some idle time.

4. Related: records have no stable identity, so redelivery duplicates them

record_id is optional at ingest (api/oss/src/core/sessions/records/dtos.py:13) and the runner supplies one only for tool-family records (services/runner/src/sessions/persist.ts:265, :375). Ordinary message records, which are the conversation itself, send none. The id is minted in the worker's mapping:

# api/oss/src/dbs/postgres/sessions/records/mappings.py:20
record_id=event.record_id or uuid.uuid4()

So any redelivery of a message record mints a fresh id and inserts a second row for the same logical record. The upsert on (project_id, record_id) cannot deduplicate exactly the records that matter.

Expected: derive the id at the boundary from (session_id, turn_id, record_index), so both an HTTP retry and a stream redelivery land on the same row.

Why these are one issue

They compound. Fixing 1 makes the runner retry, which makes 4 matter, because retries then duplicate rows. Fixing 2 leaves entries pending, which makes 3 matter, because nothing reclaims them. Any one alone moves the failure somewhere else.

Impact

Silent, permanent loss of conversation history under any Redis or Postgres hiccup, and duplicated conversation turns under retry. Both are invisible in the product: the turn still answers.

Full analysis, including why a Redis buffer does not address any of this and the one product decision it forces, is in docs/design/sessions-record-cache/design.md on fix/sessions-reconstruct-qa.

[linear-code]

AGE-3986