[feat] Last-message-only turns: runner rebuilds history from records

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.

services/runner/src/engines/sandbox_agent/reconstruct-history.ts

@@ -0,0 +1,50 @@
+/**
+ * Seam that lets the runner rebuild prior conversation from the durable record log instead of
+ * trusting a full inbound history — the server side of "client sends only the last message".
+ *
+ * Flag-gated (`AGENTA_SESSIONS_RECONSTRUCT`) and a strict no-op until BOTH the flag is on AND the
+ * client actually sent a minimal history: when the client still sends the whole conversation
+ * (`messages.length > 1`), reconstruction is skipped and behaviour is unchanged. Best-effort — any
+ * miss (no session, no records, fetch failure) leaves the inbound history untouched.
+ */
+
+import type { AgentRunRequest } from "../../protocol.ts";
+import { fetchSessionRecords } from "../../sessions/records-query.ts";
+import { reconstructMessages } from "../../sessions/reconstruct.ts";
+
+function reconstructEnabled(): boolean {
+  return (
+    String(process.env.AGENTA_SESSIONS_RECONSTRUCT ?? "").toLowerCase() === "true"
+  );
+}
+
+/**
+ * Returns a request whose `messages` are `[...reconstructed prior turns, ...inbound]` when
+ * reconstruction applies, else `null` to keep the inbound history as-is.
+ *
+ * MUST be called before the current turn's user message is persisted, so the record log holds
+ * only prior turns (no duplication of the incoming prompt).
+ */
+export async function reconstructHistoryIfNeeded(
+  request: AgentRunRequest,
+  sessionId: string | undefined,
+  auth: () => string,
+  log?: (msg: string) => void,
+): Promise<AgentRunRequest | null> {
+  if (!reconstructEnabled() || !sessionId) return null;
+  const inbound = request.messages ?? [];
+  // The client already sent the conversation — nothing to rebuild.
+  if (inbound.length > 1) return null;
+
+  const records = await fetchSessionRecords(sessionId, auth);
+  if (!records || records.length === 0) return null;
+
+  const reconstructed = reconstructMessages(records);
+  if (reconstructed.length === 0) return null;
+
+  log?.(
+    `[reconstruct] session=${sessionId} records=${records.length} ` +
+      `priorMessages=${reconstructed.length} inbound=${inbound.length}`,
+  );
+  return { ...request, messages: [...reconstructed, ...inbound] };
+}

services/runner/src/engines/sandbox_agent/run-turn.ts

@@ -75,7 +75,8 @@ import {
   nextTurnIndex,
   sessionContinuityStore,
 } from "./session-continuity.ts";
-import { priorMessages } from "./transcript.ts";
+import { reconstructHistoryIfNeeded } from "./reconstruct-history.ts";
+import { buildTurnText, priorMessages } from "./transcript.ts";
 import { resolveRunUsage } from "./usage.ts";
 
 /**
@@ -144,9 +145,28 @@ export async function runTurn(
   });
 
   try {
+    // Server-side history reconstruction (flag-gated, no-op by default): when the client sent a
+    // minimal history, rebuild prior turns from the durable record log so a cold turn still has
+    // full context. Runs before the current user turn is persisted, so records hold only prior
+    // turns. Reassigns `request` so every downstream reader (turnText, priorMessages, responder,
+    // otel) sees the same reconstructed history.
+    const reconstructed = await reconstructHistoryIfNeeded(
+      request,
+      sessionId,
+      () => runCredential(request),
+      logger,
+    );
+    if (reconstructed) request = reconstructed;
+
     const promptText = resolvePromptText(request);
-    // Cold: replay the full transcript (plan.turnText). Continuation or loaded: send only new text.
-    const turnText = sendLastMessageOnly(opts) ? promptText : plan.turnText;
+    // Cold: replay the full transcript. Continuation or loaded: send only new text. When history
+    // was rebuilt from records, recompute the transcript from it — the prebuilt plan.turnText
+    // predates the reconstruction.
+    const turnText = sendLastMessageOnly(opts)
+      ? promptText
+      : reconstructed
+        ? buildTurnText(request, logger)
+        : plan.turnText;
 
     const run = (deps.createOtel ?? createSandboxAgentOtel)({
       harness: plan.harness,