diff --git a/changelog.d/activity-monitor-slowness.md b/changelog.d/activity-monitor-slowness.md new file mode 100644 index 000000000..c0769d290 --- /dev/null +++ b/changelog.d/activity-monitor-slowness.md @@ -0,0 +1,12 @@ +--- +category: Fixes +--- + +**Fix admin dashboard slowness from response payload bloat**: the session-detail +API no longer re-sends the full cumulative conversation history inside every +turn (response payload is now O(total messages) instead of O(turns^2); the +dedup the activity-monitor frontend previously did client-side moved +server-side via `request_delta_start`), and the session-list queries fetch +preview/model payloads only for the page's sessions instead of probing every +stored request payload in the table. Markdown/JSONL exports also stop +repeating prior turns' history inside each turn. diff --git a/src/luthien_proxy/history/models.py b/src/luthien_proxy/history/models.py index ddd95901f..6df3cba03 100644 --- a/src/luthien_proxy/history/models.py +++ b/src/luthien_proxy/history/models.py @@ -65,6 +65,17 @@ class ConversationTurn(BaseModel): original_response_messages: list[ConversationMessage] | None = None # Request params (everything except messages/system, which are already parsed) request_params: dict[str, Any] | None = None + # Index into request_messages where THIS turn's new messages begin. + # Agent clients (e.g. Claude Code) re-send the full conversation history on + # every request, so raw per-turn request payloads are cumulative. To keep + # the response payload O(total messages) instead of O(turns^2), the service + # strips the re-sent prefix from request_messages for unmodified turns + # (request_delta_start stays 0 and request_messages holds only the new + # messages). For policy-modified turns the full arrays are kept so + # original-vs-final diffs still line up, and request_delta_start marks + # where the new messages start. Clients should render + # request_messages[request_delta_start:]. + request_delta_start: int = 0 class SessionSummary(BaseModel): diff --git a/src/luthien_proxy/history/service.py b/src/luthien_proxy/history/service.py index 5134282f8..0651ae572 100644 --- a/src/luthien_proxy/history/service.py +++ b/src/luthien_proxy/history/service.py @@ -491,7 +491,13 @@ async def _fetch_session_list_pg( user_id: str | None = None, search: SessionSearchParams | None = None, ) -> SessionListResponse: - """PostgreSQL version using PG-specific features (FILTER, DISTINCT ON, array_agg).""" + """PostgreSQL version: metadata page query + page-scoped payload lookups. + + Same 5-query shape as the SQLite path (count, page aggregate, models, + previews, user_ids); the payload-bearing lookups are keyed on the page's + session_ids so payload detoasting stays proportional to the page instead + of the whole table. + """ # SECURITY INVARIANT: user_id and every search value are bound as query # parameters, never interpolated into the SQL string. The user_id slot is # fixed at $3; search params (built by _build_session_filter_sql) occupy @@ -566,95 +572,127 @@ async def _fetch_session_list_pg( gate_clause = _gate_clause(where_gates) having_clause = _having_clause(having) + # PERF: the page query aggregates *metadata only* (timestamps, counts). + # models_used and preview_message are fetched by separate post-queries + # keyed on the page's session_ids (mirroring the SQLite path). The + # previous single-query shape probed ce.payload for EVERY + # 'transaction.request_recorded' row in the table (final_model and the + # max_tokens gate), which detoasts every stored payload. Agent-session + # payloads are cumulative (each request re-sends the whole + # conversation), so that scan was O(total conversation bytes) on every + # list load — the "history takes many seconds" admin-dashboard + # slowness. rows = await conn.fetch( f""" - WITH session_stats AS ( - SELECT - ce.session_id, - MIN(ce.created_at) as first_ts, - MAX(ce.created_at) as last_ts, - COUNT(*) as total_events, - COUNT(DISTINCT ce.call_id) as turn_count, - {_intervention_count_expr(True)} as policy_interventions - FROM conversation_events ce - WHERE ce.session_id IS NOT NULL - {user_call_filter} - {gate_clause} - GROUP BY ce.session_id - {having_clause} - ), - session_models AS ( - SELECT DISTINCT - ce.session_id, - ce.payload->>'final_model' as model - FROM conversation_events ce - WHERE ce.session_id IS NOT NULL - AND ce.event_type = 'transaction.request_recorded' - AND ce.payload->>'final_model' IS NOT NULL - {user_call_filter} - ), - session_first_message AS ( - SELECT DISTINCT ON (ce.session_id) - ce.session_id, - ce.payload as request_payload - FROM conversation_events ce - WHERE ce.session_id IS NOT NULL - AND ce.event_type = 'transaction.request_recorded' - -- Skip probe requests: max_tokens=1 means internal probe (token counting, quota). - -- COALESCE to 2 so requests without max_tokens are not skipped. - AND COALESCE((ce.payload->'final_request'->>'max_tokens')::int, 2) > 1 - {user_call_filter} - ORDER BY ce.session_id, ce.created_at ASC - ) SELECT - s.session_id, - s.first_ts, - s.last_ts, - s.total_events, - s.turn_count, - s.policy_interventions, - COALESCE( - array_agg(DISTINCT m.model) FILTER (WHERE m.model IS NOT NULL), - ARRAY[]::text[] - ) as models, - f.request_payload - FROM session_stats s - LEFT JOIN session_models m ON s.session_id = m.session_id - LEFT JOIN session_first_message f ON s.session_id = f.session_id - GROUP BY s.session_id, s.first_ts, s.last_ts, - s.total_events, s.turn_count, s.policy_interventions, - f.request_payload - ORDER BY s.last_ts DESC + ce.session_id, + MIN(ce.created_at) as first_ts, + MAX(ce.created_at) as last_ts, + COUNT(*) as total_events, + COUNT(DISTINCT ce.call_id) as turn_count, + {_intervention_count_expr(True)} as policy_interventions + FROM conversation_events ce + WHERE ce.session_id IS NOT NULL + {user_call_filter} + {gate_clause} + GROUP BY ce.session_id + {having_clause} + ORDER BY last_ts DESC LIMIT $1 OFFSET $2 """, *query_args, ) - # Separate user_ids lookup keyed on the page's session_ids. Distinct - # users only — never collapse via MIN/MAX. When a user filter is in - # effect the same scoping is applied so the response doesn't leak the - # *existence* of other users sharing the session. + models_by_session: dict[str, list[str]] = {} + preview_by_session: dict[str, str | None] = {} user_ids_by_session: dict[str, list[str]] = {} if rows: session_ids_on_page = [str(row["session_id"]) for row in rows] - placeholders = ", ".join(f"${i + 1}" for i in range(len(session_ids_on_page))) + + # When a user_id filter is in effect, restrict the model/preview/ + # user-id lookups to that user's call_ids — without this, + # preview_message and models_used can leak content from other + # users' calls that happen to share the session_id. + if user_id is not None: + lookup_user_filter = "AND ce.call_id IN (SELECT call_id FROM conversation_calls WHERE user_id = $2)" + lookup_extra_args: list[Any] = [user_id] + else: + lookup_user_filter = "" + lookup_extra_args = [] + + # One query for all models on this page. Only final_model is + # extracted, but restricting to page sessions keeps the detoast + # cost proportional to the page. + model_rows = await conn.fetch( + f""" + SELECT DISTINCT + ce.session_id, + ce.payload->>'final_model' as model + FROM conversation_events ce + WHERE ce.session_id = ANY($1::text[]) + AND ce.event_type = 'transaction.request_recorded' + AND ce.payload->>'final_model' IS NOT NULL + {lookup_user_filter} + """, + session_ids_on_page, + *lookup_extra_args, + ) + for r in model_rows: + sid = str(r["session_id"]) + model = str(r["model"]) + session_models = models_by_session.setdefault(sid, []) + if model not in session_models: + session_models.append(model) + + # First qualifying (non-probe) request payload per page session. + # LATERAL LIMIT 1 walks each session's events in created_at order + # and stops at the first row passing the max_tokens gate, so only + # a handful of payloads per session are detoasted instead of all + # of them. COALESCE to 2 so requests without max_tokens are not + # skipped; max_tokens=1 means internal probe (token counting, + # quota). + preview_rows = await conn.fetch( + f""" + SELECT s.sid as session_id, fm.payload as request_payload + FROM unnest($1::text[]) AS s(sid) + JOIN LATERAL ( + SELECT ce.payload + FROM conversation_events ce + WHERE ce.session_id = s.sid + AND ce.event_type = 'transaction.request_recorded' + AND COALESCE((ce.payload->'final_request'->>'max_tokens')::int, 2) > 1 + {lookup_user_filter} + ORDER BY ce.created_at ASC + LIMIT 1 + ) fm ON true + """, + session_ids_on_page, + *lookup_extra_args, + ) + for r in preview_rows: + sid = str(r["session_id"]) + if sid not in preview_by_session: + preview_by_session[sid] = _extract_preview_message(cast(_PreviewPayload, r["request_payload"])) + + # Separate user_ids lookup keyed on the page's session_ids. Distinct + # users only — never collapse via MIN/MAX. When a user filter is in + # effect the same scoping is applied so the response doesn't leak the + # *existence* of other users sharing the session. if user_id is not None: - user_id_filter_clause = f"AND cc.user_id = ${len(session_ids_on_page) + 1}" - user_id_extra_args: list[Any] = [user_id] + user_id_filter_clause = "AND cc.user_id = $2" else: user_id_filter_clause = "" - user_id_extra_args = [] user_id_rows = await conn.fetch( f""" SELECT DISTINCT ce.session_id, cc.user_id FROM conversation_events ce JOIN conversation_calls cc ON ce.call_id = cc.call_id - WHERE ce.session_id IN ({placeholders}) + WHERE ce.session_id = ANY($1::text[]) AND cc.user_id IS NOT NULL {user_id_filter_clause} """, - *session_ids_on_page, - *user_id_extra_args, + session_ids_on_page, + *lookup_extra_args, ) for r in user_id_rows: sid = str(r["session_id"]) @@ -671,8 +709,8 @@ async def _fetch_session_list_pg( turn_count=int(row["turn_count"]), # type: ignore[arg-type] total_events=int(row["total_events"]), # type: ignore[arg-type] policy_interventions=int(row["policy_interventions"]), # type: ignore[arg-type] - models_used=list(row["models"]) if row["models"] else [], # type: ignore[arg-type] - preview_message=_extract_preview_message(cast(_PreviewPayload, row["request_payload"])), + models_used=models_by_session.get(str(row["session_id"]), []), + preview_message=preview_by_session.get(str(row["session_id"])), user_ids=user_ids_by_session.get(str(row["session_id"]), []), ) for row in rows @@ -692,11 +730,10 @@ async def _fetch_session_list_sqlite( user_id: str | None = None, search: SessionSearchParams | None = None, ) -> SessionListResponse: - """SQLite version: 3 queries total (vs PostgreSQL's 2). + """SQLite version. Avoids N+1 by batching models and previews for the whole page in one - query each, then merging in Python. PostgreSQL uses array_agg/DISTINCT ON - in a single CTE; SQLite lacks those, so we use IN (session_ids) instead. + query each (keyed on the page's session_ids), then merging in Python. """ # SECURITY INVARIANT: user_id and every search value are bound as query # parameters, never interpolated into the SQL string. user_id occupies $3 @@ -822,19 +859,31 @@ async def _fetch_session_list_sqlite( *extra_args, ) - # One query for first qualifying preview per session on this page + # One query for the first qualifying (non-probe) preview payload per + # session on this page. ROW_NUMBER keeps only the earliest qualifying + # row per session *inside* SQLite, so exactly one payload per session + # crosses into Python. Request payloads are cumulative (agent clients + # re-send the whole conversation each request), so the previous shape + # — shipping every request payload for the page's sessions to Python — + # transferred O(total conversation bytes) per list load. preview_rows = await conn.fetch( f""" - SELECT ce.session_id, ce.payload as request_payload - FROM conversation_events ce - WHERE ce.session_id IN ({placeholders}) - AND ce.event_type = 'transaction.request_recorded' - AND COALESCE( - CAST(json_extract(ce.payload, '$.final_request.max_tokens') AS INTEGER), - 2 - ) > 1 - {user_call_filter_lookups} - ORDER BY ce.session_id, ce.created_at ASC + SELECT session_id, request_payload FROM ( + SELECT + ce.session_id, + ce.payload as request_payload, + ROW_NUMBER() OVER ( + PARTITION BY ce.session_id ORDER BY ce.created_at ASC + ) as qualifying_rank + FROM conversation_events ce + WHERE ce.session_id IN ({placeholders}) + AND ce.event_type = 'transaction.request_recorded' + AND COALESCE( + CAST(json_extract(ce.payload, '$.final_request.max_tokens') AS INTEGER), + 2 + ) > 1 + {user_call_filter_lookups} + ) WHERE qualifying_rank = 1 """, *session_ids, *extra_args, @@ -973,6 +1022,8 @@ async def fetch_session_detail(session_id: str, db_pool: DatabasePool) -> Sessio if turn.had_policy_intervention: total_interventions += len(turn.annotations) + _dedup_cumulative_request_messages(turns) + first_ts_str = parse_db_ts(rows[0]["created_at"]).isoformat() last_ts_str = parse_db_ts(rows[-1]["created_at"]).isoformat() @@ -1103,6 +1154,81 @@ def _build_turn(call_id: str, events: list[StoredEvent]) -> ConversationTurn: ) +# Preflight (non-conversational) request classification, mirrored by the +# activity-monitor frontend (conversation_live.js classifyPreflight): +# - Quota/token-count probe: max_tokens == 1 +# - Title generation: json_schema output format with a small token budget +_PREFLIGHT_TITLE_MAX_TOKENS = 256 + + +def _is_preflight_turn(request_params: dict[str, Any] | None) -> bool: + """True for standalone probe/title-generation requests. + + Preflight requests are independent one-shot calls (quota probes, title + generation) interleaved into a session. They are not part of the + cumulative conversation thread, so they neither get deduplicated nor + advance the dedup baseline. + """ + if not request_params: + return False + max_tokens = request_params.get("max_tokens") + if max_tokens == 1: + return True + output_config = request_params.get("output_config") + format_type = None + if isinstance(output_config, dict): + fmt = output_config.get("format") + if isinstance(fmt, dict): + format_type = fmt.get("type") + return format_type == "json_schema" and isinstance(max_tokens, int) and max_tokens <= _PREFLIGHT_TITLE_MAX_TOKENS + + +def _dedup_cumulative_request_messages(turns: list[ConversationTurn]) -> None: + """Strip re-sent conversation history from each turn's request messages. + + Agent clients (Claude Code and friends) send the entire conversation so + far on every request, so turn N's parsed ``request_messages`` repeats all + of turn N-1's messages plus the new ones. Left as-is, the session-detail + response payload grows O(turns^2) with session length — the "loading a + transcript takes a minute" admin-dashboard slowness under realistic agent + volumes. This pass keeps only each turn's *new* messages, making the + payload O(total messages). + + Rules (mutating ``turns`` in place): + - Preflight turns (quota probes / title generation) are standalone + requests: kept whole, and they do not advance the dedup baseline. + - Unmodified turns: ``request_messages`` is replaced by the delta (the + messages beyond the previous turn's count); ``request_delta_start`` + stays 0. + - Policy-modified turns: full arrays are kept so original-vs-final + diffs still line up index-by-index; ``request_delta_start`` marks + where this turn's new messages begin. + - If a turn's message count *shrinks* (context compaction, or a policy + rewrote history — the cumulative invariant is broken), the turn is + kept whole and the baseline resets to its length. + + Invariant note: like the previous client-side implementation, this trusts + the cumulative-count invariant (prior messages are re-sent unchanged); it + does not diff message contents. + """ + prev_count = 0 + for turn in turns: + if _is_preflight_turn(turn.request_params): + continue + count = len(turn.request_messages) + if count < prev_count: + # Invariant broken: keep the full array, restart the baseline. + turn.request_delta_start = 0 + prev_count = count + continue + if turn.request_was_modified: + turn.request_delta_start = prev_count + else: + turn.request_messages = turn.request_messages[prev_count:] + turn.request_delta_start = 0 + prev_count = count + + def _extract_policy_name(event_type: str) -> str: """Extract policy name from event type like 'policy.judge.tool_call_blocked'.""" parts = event_type.split(".") @@ -1143,8 +1269,10 @@ def export_session_markdown(session: SessionDetail) -> str: lines.append(f"*Model: {turn.model}*") lines.append("") - # Request messages - for msg in turn.request_messages: + # Request messages new to this turn (request_messages is already the + # delta for unmodified turns; modified turns keep full arrays with + # request_delta_start marking where the new messages begin). + for msg in turn.request_messages[turn.request_delta_start :]: lines.append(_format_message_markdown(msg)) lines.append("") @@ -1179,7 +1307,10 @@ def export_session_jsonl(session: SessionDetail) -> str: "session_id": session.session_id, "timestamp": turn.timestamp, "model": turn.model, - "request_messages": [m.model_dump(mode="json") for m in turn.request_messages], + # Delta only: messages new to this turn. Agent clients re-send the + # whole conversation each request; exporting the cumulative arrays + # per turn made exports O(turns^2). + "request_messages": [m.model_dump(mode="json") for m in turn.request_messages[turn.request_delta_start :]], "response_messages": [m.model_dump(mode="json") for m in turn.response_messages], "annotations": [a.model_dump(mode="json") for a in turn.annotations], "had_policy_intervention": turn.had_policy_intervention, diff --git a/src/luthien_proxy/static/conversation_live.js b/src/luthien_proxy/static/conversation_live.js index 855e1e18e..a6f6c547b 100644 --- a/src/luthien_proxy/static/conversation_live.js +++ b/src/luthien_proxy/static/conversation_live.js @@ -266,42 +266,24 @@ function conversationViewer() { this.turns = this.presentTurns(rawTurns); }, - // Presentation pipeline: classify preflight turns and compute - // display messages (dedup) entirely on the client side. + // Presentation pipeline: classify preflight turns (for badges/styling) + // and compute display messages. // - // The API sends the full conversation history on every request: - // Turn 1: [user₀] - // Turn 2: [user₀, assistant₁, user₂] - // Turn 3: [user₀, assistant₁, user₂, tool_call₂, tool_result₂, user₃] - // - // user₀ (the initial message with all preamble) is re-sent identically - // every turn. New content appears at the end, after the previous turn's - // messages. So for turn N, display = request_messages.slice(prevCount). - // Preflight turns are excluded from the count so they don't disrupt the - // sequence. - // - // Invariant: the API sends a stable, strictly-growing cumulative - // message array. If a policy rewrites or reorders earlier messages, - // the slicing will produce incorrect results. + // Deduplication of the cumulative request history now happens on the + // SERVER (history/service.py _dedup_cumulative_request_messages), so + // the response payload is O(total messages) instead of O(turns²): + // - Unmodified turns arrive with request_messages already reduced to + // this turn's new messages (request_delta_start = 0). + // - Policy-modified turns keep their full arrays so the + // original-vs-final diff panels line up, and request_delta_start + // marks where this turn's new messages begin. + // Either way, display = request_messages.slice(request_delta_start). presentTurns(rawTurns) { - let prevRealMsgCount = 0; - return rawTurns.map(turn => { const isPreflight = this.classifyPreflight(turn); const messages = turn.request_messages || []; - - let displayMessages; - if (isPreflight) { - displayMessages = messages; - } else { - displayMessages = messages.slice(prevRealMsgCount); - if (displayMessages.length === 0 && messages.length > 0) { - console.warn('Dedup produced empty messages for turn', turn.call_id, - '— cumulative array invariant may be violated'); - } - prevRealMsgCount = messages.length; - } - + const deltaStart = turn.request_delta_start || 0; + const displayMessages = deltaStart > 0 ? messages.slice(deltaStart) : messages; return { ...turn, _isPreflight: isPreflight, _displayMessages: displayMessages }; }); }, diff --git a/tests/luthien_proxy/unit_tests/history/test_service.py b/tests/luthien_proxy/unit_tests/history/test_service.py index de7d489d5..5c2d4508d 100644 --- a/tests/luthien_proxy/unit_tests/history/test_service.py +++ b/tests/luthien_proxy/unit_tests/history/test_service.py @@ -986,15 +986,23 @@ async def test_successful_fetch(self): "total_events": 10, "turn_count": 3, "policy_interventions": 1, - "models": ["gpt-4", "claude-3"], + }, + ] + mock_model_rows = [ + {"session_id": "session-1", "model": "gpt-4"}, + {"session_id": "session-1", "model": "claude-3"}, + ] + mock_preview_rows = [ + { + "session_id": "session-1", "request_payload": {"final_request": {"messages": [{"role": "user", "content": "Hello world"}]}}, }, ] mock_conn = AsyncMock() mock_conn.fetchval.return_value = 1 # Total count - # First fetch() = main session aggregation; second = user_ids lookup. - mock_conn.fetch.side_effect = [mock_rows, []] + # Fetches: main aggregation, then models / previews / user_ids lookups. + mock_conn.fetch.side_effect = [mock_rows, mock_model_rows, mock_preview_rows, []] mock_pool = MagicMock() mock_pool.is_sqlite = False @@ -1024,14 +1032,14 @@ async def test_fetch_with_offset(self): "total_events": 5, "turn_count": 2, "policy_interventions": 0, - "models": ["gpt-4"], - "request_payload": None, # Test with no first message }, ] + mock_model_rows = [{"session_id": "session-2", "model": "gpt-4"}] mock_conn = AsyncMock() mock_conn.fetchval.return_value = 100 # Total count - mock_conn.fetch.side_effect = [mock_rows, []] + # No preview row for this session (no qualifying first message). + mock_conn.fetch.side_effect = [mock_rows, mock_model_rows, [], []] mock_pool = MagicMock() mock_pool.is_sqlite = False @@ -1385,3 +1393,275 @@ def test_includes_original_messages_when_modified(self): assert record["request_was_modified"] is False assert record["original_response_messages"][0]["content"] == "Hi" assert "original_request_messages" not in record + + +def _mock_detail_pool(rows: list[dict]) -> MagicMock: + """Build a mock DatabasePool whose single fetch returns the given event rows.""" + mock_conn = AsyncMock() + mock_conn.fetch.return_value = rows + mock_pool = MagicMock() + mock_pool.connection.return_value.__aenter__.return_value = mock_conn + return mock_pool + + +def _request_event(call_id: str, minute: int, messages: list[dict], **payload_extra) -> dict: + """Build a transaction.request_recorded event row for a cumulative session.""" + payload = { + "final_model": "claude-3-opus", + "final_request": {"messages": messages, "max_tokens": 4096}, + } + payload.update(payload_extra) + return { + "call_id": call_id, + "event_type": "transaction.request_recorded", + "payload": payload, + "created_at": datetime(2025, 1, 15, 10, minute, 0), + } + + +def _response_event(call_id: str, minute: int, text: str) -> dict: + return { + "call_id": call_id, + "event_type": "transaction.streaming_response_recorded", + "payload": {"final_response": {"role": "assistant", "content": [{"type": "text", "text": text}]}}, + "created_at": datetime(2025, 1, 15, 10, minute, 30), + } + + +class TestCumulativeRequestDedup: + """Session detail strips the re-sent conversation prefix from each turn. + + Agent clients (Claude Code) re-send the full conversation history on + every request, so raw per-turn request payloads are cumulative and the + detail response used to grow O(turns^2). The service now returns only + each turn's new messages (request_delta_start marks the boundary for + turns that must keep full arrays). + """ + + @pytest.mark.asyncio + async def test_unmodified_turns_return_only_new_messages(self): + turn1_messages = [{"role": "user", "content": "Q1"}] + turn2_messages = [ + {"role": "user", "content": "Q1"}, + {"role": "assistant", "content": "A1"}, + {"role": "user", "content": "Q2"}, + ] + turn3_messages = turn2_messages + [ + {"role": "assistant", "content": "A2"}, + {"role": "user", "content": "Q3"}, + ] + rows = [ + _request_event("call-1", 0, turn1_messages), + _response_event("call-1", 0, "A1"), + _request_event("call-2", 1, turn2_messages), + _response_event("call-2", 1, "A2"), + _request_event("call-3", 2, turn3_messages), + _response_event("call-3", 2, "A3"), + ] + + result = await fetch_session_detail("session-1", _mock_detail_pool(rows)) + + assert [m.content for m in result.turns[0].request_messages] == ["Q1"] + assert [m.content for m in result.turns[1].request_messages] == ["A1", "Q2"] + assert [m.content for m in result.turns[2].request_messages] == ["A2", "Q3"] + assert all(turn.request_delta_start == 0 for turn in result.turns) + # Response messages are per-turn already and must be untouched. + assert [m.content for m in result.turns[0].response_messages] == ["A1"] + assert [m.content for m in result.turns[2].response_messages] == ["A3"] + + @pytest.mark.asyncio + async def test_payload_size_grows_linearly_not_quadratically(self): + """Total messages across turns equals the conversation length, not its square.""" + n_turns = 10 + rows = [] + cumulative: list[dict] = [] + for i in range(n_turns): + cumulative = cumulative + [ + {"role": "user", "content": f"Q{i}"}, + ] + rows.append(_request_event(f"call-{i}", i, list(cumulative))) + rows.append(_response_event(f"call-{i}", i, f"A{i}")) + cumulative = cumulative + [{"role": "assistant", "content": f"A{i}"}] + + result = await fetch_session_detail("session-1", _mock_detail_pool(rows)) + + total_request_messages = sum(len(t.request_messages) for t in result.turns) + # Each turn contributes exactly its new messages (user + prior assistant), + # so the total is 2*n - 1, not sum(1..2n) ~ n^2. + assert total_request_messages == 2 * n_turns - 1 + + @pytest.mark.asyncio + async def test_modified_turn_keeps_full_arrays_with_delta_start(self): + turn1_messages = [{"role": "user", "content": "Q1"}] + turn2_original = [ + {"role": "user", "content": "Q1"}, + {"role": "assistant", "content": "A1"}, + {"role": "user", "content": "Q2"}, + ] + turn2_final = [ + {"role": "user", "content": "Q1"}, + {"role": "assistant", "content": "A1"}, + {"role": "user", "content": "Q2 [redacted]"}, + ] + turn3_messages = turn2_final + [ + {"role": "assistant", "content": "A2"}, + {"role": "user", "content": "Q3"}, + ] + rows = [ + _request_event("call-1", 0, turn1_messages), + _request_event("call-2", 1, turn2_final, original_request={"messages": turn2_original}), + _request_event("call-3", 2, turn3_messages), + ] + + result = await fetch_session_detail("session-1", _mock_detail_pool(rows)) + + modified_turn = result.turns[1] + assert modified_turn.request_was_modified is True + # Full arrays preserved so original-vs-final diff lines up index-by-index. + assert len(modified_turn.request_messages) == 3 + assert modified_turn.original_request_messages is not None + assert len(modified_turn.original_request_messages) == 3 + # Delta boundary points at this turn's new messages. + assert modified_turn.request_delta_start == 1 + assert [m.content for m in modified_turn.request_messages[modified_turn.request_delta_start :]] == [ + "A1", + "Q2 [redacted]", + ] + # Following turn dedups against the modified turn's full length. + assert [m.content for m in result.turns[2].request_messages] == ["A2", "Q3"] + + @pytest.mark.asyncio + async def test_preflight_turns_kept_whole_and_do_not_advance_baseline(self): + turn1_messages = [{"role": "user", "content": "Q1"}] + probe_messages = [{"role": "user", "content": "quota"}] + turn2_messages = [ + {"role": "user", "content": "Q1"}, + {"role": "assistant", "content": "A1"}, + {"role": "user", "content": "Q2"}, + ] + probe = _request_event("call-probe", 1, probe_messages) + probe["payload"]["final_request"]["max_tokens"] = 1 + rows = [ + _request_event("call-1", 0, turn1_messages), + probe, + _request_event("call-2", 2, turn2_messages), + ] + + result = await fetch_session_detail("session-1", _mock_detail_pool(rows)) + + # Probe turn untouched. + assert [m.content for m in result.turns[1].request_messages] == ["quota"] + assert result.turns[1].request_delta_start == 0 + # Turn after the probe still dedups against turn 1, not the probe. + assert [m.content for m in result.turns[2].request_messages] == ["A1", "Q2"] + + @pytest.mark.asyncio + async def test_title_generation_preflight_kept_whole(self): + turn1_messages = [{"role": "user", "content": "Q1"}] + title_messages = [{"role": "user", "content": "Summarize this session"}] + title = _request_event("call-title", 1, title_messages) + title["payload"]["final_request"]["max_tokens"] = 128 + title["payload"]["final_request"]["output_config"] = {"format": {"type": "json_schema", "schema": {}}} + rows = [ + _request_event("call-1", 0, turn1_messages), + title, + ] + + result = await fetch_session_detail("session-1", _mock_detail_pool(rows)) + + assert [m.content for m in result.turns[1].request_messages] == ["Summarize this session"] + assert result.turns[1].request_delta_start == 0 + + @pytest.mark.asyncio + async def test_shrinking_history_resets_baseline(self): + """Context compaction (shorter request than the previous turn) keeps the turn whole.""" + turn1_messages = [ + {"role": "user", "content": "Q1"}, + {"role": "assistant", "content": "A1"}, + {"role": "user", "content": "Q2"}, + ] + compacted_messages = [{"role": "user", "content": "Summary of Q1/Q2"}] + turn3_messages = compacted_messages + [ + {"role": "assistant", "content": "A3"}, + {"role": "user", "content": "Q4"}, + ] + rows = [ + _request_event("call-1", 0, turn1_messages), + _request_event("call-2", 1, compacted_messages), + _request_event("call-3", 2, turn3_messages), + ] + + result = await fetch_session_detail("session-1", _mock_detail_pool(rows)) + + # Compacted turn kept whole (previous behavior dropped it entirely). + assert [m.content for m in result.turns[1].request_messages] == ["Summary of Q1/Q2"] + assert result.turns[1].request_delta_start == 0 + # Baseline restarts from the compacted turn. + assert [m.content for m in result.turns[2].request_messages] == ["A3", "Q4"] + + def test_markdown_export_does_not_repeat_history(self): + turns = [ + ConversationTurn( + call_id="call-1", + timestamp="2026-03-31T10:00:00", + model="claude-3-opus", + request_messages=[ConversationMessage(message_type=MessageType.USER, content="UNIQUE-Q1")], + response_messages=[ConversationMessage(message_type=MessageType.ASSISTANT, content="A1")], + annotations=[], + ), + # Modified turn: full cumulative array with delta boundary. + ConversationTurn( + call_id="call-2", + timestamp="2026-03-31T10:01:00", + model="claude-3-opus", + request_messages=[ + ConversationMessage(message_type=MessageType.USER, content="UNIQUE-Q1"), + ConversationMessage(message_type=MessageType.ASSISTANT, content="A1"), + ConversationMessage(message_type=MessageType.USER, content="Q2"), + ], + response_messages=[], + annotations=[], + request_was_modified=True, + request_delta_start=1, + ), + ] + session = SessionDetail( + session_id="sess-1", + first_timestamp="2026-03-31T10:00:00", + last_timestamp="2026-03-31T10:01:00", + turns=turns, + total_policy_interventions=0, + models_used=["claude-3-opus"], + ) + + markdown = export_session_markdown(session) + + assert markdown.count("UNIQUE-Q1") == 1 + + def test_jsonl_export_respects_delta_start(self): + turn = ConversationTurn( + call_id="call-2", + timestamp="2026-03-31T10:01:00", + model="claude-3-opus", + request_messages=[ + ConversationMessage(message_type=MessageType.USER, content="Q1"), + ConversationMessage(message_type=MessageType.ASSISTANT, content="A1"), + ConversationMessage(message_type=MessageType.USER, content="Q2"), + ], + response_messages=[], + annotations=[], + request_was_modified=True, + request_delta_start=1, + ) + session = SessionDetail( + session_id="sess-1", + first_timestamp="2026-03-31T10:01:00", + last_timestamp="2026-03-31T10:01:00", + turns=[turn], + total_policy_interventions=0, + models_used=["claude-3-opus"], + ) + + record = json.loads(export_session_jsonl(session)) + + assert [m["content"] for m in record["request_messages"]] == ["A1", "Q2"]