fix(sessions): paused-turn transcript hygiene + pause-sentinel replay nudges

api/oss/src/core/sessions/records/streaming.py

@@ -12,6 +12,7 @@
 
 from oss.src.core.sessions.records.dtos import SessionRecordEvent
 from oss.src.dbs.redis.shared.engine import get_streams_engine
+from oss.src.utils.env import env
 from oss.src.utils.logging import get_module_logger
 
 log = get_module_logger(__name__)
@@ -21,13 +22,61 @@
 # Truncate attributes at ingest to avoid storing unbounded record bodies.
 MAX_ATTRIBUTES_BYTES = 64 * 1024  # 64 KB per record
 
+_TRUNCATION_MARKER = "…[truncated]"
+
 
 def _orjson_default(obj):
     if AsyncpgUUID is not None and isinstance(obj, AsyncpgUUID):
         return str(obj)
     raise TypeError(f"Type is not JSON serializable: {type(obj)}")
 
 
+def _truncate_attributes(attributes, budget: int, original_bytes: int):
+    """Shrink an oversized record body to fit `budget` while PRESERVING structure: small fields
+    (``type``/``id``/``name``…) stay intact and only the largest string values are trimmed, each
+    marked, so server-side history reconstruction still gets the event shape + partial content.
+    Falls back to a minimal discriminator-only shape when non-string bloat can't be trimmed."""
+    if original_bytes <= budget:
+        return attributes
+    if not isinstance(attributes, dict):
+        return {"_truncated": True, "_original_bytes": original_bytes}
+
+    result = dict(attributes)
+    trimmed: list[str] = []
+    # Trim the largest string field repeatedly until the serialized body fits (or none remain).
+    for _ in range(len(attributes)):
+        size = len(dumps(result, default=_orjson_default))
+        if size <= budget:
+            break
+        str_fields = [
+            (k, v)
+            for k, v in result.items()
+            if isinstance(v, str) and not k.startswith("_")
+        ]
+        if not str_fields:
+            break
+        key, value = max(str_fields, key=lambda kv: len(kv[1]))
+        overflow = size - budget
+        keep = max(
+            0, len(value) - overflow - 128
+        )  # margin for the marker + json overhead
+        result[key] = value[:keep] + _TRUNCATION_MARKER
+        if key not in trimmed:
+            trimmed.append(key)
+
+    if len(dumps(result, default=_orjson_default)) > budget:
+        # Non-string bloat remains → keep only the discriminator fields reconstruction needs.
+        return {
+            "type": attributes.get("type"),
+            "id": attributes.get("id"),
+            "_truncated": True,
+            "_original_bytes": original_bytes,
+        }
+
+    result["_truncated"] = {"fields": trimmed, "original_bytes": original_bytes}
+    return result
+
+
 def _get_redis():
     engine = get_streams_engine()
     return engine.get_redis() if engine else None
@@ -72,8 +121,19 @@ async def publish_record(
                     session_id=str(record_event.session_id),
                     original_bytes=len(raw_attributes),
                 )
