Skip to content

Commit 390d85e

Browse files
authored
Merge branch 'feat/sessions-storage-rework' into feat/agent-cancel-steer
2 parents d748049 + b18089a commit 390d85e

252 files changed

Lines changed: 2838 additions & 659 deletions

File tree

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

.github/workflows/16-website-production.yml

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -51,6 +51,8 @@ jobs:
5151
# PostHog project key: public by design (ingest-only); analytics also
5252
# requires the runtime agenta.ai host guard, so previews stay silent.
5353
PUBLIC_POSTHOG_KEY: phc_cFT4a4081mXuKxYIrhIt6cmDwSjAO8zQlw9CFE05oSI
54+
# GA4 measurement ID: public by design, same property as the pre-pivot site.
55+
PUBLIC_GA_ID: G-368ZWZSH5D
5456
run: cd website && pnpm run build
5557

5658
- name: Deploy production

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

Lines changed: 61 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,7 @@
1212

1313
from oss.src.core.sessions.records.dtos import SessionRecordEvent
1414
from oss.src.dbs.redis.shared.engine import get_streams_engine
15+
from oss.src.utils.env import env
1516
from oss.src.utils.logging import get_module_logger
1617

1718
log = get_module_logger(__name__)
@@ -21,13 +22,61 @@
2122
# Truncate attributes at ingest to avoid storing unbounded record bodies.
2223
MAX_ATTRIBUTES_BYTES = 64 * 1024 # 64 KB per record
2324

25+
_TRUNCATION_MARKER = "…[truncated]"
26+
2427

2528
def _orjson_default(obj):
2629
if AsyncpgUUID is not None and isinstance(obj, AsyncpgUUID):
2730
return str(obj)
2831
raise TypeError(f"Type is not JSON serializable: {type(obj)}")
2932

3033

34+
def _truncate_attributes(attributes, budget: int, original_bytes: int):
35+
"""Shrink an oversized record body to fit `budget` while PRESERVING structure: small fields
36+
(``type``/``id``/``name``…) stay intact and only the largest string values are trimmed, each
37+
marked, so server-side history reconstruction still gets the event shape + partial content.
38+
Falls back to a minimal discriminator-only shape when non-string bloat can't be trimmed."""
39+
if original_bytes <= budget:
40+
return attributes
41+
if not isinstance(attributes, dict):
42+
return {"_truncated": True, "_original_bytes": original_bytes}
43+
44+
result = dict(attributes)
45+
trimmed: list[str] = []
46+
# Trim the largest string field repeatedly until the serialized body fits (or none remain).
47+
for _ in range(len(attributes)):
48+
size = len(dumps(result, default=_orjson_default))
49+
if size <= budget:
50+
break
51+
str_fields = [
52+
(k, v)
53+
for k, v in result.items()
54+
if isinstance(v, str) and not k.startswith("_")
55+
]
56+
if not str_fields:
57+
break
58+
key, value = max(str_fields, key=lambda kv: len(kv[1]))
59+
overflow = size - budget
60+
keep = max(
61+
0, len(value) - overflow - 128
62+
) # margin for the marker + json overhead
63+
result[key] = value[:keep] + _TRUNCATION_MARKER
64+
if key not in trimmed:
65+
trimmed.append(key)
66+
67+
if len(dumps(result, default=_orjson_default)) > budget:
68+
# Non-string bloat remains → keep only the discriminator fields reconstruction needs.
69+
return {
70+
"type": attributes.get("type"),
71+
"id": attributes.get("id"),
72+
"_truncated": True,
73+
"_original_bytes": original_bytes,
74+
}
75+
76+
result["_truncated"] = {"fields": trimmed, "original_bytes": original_bytes}
77+
return result
78+
79+
3180
def _get_redis():
3281
engine = get_streams_engine()
3382
return engine.get_redis() if engine else None
@@ -72,8 +121,19 @@ async def publish_record(
72121
session_id=str(record_event.session_id),
73122
original_bytes=len(raw_attributes),
74123
)
124+
# Smart truncation keeps the event shape + partial content so records stay
125+
# reconstructable; legacy path drops the whole body. Flag-gated (additive).
126+
new_attributes = (
127+
_truncate_attributes(
128+
record_event.attributes,
129+
MAX_ATTRIBUTES_BYTES,
130+
len(raw_attributes),
131+
)
132+
if env.agenta.sessions.records.smart_truncation
133+
else {"_truncated": True}
134+
)
75135
truncated_event = record_event.model_copy(
76-
update={"attributes": {"_truncated": True}}
136+
update={"attributes": new_attributes}
77137
)
78138

