feat(sessions): separate session history from tracing retention and gate immutable record writes - #6517
feat(sessions): separate session history from tracing retention and gate immutable record writes#6517mmabrouk wants to merge 5 commits into
Conversation
Reject stable-ID content conflicts per record, expose the typed conflict contract, and preserve the legacy runner path while adding producer IDs. Claude-Session: https://claude.ai/code/session_01GAqSs7fw6QRi2n1ZJ2tmAV
|
The latest updates on your projects. Learn more about Vercel for GitHub.
|
📝 SummarySummary by CodeRabbit
WalkthroughThe change adds immutable session-record writes with stable producer IDs, typed content conflicts, updated worker acknowledgment behavior, permanent incomplete-history markers, revised runner checkpointing, and global session-level retention. ChangesSession history durability
Estimated code review effort: 5 (Critical) | ~120 minutes Merge Risk: 🟡 Moderate · up to Retry and conflict edge cases can duplicate session history or leave committed records without required follow-up processing. These durability issues should be fixed before merge. Sequence Diagram(s)sequenceDiagram
participant Runner
participant IngestAPI
participant RecordsWorker
participant RecordsDAO
participant SessionStreamsDAO
Runner->>IngestAPI: send record with producer_id
IngestAPI->>RecordsDAO: append record
RecordsDAO-->>IngestAPI: committed record or content conflict
RecordsWorker->>RecordsDAO: append queued records
RecordsDAO-->>RecordsWorker: append result with conflicts
RecordsWorker->>SessionStreamsDAO: mark incomplete history when required
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches 💡 1📝 Generate docstrings 💡
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
…on id PR #6503 already uses oss000000022 on the same parent. This file becomes oss000000025 so both branches can merge; whichever lands second re-points down_revision to the new head. Claude-Session: https://claude.ai/code/session_01GAqSs7fw6QRi2n1ZJ2tmAV
PR #6517 already uses oss000000025 on this chain. This file becomes oss000000026 on top of 024. Claude-Session: https://claude.ai/code/session_01GAqSs7fw6QRi2n1ZJ2tmAV
Re-point oss000000025 onto oss000000026 so the core chain is 024 -> 026 -> 025. Claude-Session: https://claude.ai/code/session_01GAqSs7fw6QRi2n1ZJ2tmAV
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 2
🧹 Nitpick comments (1)
api/oss/src/core/sessions/records/service.py (1)
92-103: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winReturn a Pydantic DTO from
RecordsService.mark_history_incomplete.The API convention requires service methods to return Pydantic
BaseModelDTOs. This method forwards theintreturned bySessionStreamsDAOInterface, and no current caller uses the result. If the method remains, return a DTO containing the count; otherwise remove the unused method.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository YAML (base), Organization UI (inherited)
Review profile: CHILL
Plan: Team
Run ID: 3a934d4d-6884-47e1-827d-d53d72575f55
📒 Files selected for processing (28)
api/ee/src/core/sessions/records/service.pyapi/ee/src/dbs/postgres/sessions/records/dao.pyapi/entrypoints/worker_streams.pyapi/oss/databases/postgres/migrations/core_oss/versions/oss000000025_add_session_history_incomplete.pyapi/oss/src/apis/fastapi/sessions/models.pyapi/oss/src/apis/fastapi/sessions/router.pyapi/oss/src/core/sessions/records/dtos.pyapi/oss/src/core/sessions/records/interfaces.pyapi/oss/src/core/sessions/records/service.pyapi/oss/src/core/sessions/records/types.pyapi/oss/src/core/sessions/streams/dtos.pyapi/oss/src/core/sessions/streams/interfaces.pyapi/oss/src/dbs/postgres/sessions/records/dao.pyapi/oss/src/dbs/postgres/sessions/streams/dao.pyapi/oss/src/dbs/postgres/sessions/streams/dbes.pyapi/oss/src/dbs/postgres/sessions/streams/mappings.pyapi/oss/src/tasks/asyncio/sessions/records_worker.pyapi/oss/src/utils/env.pyapi/oss/tests/pytest/unit/migrations/test_single_heads.pyapi/oss/tests/pytest/unit/sessions/test_orphaned_gate_reconciliation.pyapi/oss/tests/pytest/unit/sessions/test_record_ingest_endpoint.pyapi/oss/tests/pytest/unit/sessions/test_records_history_durability.pyapi/oss/tests/pytest/unit/sessions/test_records_history_writes.pyapi/oss/tests/pytest/unit/sessions/test_records_worker_batching.pyapi/oss/tests/pytest/unit/sessions/test_watch_publish.pyservices/runner/src/sessions/persist.tsservices/runner/src/sessions/record-id.tsservices/runner/tests/unit/session-persist.test.ts
Included review availability: Your plan provides up to 8 included reviews per hour; 6 remain after this review.
| if len(entries) > 1: | ||
| for entry in entries: | ||
| ( | ||
| appended, | ||
| committed_ids, | ||
| committed_events, | ||
| ) = await self._append_one( | ||
| project_id=project_batch["project_id"], | ||
| entry=entry, | ||
| ) | ||
| total_appended += appended | ||
| acknowledged_ids.extend(committed_ids) | ||
| await self._after_commit( | ||
| project_id=project_batch["project_id"], | ||
| events=committed_events, | ||
| ) |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy lift
🔎 Supported by static analysis
🏁 Script executed:
#!/bin/bash
set -euo pipefail
ast-grep outline api/oss/src/dbs/postgres/sessions/records/dao.py --items all --type function --match 'append|append_many'
rg -n -C 12 \
'async def append_many|async def append|AGENTA_SESSIONS_HISTORY_WRITES|commit\(|rollback\(|record_id|producer_id|RecordContentConflict' \
api/oss/src/dbs/postgres/sessions/records/dao.py \
api/oss/src/core/sessions/records/interfaces.py \
api/oss/src/core/sessions/records/dtos.pyRepository: Agenta-AI/agenta
Length of output: 31964
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '%s\n' '--- worker outline and relevant source ---'
ast-grep outline api/oss/src/tasks/asyncio/sessions/records_worker.py --items all --type function
rg -n -C 18 \
'append_many|_append_one|except|AGENTA_SESSIONS_HISTORY_WRITES|producer_id|record_id|RecordsService' \
api/oss/src/tasks/asyncio/sessions/records_worker.py
printf '%s\n' '--- service bindings ---'
fd -i -t f 'service' api/oss/src/core/sessions/records api/oss/src/tasks/asyncio/sessions
rg -n -C 12 \
'async def append_many|async def append|append_many\(|records_dao|RecordsDAOInterface' \
api/oss/src/core/sessions/records api/oss/src/tasks/asyncio/sessions
printf '%s\n' '--- database entity/schema ---'
fd -i -t f 'dbe|model|record' api/oss/src/dbs/postgres/sessions/records
rg -n -C 16 \
'class RecordDBE|record_id|UniqueConstraint|primary_key|server_default' \
api/oss/src/dbs/postgres/sessions/recordsRepository: Agenta-AI/agenta
Length of output: 50372
🏁 Script executed:
#!/bin/bash
set -euo pipefail
rg -n -C 18 \
'append_many|_append_one|except|AGENTA_SESSIONS_HISTORY_WRITES|producer_id|record_id|RecordsService' \
api/oss/src/tasks/asyncio/sessions/records_worker.py
rg -n -C 12 \
'async def append_many|async def append|append_many\(|records_dao|RecordsDAOInterface' \
api/oss/src/core/sessions/records api/oss/src/tasks/asyncio/sessions
rg -n -C 16 \
'class RecordDBE|record_id|UniqueConstraint|primary_key|server_default' \
api/oss/src/dbs/postgres/sessions/recordsRepository: Agenta-AI/agenta
Length of output: 50372
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '%s\n' '--- history_writes default ---'
rg -n -C 8 \
'history_writes|AGENTA_SESSIONS_HISTORY_WRITES' \
api/oss/src
printf '%s\n' '--- record-event construction and serialization ---'
rg -n -C 8 \
'SessionRecordEvent\(|record_event|producer_id|record_id' \
services api/oss/src 2>/dev/null | head -n 500Repository: Agenta-AI/agenta
Length of output: 49637
🏁 Script executed:
#!/bin/bash
set -euo pipefail
rg -n -C 8 \
'history_writes|AGENTA_SESSIONS_HISTORY_WRITES' \
api/oss/src
rg -n -C 8 \
'SessionRecordEvent\(|record_event|producer_id|record_id' \
services api/oss/src 2>/dev/null | head -n 500Repository: Agenta-AI/agenta
Length of output: 47741
Make replay idempotent in legacy mode. When AGENTA_SESSIONS_HISTORY_WRITES=false and record_id is absent, RecordsDAO ignores producer_id and generates a new uuid4 for each append_many call. If the initial commit succeeds but raises before process_batch receives success, this loop can retry each entry with a different key and insert duplicate rows. Use a retry-stable key in legacy mode or avoid replay after an uncertain commit.
| conflicts = set(result.conflicting_record_ids) | ||
| committed_events = [ | ||
| msg | ||
| for _, msg in entries | ||
| if (msg.record_event.record_id or msg.record_event.producer_id) | ||
| not in conflicts | ||
| ] |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
Derive committed events from result.records
When immutable history writes are enabled, RecordsDAO._dedupe_immutable_values keeps the first event for a stable ID and reports the later differing event in conflicting_record_ids. The worker then excludes every event with that ID, including the event that append_many committed. This leaves committed_events empty, so _after_commit skips gate reconciliation and records_changed publication. Use the returned records to select the accepted event once per ID, or return per-entry outcomes. Add a regression test for two same-ID events with different content.
PR #6517 already uses oss000000025 on this chain. This file becomes oss000000026 on top of 024. Claude-Session: https://claude.ai/code/session_01GAqSs7fw6QRi2n1ZJ2tmAV
|
🤖 The AI agent says: Suggestion (non-blocking, if this closed PR is revived): remove the unreachable HTTP conflict handler. This proposal combines history retention with gated immutable record writes. The PR body already notes that the HTTP 409 cannot be reached: I would delete that handler and describe ingestion as asynchronous acceptance, retaining immutable-write enforcement and conflict reporting in the worker. That removes misleading API behavior without adding a synchronous database write just to make the catch reachable. If callers actually require an immediate conflict verdict, that is a separate API contract decision. Static simplification review of the handler at the reviewed head. No implementation or live tests were run for this suggestion. |
|
Agent-generated, low weight. This PR is out of scope for the release and is being closed, so the unreachable branch goes with it. |
|
🤖 The AI agent says: [P2] Correctness, if this closed proposal is revived: mark retained sessions incomplete before deleting their history. When Reuse Source: retention deletion loop. Confirmed by static tracing of the retention service/DAO and marker writers at |
1 participant
Session records are conversation history, but today they sit under the tracing quota and retention: the records worker drops an over-quota org's records and marks nothing, so history can vanish silently. This PR is increment 3 of the session-control design (docs/design/session-control-and-live-events on PR #6495): the durable-history producer work that needs no client change.
What changes
env.py. When a record is still lost,session_streams.history_incompleteis set (additive nullable column, migrationoss000000025; it was 22, renumbered because PR feat(sessions): deliver durable Stop directly to the runner #6503 uses 22 on the same parent; whichever PR lands second re-pointsdown_revisionto the new head).AGENTA_SESSIONS_HISTORY_WRITES(default off): stable record ids that a retry reuses, one open slot per tool id so a completed tool call is durable before the turn ends, and per-record rejection of a retry that carries different content under the same id. The rejection is a typed core exception with coderecord_conflict, 409 at the router. Other records in the batch still commit.producer_idas an additive field that an older API ignores; with the flag off the API path is unchanged.Not in this PR: the per-session sequence (open question O1 in the design), snapshot, replay.
Tests
pytest oss/tests/pytest/unit/sessions -qagainst Postgres: 539 passed.pnpm testinservices/runner: 2,688 passed;session-persist.test.ts: 28 passed.processed_idsis filled at deserialization; pre-existing, tracked in the durable-history package.Known items for review
history_incomplete; the design contract names the snapshot fieldhistory_complete. One of the two should flip before the snapshot ships.Agent-generated, low weight. Not merged.
https://claude.ai/code/session_01GAqSs7fw6QRi2n1ZJ2tmAV