+                # Smart truncation keeps the event shape + partial content so records stay
+                # reconstructable; legacy path drops the whole body. Flag-gated (additive).
+                new_attributes = (
+                    _truncate_attributes(
+                        record_event.attributes,
+                        MAX_ATTRIBUTES_BYTES,
+                        len(raw_attributes),
+                    )
+                    if env.agenta.sessions.records.smart_truncation
+                    else {"_truncated": True}
+                )
                 truncated_event = record_event.model_copy(
-                    update={"attributes": {"_truncated": True}}
+                    update={"attributes": new_attributes}
                 )
 
         message = {

api/oss/src/utils/env.py

@@ -486,6 +486,32 @@ def _validate_mode(self) -> "RedactionConfig":
         return self
 
 
+# ---------------------------------------------------------------------------
+# agenta.sessions
+# ---------------------------------------------------------------------------
+
+
+class SessionsRecordsConfig(BaseModel):
+    """Durable session-record ingest tuning (server-side history reconstruction)."""
+
+    # When a record body exceeds the cap, preserve its structure + partial content (trim only
+    # the large field values) instead of replacing the whole body with {"_truncated": True}.
+    # Off = legacy whole-body drop. On makes reconstruction from records higher-fidelity.
+    smart_truncation: bool = (
+        os.getenv("AGENTA_RECORDS_SMART_TRUNCATION") or "false"
+    ).lower() in _TRUTHY
+
+    model_config = ConfigDict(extra="ignore")
+
+
+class SessionsConfig(BaseModel):
+    """Agenta sessions sub-namespace."""
+
+    records: SessionsRecordsConfig = SessionsRecordsConfig()
+
+    model_config = ConfigDict(extra="ignore")
+
+
 # ---------------------------------------------------------------------------
 # agenta — top-level Agenta core config.
 # ---------------------------------------------------------------------------

api/oss/tests/pytest/unit/sessions/test_records_truncation.py

@@ -0,0 +1,73 @@
+"""Smart truncation of oversized record bodies (`_truncate_attributes`).
+
+The legacy path replaces an over-cap body with `{"_truncated": True}`, losing the event's
+type/id and all content. Smart truncation preserves the event shape + partial content so the
+record log stays reconstructable server-side. These pin that contract.
+"""
+
+from orjson import dumps
+
+from oss.src.core.sessions.records.streaming import (
+    MAX_ATTRIBUTES_BYTES,
+    _TRUNCATION_MARKER,
+    _truncate_attributes,
+)
+
+
+def _size(obj) -> int:
+    return len(dumps(obj))
+
+
+def test_under_budget_returns_unchanged():
+    attrs = {"type": "message", "text": "hi"}
+    out = _truncate_attributes(attrs, MAX_ATTRIBUTES_BYTES, _size(attrs))
+    assert out is attrs  # untouched, no _truncated marker
+
+
+def test_large_string_field_is_trimmed_but_structure_preserved():
+    big = "x" * (MAX_ATTRIBUTES_BYTES * 2)
+    attrs = {"type": "tool_result", "id": "call-1", "output": big}
+    out = _truncate_attributes(attrs, MAX_ATTRIBUTES_BYTES, _size(attrs))
+
+    # Discriminator fields survive (unlike the legacy whole-body drop).
+    assert out["type"] == "tool_result"
+    assert out["id"] == "call-1"
+    # The big field is trimmed + marked, and the whole body now fits the cap.
+    assert out["output"].endswith(_TRUNCATION_MARKER)
+    assert len(out["output"]) < len(big)
+    assert _size(out) <= MAX_ATTRIBUTES_BYTES
+    # Metadata records what was trimmed.
+    assert out["_truncated"]["fields"] == ["output"]
+    assert out["_truncated"]["original_bytes"] == _size(attrs)
+
+
+def test_trims_the_largest_of_several_string_fields():
+    attrs = {
+        "type": "message",
+        "small": "ok",
+        "text": "y" * (MAX_ATTRIBUTES_BYTES * 2),
+    }
+    out = _truncate_attributes(attrs, MAX_ATTRIBUTES_BYTES, _size(attrs))
+    assert out["small"] == "ok"  # small field untouched
+    assert out["text"].endswith(_TRUNCATION_MARKER)
+    assert _size(out) <= MAX_ATTRIBUTES_BYTES
+
+
+def test_non_string_bloat_falls_back_to_discriminator_only():
+    # A huge nested structure with no single big string leaf can't be string-trimmed.
+    attrs = {
+        "type": "tool_call",
+        "id": "call-9",
+        "input": {str(i): i for i in range(MAX_ATTRIBUTES_BYTES)},
+    }
+    out = _truncate_attributes(attrs, MAX_ATTRIBUTES_BYTES, _size(attrs))
+    assert out["type"] == "tool_call"
+    assert out["id"] == "call-9"
+    assert out["_truncated"] is True
+    assert _size(out) <= MAX_ATTRIBUTES_BYTES
+
+
+def test_non_dict_attributes_fall_back():
+    huge = "z" * (MAX_ATTRIBUTES_BYTES * 2)
+    out = _truncate_attributes(huge, MAX_ATTRIBUTES_BYTES, _size(huge))
+    assert out == {"_truncated": True, "_original_bytes": _size(huge)}

docs/design/sessions-last-message-only/design.md

@@ -0,0 +1,55 @@
+# Last-message-only sends + server-side history reconstruction
+
+**Branch:** `feat/sessions-last-message-only` (off `feat/sessions-storage-rework` / #5436)
+**Motivation:** AGE-3970 (trace-drawer OOM on long turns) + device-independent continuity.
+Shrink request/trace payloads by sending only the newest user message per turn and
+reconstructing prior conversation server-side from durable session **records**.
+
+## Decisions (locked)
+- **Reconstruction lives in the runner (TS)** — at the `buildTurnText`/`priorMessages` seam,
+  reusing the runner's records client + session plumbing; a TS↔TS port of the FE
+  `transcriptToMessages.ts` folding. (Shared infra with JP/Mahmoud — keep additive/flagged.)
+- **Harden records durability FIRST**, before the FE trusts reconstruction — so reconstructed
+  history is authoritative, not lossy.
+
+## Current architecture (verified FE / SDK / runner / api)
+Two continuity paths already exist in the runner:
+- **Warm** (pool hit / same harness authored last turn): harness native `resumeSession()`
+  supplies history; `sendLastMessageOnly` is ALREADY true (`runtime-contracts.ts:168`).
+- **Cold** (evicted / different harness / cross-device): runner rebuilds the full transcript
+  from inbound `request.messages` (`transcript.ts:buildTurnText`/`priorMessages`; selected at
+  `run-turn.ts:134`).
+
+The FE always sends full history (`agentRequest.ts:401`) because it can't predict warm/cold.
+Records are **write-only** from the run's perspective: the runner persists every event
+per-turn (`persist.ts` → `POST /sessions/records/ingest`) but never reads them back for model
+context. Records ARE sufficient to reconstruct full ordered history (roles, text, reasoning,
+tool_call/tool_result, approvals) — each carries the whole ACP `AgentEvent`.
+
+## Hard constraint: HITL cold-replay
+Resume is message-driven, not interaction-driven. The runner binds tool_result→tool_call by
+scanning `request.messages` (`responder.ts:413,470`) and scans full history for approval
+verdicts (`responder.ts:358`). Durable `session_interactions` rows exist but aren't read to
+rebuild history. ⇒ Dropping history on a cold resume breaks approvals/client-tools/elicitation
+until the runner reconstructs matched tool_call/tool_result pairs from records. **The BE
+reconstructor + durable records are the prerequisite for the FE change.**
+
+## Phases
+- **Phase 1 — records durability** (enabler): make `message` + `tool_result` (and `tool_call`,
+  `interaction_request`) record persistence reliable end-to-end — reduce the runner's
+  fire-and-forget drop (ack/retry-to-DLQ) and fix the 64 KB attribute cap
+  (`records/streaming.py`) for tool bodies (raise + spill). Order records by
+  `session_turns.turn_index` then `record_index`.
+- **Phase 2 — runner reconstructor**: new `services/runner/src/sessions/` module folds
+  records→`ChatMessage[]` (inverse of `buildPersistingEmitter`); generalize
+  `sendLastMessageOnly` to fire whenever `session_id` yields reconstructable history; feed
+  `buildTurnText`/`priorMessages` + otel `run.start`. Behind a flag; FE unchanged.
+- **Phase 3 — FE last-message-only**: `buildAgentRequest` sends only the trailing user message
+  when the session is server-known with records; falls back to full history otherwise. Trace/
+  request payloads shrink → AGE-3970 mitigated.
+
+## Open risks tracked
+- Runner is ephemeral: guaranteeing delivery may mean blocking the run on a persist ack
+  (latency) or a durable outbox — Phase 1 decides the mechanism.
+- Interaction-row verdict as a backstop source (if a settled answer must survive without the
+  last message) — evaluate in Phase 2.

hosting/docker-compose/ee/docker-compose.dev.yml

@@ -424,6 +424,12 @@ services:
             AGENTA_RUNNER_DAYTONA_SESSION_IDLE_TTL_MS: ${AGENTA_RUNNER_DAYTONA_SESSION_IDLE_TTL_MS:-}
             AGENTA_RUNNER_DAYTONA_SESSION_MAX_WARM: ${AGENTA_RUNNER_DAYTONA_SESSION_MAX_WARM:-}
             AGENTA_RUNNER_DAYTONA_AUTODELETE_MINUTES: ${AGENTA_RUNNER_DAYTONA_AUTODELETE_MINUTES:-}
+            # This service has no env_file (see above), so the session flags must be listed
+            # here or they can never be set. AGENTA_SESSIONS_RECONSTRUCT must be enabled
+            # together with the web NEXT_PUBLIC_SESSIONS_LAST_MESSAGE_ONLY, or a turn that
+            # carries only its last message loses the conversation.
+            AGENTA_SESSIONS_RECONSTRUCT: ${AGENTA_SESSIONS_RECONSTRUCT:-}
+            AGENTA_RECORDS_DURABLE: ${AGENTA_RECORDS_DURABLE:-}
         # === STORAGE ============================================== #
         volumes:
             - ../../../services/runner/src:/app/src

hosting/docker-compose/ee/docker-compose.gh.yml

@@ -295,6 +295,12 @@ services:
             AGENTA_RUNNER_DAYTONA_SESSION_IDLE_TTL_MS: ${AGENTA_RUNNER_DAYTONA_SESSION_IDLE_TTL_MS:-}
             AGENTA_RUNNER_DAYTONA_SESSION_MAX_WARM: ${AGENTA_RUNNER_DAYTONA_SESSION_MAX_WARM:-}
             AGENTA_RUNNER_DAYTONA_AUTODELETE_MINUTES: ${AGENTA_RUNNER_DAYTONA_AUTODELETE_MINUTES:-}
+            # This service has no env_file (see above), so the session flags must be listed
+            # here or they can never be set. AGENTA_SESSIONS_RECONSTRUCT must be enabled
+            # together with the web NEXT_PUBLIC_SESSIONS_LAST_MESSAGE_ONLY, or a turn that
+            # carries only its last message loses the conversation.
+            AGENTA_SESSIONS_RECONSTRUCT: ${AGENTA_SESSIONS_RECONSTRUCT:-}
+            AGENTA_RECORDS_DURABLE: ${AGENTA_RECORDS_DURABLE:-}
         # === SUBSCRIPTION MOUNTS (opt-in) ========================= #
         # Use your own harness subscription for LOCAL runs instead of a managed API
         # key. Mount the login READ-WRITE: the harness runs directly out of the mount and

hosting/docker-compose/ee/env.ee.dev.example

@@ -111,6 +111,18 @@ AGENTA_RUNNER_DEFAULT_SANDBOX_PROVIDER=local
 # 5 min per tool call.
 # AGENTA_RUNNER_TOOL_CALL_TIMEOUT_MS=300000
 
+# --- Session history: last-message-only turns (all four default off) ---
+# The client sends only the newest user message and the runner rebuilds the earlier turns
+# from the durable session record log, which keeps requests and traces small on a long
+# conversation. The first two must be flipped together: with the web flag on and the runner
+# flag off, a turn arrives with no history at all.
+# AGENTA_SESSIONS_RECONSTRUCT=true
+# NEXT_PUBLIC_SESSIONS_LAST_MESSAGE_ONLY=true
+# Records become the only copy of the conversation, so make writing them durable and keep
+# the structure of a record whose body exceeds the API size cap.
+# AGENTA_RECORDS_DURABLE=true
+# AGENTA_RECORDS_SMART_TRUNCATION=true
+
 # ================================================================== #
 # Agenta - API
 # ================================================================== #

hosting/docker-compose/ee/env.ee.gh.example

@@ -113,6 +113,18 @@ AGENTA_RUNNER_TOKEN=replace-me
 # 5 min per tool call.
 # AGENTA_RUNNER_TOOL_CALL_TIMEOUT_MS=300000
 
+# --- Session history: last-message-only turns (all four default off) ---
+# The client sends only the newest user message and the runner rebuilds the earlier turns
+# from the durable session record log, which keeps requests and traces small on a long
+# conversation. The first two must be flipped together: with the web flag on and the runner
+# flag off, a turn arrives with no history at all.
+# AGENTA_SESSIONS_RECONSTRUCT=true
+# NEXT_PUBLIC_SESSIONS_LAST_MESSAGE_ONLY=true
+# Records become the only copy of the conversation, so make writing them durable and keep
+# the structure of a record whose body exceeds the API size cap.
+# AGENTA_RECORDS_DURABLE=true
+# AGENTA_RECORDS_SMART_TRUNCATION=true
+
 # ================================================================== #
 # Agenta - API
 # ================================================================== #

hosting/docker-compose/oss/docker-compose.dev.yml

@@ -410,6 +410,12 @@ services:
             AGENTA_RUNNER_DAYTONA_SESSION_IDLE_TTL_MS: ${AGENTA_RUNNER_DAYTONA_SESSION_IDLE_TTL_MS:-}
             AGENTA_RUNNER_DAYTONA_SESSION_MAX_WARM: ${AGENTA_RUNNER_DAYTONA_SESSION_MAX_WARM:-}
             AGENTA_RUNNER_DAYTONA_AUTODELETE_MINUTES: ${AGENTA_RUNNER_DAYTONA_AUTODELETE_MINUTES:-}
+            # This service has no env_file (see above), so the session flags must be listed
+            # here or they can never be set. AGENTA_SESSIONS_RECONSTRUCT must be enabled
+            # together with the web NEXT_PUBLIC_SESSIONS_LAST_MESSAGE_ONLY, or a turn that
+            # carries only its last message loses the conversation.
+            AGENTA_SESSIONS_RECONSTRUCT: ${AGENTA_SESSIONS_RECONSTRUCT:-}
+            AGENTA_RECORDS_DURABLE: ${AGENTA_RECORDS_DURABLE:-}
         # === STORAGE ============================================== #
         volumes:
             - ../../../services/runner/src:/app/src

hosting/docker-compose/oss/docker-compose.gh.yml

@@ -312,6 +312,12 @@ services:
             AGENTA_RUNNER_DAYTONA_SESSION_IDLE_TTL_MS: ${AGENTA_RUNNER_DAYTONA_SESSION_IDLE_TTL_MS:-}
             AGENTA_RUNNER_DAYTONA_SESSION_MAX_WARM: ${AGENTA_RUNNER_DAYTONA_SESSION_MAX_WARM:-}
             AGENTA_RUNNER_DAYTONA_AUTODELETE_MINUTES: ${AGENTA_RUNNER_DAYTONA_AUTODELETE_MINUTES:-}
+            # This service has no env_file (see above), so the session flags must be listed
+            # here or they can never be set. AGENTA_SESSIONS_RECONSTRUCT must be enabled
+            # together with the web NEXT_PUBLIC_SESSIONS_LAST_MESSAGE_ONLY, or a turn that
+            # carries only its last message loses the conversation.
+            AGENTA_SESSIONS_RECONSTRUCT: ${AGENTA_SESSIONS_RECONSTRUCT:-}
+            AGENTA_RECORDS_DURABLE: ${AGENTA_RECORDS_DURABLE:-}
         # === SUBSCRIPTION MOUNTS (opt-in) ========================= #
         # Use your own harness subscription for LOCAL runs instead of a managed API
         # key. Mount the login READ-WRITE: the harness runs directly out of the mount and

hosting/docker-compose/oss/env.oss.dev.example

@@ -111,6 +111,18 @@ AGENTA_RUNNER_DEFAULT_SANDBOX_PROVIDER=local
 # 5 min per tool call.
 # AGENTA_RUNNER_TOOL_CALL_TIMEOUT_MS=300000
 
+# --- Session history: last-message-only turns (all four default off) ---
+# The client sends only the newest user message and the runner rebuilds the earlier turns
+# from the durable session record log, which keeps requests and traces small on a long
+# conversation. The first two must be flipped together: with the web flag on and the runner
+# flag off, a turn arrives with no history at all.
+# AGENTA_SESSIONS_RECONSTRUCT=true
+# NEXT_PUBLIC_SESSIONS_LAST_MESSAGE_ONLY=true
+# Records become the only copy of the conversation, so make writing them durable and keep
+# the structure of a record whose body exceeds the API size cap.
+# AGENTA_RECORDS_DURABLE=true
+# AGENTA_RECORDS_SMART_TRUNCATION=true
+
 # ================================================================== #
 # Agenta - API
 # ================================================================== #

hosting/docker-compose/oss/env.oss.gh.example

@@ -113,6 +113,18 @@ AGENTA_RUNNER_TOKEN=replace-me
 # 5 min per tool call.
 # AGENTA_RUNNER_TOOL_CALL_TIMEOUT_MS=300000
 
+# --- Session history: last-message-only turns (all four default off) ---
+# The client sends only the newest user message and the runner rebuilds the earlier turns
+# from the durable session record log, which keeps requests and traces small on a long
+# conversation. The first two must be flipped together: with the web flag on and the runner
+# flag off, a turn arrives with no history at all.
+# AGENTA_SESSIONS_RECONSTRUCT=true
+# NEXT_PUBLIC_SESSIONS_LAST_MESSAGE_ONLY=true
+# Records become the only copy of the conversation, so make writing them durable and keep
+# the structure of a record whose body exceeds the API size cap.
+# AGENTA_RECORDS_DURABLE=true
+# AGENTA_RECORDS_SMART_TRUNCATION=true
+
 # ================================================================== #
 # Agenta - API
 # ================================================================== #