79139
message = {

api/oss/src/dbs/postgres/sessions/records/dao.py

Lines changed: 11 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -109,7 +109,17 @@ async def get_records(
109109
RecordDBE.project_id == project_id,
110110
RecordDBE.session_id == session_id,
111111
)
112-
.order_by(RecordDBE.created_at.asc(), RecordDBE.record_index.asc())
112+
# Producer event time first: it is the only key that is monotonic across
113+
# turns. `record_index` restarts at 0 every turn, and the worker can batch
114+
# records from two turns into one write so they share `created_at` — the old
115+
# (created_at, record_index) order then sorted the NEXT turn's first record
116+
# ahead of the PREVIOUS turn's later ones, interleaving the conversation.
117+
# Rows written before `timestamp` existed sort last within their ingest batch.
118+
.order_by(
119+
RecordDBE.timestamp.asc().nullslast(),
120+
RecordDBE.created_at.asc(),
121+
RecordDBE.record_index.asc(),
122+
)
113123
)
114124

115125
dbes = (await session.execute(stmt)).scalars().all()

api/oss/src/utils/env.py

Lines changed: 27 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -486,6 +486,32 @@ def _validate_mode(self) -> "RedactionConfig":
486486
return self
487487

488488

489+
# ---------------------------------------------------------------------------
490+
# agenta.sessions
491+
# ---------------------------------------------------------------------------
492+
493+
494+
class SessionsRecordsConfig(BaseModel):
495+
"""Durable session-record ingest tuning (server-side history reconstruction)."""
496+
497+
# When a record body exceeds the cap, preserve its structure + partial content (trim only
498+
# the large field values) instead of replacing the whole body with {"_truncated": True}.
499+
# Off = legacy whole-body drop. On makes reconstruction from records higher-fidelity.
500+
smart_truncation: bool = (
501+
os.getenv("AGENTA_RECORDS_SMART_TRUNCATION") or "false"
502+
).lower() in _TRUTHY
503+
504+
model_config = ConfigDict(extra="ignore")
505+
506+
507+
class SessionsConfig(BaseModel):
508+
"""Agenta sessions sub-namespace."""
509+
510+
records: SessionsRecordsConfig = SessionsRecordsConfig()
511+
512+
model_config = ConfigDict(extra="ignore")
513+
514+
489515
# ---------------------------------------------------------------------------
490516
# agenta — top-level Agenta core config.
491517
# ---------------------------------------------------------------------------
@@ -513,6 +539,7 @@ class AgentaConfig(BaseModel):
513539
otlp: OTLPConfig = OTLPConfig()
514540
redaction: RedactionConfig = RedactionConfig()
515541
services: ServicesConfig = ServicesConfig()
542+
sessions: SessionsConfig = SessionsConfig()
516543
webhooks: WebhooksConfig = WebhooksConfig()
517544
workers: WorkersConfig = WorkersConfig()
518545

Lines changed: 84 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,84 @@
1+
"""Smart truncation of oversized record bodies (`_truncate_attributes`).
2+
3+
The legacy path replaces an over-cap body with `{"_truncated": True}`, losing the event's
4+
type/id and all content. Smart truncation preserves the event shape + partial content so the
5+
record log stays reconstructable server-side. These pin that contract.
6+
"""
7+
8+
from orjson import dumps
9+
10+
from oss.src.core.sessions.records.streaming import (
11+
MAX_ATTRIBUTES_BYTES,
12+
_TRUNCATION_MARKER,
13+
_truncate_attributes,
14+
)
15+
16+
17+
def _size(obj) -> int:
18+
return len(dumps(obj))
19+
20+
21+
def test_under_budget_returns_unchanged():
22+
attrs = {"type": "message", "text": "hi"}
23+
out = _truncate_attributes(attrs, MAX_ATTRIBUTES_BYTES, _size(attrs))
24+
assert out is attrs # untouched, no _truncated marker
25+
26+
27+
def test_large_string_field_is_trimmed_but_structure_preserved():
28+
big = "x" * (MAX_ATTRIBUTES_BYTES * 2)
29+
attrs = {"type": "tool_result", "id": "call-1", "output": big}
30+
out = _truncate_attributes(attrs, MAX_ATTRIBUTES_BYTES, _size(attrs))
31+
32+
# Discriminator fields survive (unlike the legacy whole-body drop).
33+
assert out["type"] == "tool_result"
34+
assert out["id"] == "call-1"
35+
# The big field is trimmed + marked, and the whole body now fits the cap.
36+
assert out["output"].endswith(_TRUNCATION_MARKER)
37+
assert len(out["output"]) < len(big)
38+
assert _size(out) <= MAX_ATTRIBUTES_BYTES
39+
# Metadata records what was trimmed.
40+
assert out["_truncated"]["fields"] == ["output"]
41+
assert out["_truncated"]["original_bytes"] == _size(attrs)
42+
43+
44+
def test_trims_the_largest_of_several_string_fields():
45+
attrs = {
46+
"type": "message",
47+
"small": "ok",
48+
"text": "y" * (MAX_ATTRIBUTES_BYTES * 2),
49+
}
50+
out = _truncate_attributes(attrs, MAX_ATTRIBUTES_BYTES, _size(attrs))
51+
assert out["small"] == "ok" # small field untouched
52+
assert out["text"].endswith(_TRUNCATION_MARKER)
53+
assert _size(out) <= MAX_ATTRIBUTES_BYTES
54+
55+
56+
def test_non_string_bloat_falls_back_to_discriminator_only():
57+
# A huge nested structure with no single big string leaf can't be string-trimmed.
58+
attrs = {
59+
"type": "tool_call",
60+
"id": "call-9",
61+
"input": {str(i): i for i in range(MAX_ATTRIBUTES_BYTES)},
62+
}
63+
out = _truncate_attributes(attrs, MAX_ATTRIBUTES_BYTES, _size(attrs))
64+
assert out["type"] == "tool_call"
65+
assert out["id"] == "call-9"
66+
assert out["_truncated"] is True
67+
assert _size(out) <= MAX_ATTRIBUTES_BYTES
68+
69+
70+
def test_non_dict_attributes_fall_back():
71+
huge = "z" * (MAX_ATTRIBUTES_BYTES * 2)
72+
out = _truncate_attributes(huge, MAX_ATTRIBUTES_BYTES, _size(huge))
73+
assert out == {"_truncated": True, "_original_bytes": _size(huge)}
74+
75+
76+
def test_smart_truncation_flag_is_reachable_from_the_env_object():
77+
"""The publish path reads `env.agenta.sessions.records.smart_truncation` inside a
78+
try/except that swallows anything and drops the record. When `SessionsConfig` was not
79+
attached to `AgentaConfig` this raised AttributeError, so every over-cap record was
80+
discarded instead of truncated, and the tests above still passed because they call
81+
`_truncate_attributes` directly and never touch the flag."""
82+
from oss.src.utils.env import env
83+
84+
assert isinstance(env.agenta.sessions.records.smart_truncation, bool)

0 commit comments

Comments
 (0)