In-run history compaction via pydantic-ai history_processors

[JSv4]

Summary

• Wires shrink_old_artifacts_processor into the single agent-construction chokepoint (make_pydantic_ai_agent) as a ProcessHistory capability so every chat path (corpus, document, structured extraction, deep-research) gets in-run history compaction with zero per-call-site changes.
• Threshold-gated: cheap no-op when under threshold_ratio * context_window. When over threshold, truncates older ToolReturnPart.content to in_run_tool_return_target_chars and drops older ThinkingPart instances, preserving tool_call_id correlation throughout.
• Streamed chats emit a ThoughtEvent with metadata["in_run_shrink"]; non-streamed chats get log-only INFO telemetry.

Closes the gap surfaced by PR #1814 deep-research review: the existing once-per-turn compaction only handled persisted ChatMessage rows; in-loop tool-call/tool-result messages were accumulating unbounded.

Architecture

• opencontractserver/llms/history_processors.py (new) — pure-function shrink_old_artifacts_processor + InRunShrinkEvent dataclass + helpers (_estimate_total_tokens, _resolve_config, _split_protected_suffix, _shrink_tool_return_part, _shrink_message). Defensive top-level try/except so processor bugs can never break an agent run.
• CompactionConfig gains four fields (in_run_enabled, in_run_keep_recent_pairs, in_run_tool_return_target_chars, in_run_drop_thinking) with sensible defaults. in_run_enabled=False is the kill switch.
• PydanticAIDependencies gains a compaction: CompactionConfig field and an on_in_run_shrink telemetry callback.
• make_pydantic_ai_agent unconditionally prepends ProcessHistory(shrink_old_artifacts_processor) to the agent's capabilities= list. Both legacy history_processors=[...] (auto-wrapped) and modern capabilities=[...] caller kwargs are accepted and run after ours.
• PydanticAICoreAgent._stream_core wires the telemetry callback to append a ThoughtEvent to the streaming TimelineBuilder. Non-streaming chats leave the callback unset → log-only.

API choice

Uses the modern pydantic-ai capabilities API (pydantic_ai.capabilities.ProcessHistory), available since pydantic-ai 1.71+. The codebase is pinned to >=1.102.0,<2, so the new API is always available — no warning suppression or deprecation-form fallback needed.

Test plan

• Unit: 12 tests in test_history_processors.py covering threshold gate, shrink, ThinkingPart drop, tool_call_id preservation, last-message invariant, short histories, disabled config, deps=None, callback contract, production-path config resolution, empty-parts guard.
• Factory: 9 tests in test_pydantic_ai_factory.py — existing system_prompt guards + 3 new tests pinning ProcessHistory injection order, legacy-kwarg auto-wrapping, and modern-kwarg passthrough.
• Manual: deep-research run against a verbose retrieval task; confirm [history_processors] In-run shrink log line fires and the report still finalises with citations intact.

[claude]

Code Review: In-run history compaction via pydantic-ai history_processors

Overview

This PR wires a threshold-gated shrink_old_artifacts_processor into make_pydantic_ai_agent as a ProcessHistory capability, closing the gap where tool-call/tool-result messages accumulated unbounded inside a single Agent.run() loop. The architecture is clean — pure functions in a dedicated history_processors.py, injected at the single factory chokepoint, with a defensive top-level try/except that prevents processor bugs from crashing agent runs. The test coverage (12 unit tests + 3 factory tests) is solid.

---

Bug: _apply_context_budget never copies CompactionConfig to agent_deps.compaction

File: opencontractserver/llms/agents/pydantic_ai_agents.py:448
Severity: Medium — per-conversation overrides to in_run_* fields silently don't reach the processor

The new compaction: CompactionConfig field on PydanticAIDependencies has a doc comment that says:

PydanticAICoreAgent subclasses copy AgentConfig.compaction here at construction time (see _apply_context_budget).

But _apply_context_budget (line 415–453) only copies the scalar config.compaction.threshold_ratio → agent_deps.compaction_threshold_ratio. It never does agent_deps.compaction = config.compaction. The result: agent_deps.compaction is always the default CompactionConfig(), regardless of what the caller passed in AgentConfig. Any per-conversation tuning of in_run_enabled, in_run_keep_recent_pairs, etc. is silently dropped.

The fix is one line in _apply_context_budget:

agent_deps.compaction = config.compaction

---

Code smell: production code with a test-stub fallback path

File: opencontractserver/llms/history_processors.py:368–375

_resolve_config falls back to a config_compaction attribute name that exists only in the test stub _FakeDeps, not in PydanticAIDependencies:

direct = getattr(deps, "compaction", None)
if isinstance(direct, CompactionConfig):
    cfg = direct
else:
    legacy = getattr(deps, "config_compaction", None)  # test-stub name only
    ...

Production code should not accommodate test-stub attribute names. The cleaner fix is to update _FakeDeps to use compaction (matching the production field name) and remove the config_compaction fallback branch entirely. The test test_resolves_compaction_from_deps_compaction_field already validates the production path — the stub should be consistent with it.

---

Brittle test: access to pydantic-ai internal root_capability.capabilities

File: opencontractserver/tests/test_pydantic_ai_factory.py:1176

caps = agent.root_capability.capabilities

root_capability is a pydantic-ai internal attribute that isn't part of the public API. Given the version constraint is >=1.102.0,<2, the upper bound leaves room for a pydantic-ai release to rename or restructure this. If the attribute moves, these three new tests will break without any change to OpenContracts code.

Consider wrapping the access in a try/AttributeError with a clear failure message, or documenting a minimum pydantic-ai version that guarantees this shape. At a minimum, add a comment noting this is an internal API access.

---

Minor: hardcoded message index in test_drops_older_thinking_parts

File: opencontractserver/tests/test_history_processors.py:884

recent_resp = result[7]  # first ModelResponse in the protected recent suffix

The long block comment above explains why index 7 is correct for the current fixture, but if the fixture ever changes (e.g. keep_recent_pairs default changes or another pair is added), the assertion silently starts checking the wrong message. A more robust approach:

recent_resp = next(m for m in result if isinstance(m, ModelResponse) and any(isinstance(p, ThinkingPart) for p in m.parts))

This makes the intent self-documenting and doesn't depend on positional arithmetic.

---

Minor: shrink_old_artifacts_processor is async but never awaits

File: opencontractserver/llms/history_processors.py:483

The function is declared async but contains no await. This is correct given pydantic-ai expects an async processor, but worth noting: if pydantic-ai's ProcessHistory ever calls the processor in a thread-pool executor (for sync processors), the async declaration is the right future-proofing.

---

Minor: dual import style for dataclasses

File: opencontractserver/llms/history_processors.py:301–302

import dataclasses
from dataclasses import dataclass

Both are used (dataclasses.replace and @dataclass), so neither is spurious, but it reads slightly odd. Using from dataclasses import dataclass, replace and then replace(...) is more consistent with the rest of the codebase.

---

Positive notes

• The defensive try/except wrapping the entire processor body is the right call — processor bugs must never kill an agent run.
• Threshold-gating the hot path (returning messages unchanged below threshold) keeps the common case very cheap.
• The kill switch (in_run_enabled=False) is in exactly the right place in CompactionConfig.
• Preserving tool_call_id during truncation (shrinking content in-place rather than dropping the ToolReturnPart) is the correct design — it avoids confusing the model about orphaned ToolCallPart entries.
• The empty-parts guard in _shrink_message (never producing a ModelResponse with zero parts) covers a real edge case cleanly.
• Putting all constants in context_guardrails.py rather than hardcoding them in history_processors.py follows the project convention.
• The streaming telemetry path (sync callback → ThoughtEvent) and non-streaming path (log-only) are well separated.
• CHANGELOG is updated correctly.

---

Summary

The core fix (one missing line in _apply_context_budget) and the test-stub cleanup are the two changes worth addressing before merge. The rest are minor polish items.

[claude]

Code Review: In-run history compaction via pydantic-ai history_processors

Overview

This PR closes a genuine gap: the existing compaction pipeline ran once per turn (before Agent.run()), but tool calls/returns generated inside the autonomous loop accumulated unbounded. The approach — a pydantic-ai ProcessHistory capability injected at the single agent-construction chokepoint — is architecturally sound and the implementation is careful and well-tested.

---

What Works Well

• Defensive outer try/except in shrink_old_artifacts_processor — processor bugs cannot crash agent runs. The same pattern is applied to the telemetry callback.
• Empty-parts guard for ModelResponse (a thinking-only response is left intact rather than producing an empty parts list).
• tool_call_id correlation preserved — shrinking is in-place, never dropping ToolCallPart or its paired ToolReturnPart.
• Kill switch (in_run_enabled=False) is cheap to flip without a deploy.
• Constants in the right place — new IN_RUN_* values land in context_guardrails.py per the no-magic-numbers convention.
• Test coverage is comprehensive: threshold gate, shrink, ThinkingPart drop, id preservation, short history no-op, disabled config, deps=None, callback contract, factory injection order.

---

Issues to Address

1. dataclasses.replace fragility on pydantic-ai message parts (moderate risk)

history_processors.py imports replace from dataclasses and applies it to ToolReturnPart, ModelRequest, and ModelResponse. dataclasses.replace only works on Python stdlib @dataclass instances. If a future pydantic-ai version switches any of these types to Pydantic v2 BaseModel, this will raise TypeError: replace() should be called on dataclass instances at runtime — silently swallowed by the outer except Exception and the processor becomes a no-op with no visible error.

Suggestion: Add an import-time assertion to make the assumption explicit and the failure loud:

from dataclasses import fields as _dc_fields
assert hasattr(ToolReturnPart, '__dataclass_fields__'), (
    "pydantic-ai changed ToolReturnPart to a non-dataclass type; "
    "replace() in history_processors.py will silently fail."
)

---

2. Token estimation always runs — even below threshold

shrink_old_artifacts_processor calls _estimate_total_tokens unconditionally on every pre-request invocation. For a long deep-research run with many large tool returns, this O(n*m) scan runs on every model call including the hot path (under threshold). The actual LLM call dominates latency so this is unlikely to be visible in practice — but a comment acknowledging the tradeoff would help future readers.

---

3. _split_protected_suffix counts ModelResponse boundaries, not strict pairs

The docstring says "keep_recent_pairs pairs of ModelResponse+ModelRequest" but the algorithm only counts ModelResponse boundaries. If a ModelResponse is ever followed by multiple ModelRequests, the "pairs" count would be off. In normal pydantic-ai flow this doesn't occur, but the docstring claim is slightly stronger than the implementation. Either tighten the docstring or add a comment clarifying the single-response-per-pair assumption.

---

4. Missing test: callback exception is silently swallowed

The processor defensively catches exceptions from the telemetry callback with # pragma: no cover - defensive. Since the callback contract is a new external-facing seam, consider adding a test to verify a raising callback doesn't propagate. This makes the defensive intent verifiable rather than just asserted.

---

5. threshold_ratio shared between outer and in-run compaction — document the intent

shrink_old_artifacts_processor uses cfg.threshold_ratio (the same field as the outer turn-level compaction). A brief comment stating "shares threshold_ratio with outer compaction by design" would prevent future readers from incorrectly "fixing" this.

---

6. _agent_capabilities uses a pydantic-ai private accessor

test_pydantic_ai_factory.py accesses agent.root_capability.capabilities. The helper correctly documents this fragility with a pointed error message. Worth noting in the changelog that these tests are sensitive to pydantic-ai minor version bumps.

---

Minor / Nits

• on_in_run_shrink: Optional[Callable[[Any], None]] in PydanticAIDependencies — the Any avoids a circular import, but a TYPE_CHECKING import of InRunShrinkEvent would give better IDE/mypy support without runtime cost.
• _TRIM_NOTICE_TEMPLATE elided count is relative to post-stringify length; a short comment would pre-empt confusion.
• test_drops_older_thinking_parts — the assertion thinking_parts_dropped == 3 is correct but the fixture is non-obvious; a brief inline enumeration (old_resp + R0 + R1) in the comment would help maintainers trace through it.

---

Summary

Solid approach, well-executed. Three items worth addressing before merge:

1. Verify dataclasses.replace assumption — add an import-time guard so a silent no-op becomes a loud failure when pydantic-ai changes these types.
2. Clarify threshold_ratio dual-use in CompactionConfig docs or the processor docstring.
3. Add a test for the raising-callback defensive path (or explicitly accept the gap).

[claude]

Review posted below

[claude]

UPDATE: The review has been posted above but the content needs to be replaced. Please see the detailed review analysis below.

[claude]

Code Review — PR 1817: In-run history compaction via pydantic-ai history_processors

Overview

This PR wires shrink_old_artifacts_processor into every pydantic-ai agent via make_pydantic_ai_agent, addressing the gap from PR 1814 where in-loop tool call/return messages accumulated unbounded. The architecture is clean: a single injection point, threshold-gated, with a defensively-wrapped outer try/except so a processor bug can never break an agent run.

---

Strengths

• Single chokepoint injection via make_pydantic_ai_agent. Zero per-call-site changes required; the invariant cannot be accidentally bypassed on new call sites.
• Layered fault tolerance: outer try/except for processor bugs, inner try/except for callback failures, correctly separated so a telemetry sink dying does not undo actual compaction. test_callback_exception_does_not_propagate pins this contract precisely.
• Empty-parts guard in _shrink_message (returning msg, 0, 0 when dropping the only ThinkingPart would empty the parts list) is subtle but essential.
• Constants: all new tunable values land in constants/context_guardrails.py per project conventions.
• Test coverage: 12 unit tests + 3 factory tests + 4 new CompactionConfig tests covering threshold gate, shrink, ThinkingPart drop, tool_call_id preservation, last-message invariant, short histories, disabled config, deps=None, callback contract, production-path config resolution, and the empty-parts guard.

---

Issues

Issue 1: Legacy processor wrapping may fail at runtime (medium)

In pydantic_ai_factory.py the factory wraps caller-supplied history_processors with ProcessHistory(fn). ProcessHistory expects (RunContext, list[ModelMessage]) -> list[ModelMessage]. The factory tests pass functions that only accept messages (one argument), which passes inspection but raises TypeError at runtime when pydantic-ai invokes fn(ctx, messages). If any existing call site passes single-argument processors, they are silently wrapped into something that crashes on the first tool call.

Recommendation: document the required (ctx, messages) signature in the factory docstring, or validate at wrap time with inspect.signature and raise a clear error early.

Issue 2: Import-time assertion introduces a startup failure path (low)

The module-level loop in history_processors.py checking dataclass_fields on pydantic-ai message types is intentional (fail loudly at import time), but is not in the Common Pitfalls section of CLAUDE.md. A developer doing a routine pip install --upgrade pydantic-ai will get a RuntimeError on manage.py runserver with a non-obvious traceback. Worth adding a pitfall entry.

---

Observations

3. _agent_capabilities reaches into a pydantic-ai internal (test_pydantic_ai_factory.py)

Uses agent.root_capability.capabilities. Well-documented with a nice AttributeError tripwire. Consider adding a link to the pydantic-ai source line so a future version bump is easier to diagnose.

4. on_in_run_shrink typed as Callable[[Any], None] (pydantic_ai_tools.py)

The Any presumably avoids a circular import from history_processors.py. A TYPE_CHECKING guard would keep the specific InRunShrinkEvent type in annotations without a runtime import cycle. Minor, but helps IDEs and mypy.

5. Over-threshold-but-unshrinkable log level

When over threshold but the only large content is in UserPromptPart (no ToolReturnPart or ThinkingPart in the older prefix), the processor logs WARNING. This is correct but slightly misleading since the content is structurally uncompressible by this processor. An INFO with a note that UserPromptPart is not compacted might be less alarming in production log monitors.

6. test_drops_older_thinking_parts uses a fragile positional index

The test checks result[1] for the old ModelResponse. If the fixture changes the index breaks silently. Prefer the next(...) pattern used a few lines down for the recent ThinkingPart check.

7. _TRIM_NOTICE_TEMPLATE is private but relied on by tests

Tests assert "in-run trim" in old_return_part.content. Consider promoting the stable substring to a public constant in constants/context_guardrails.py so tests can import it rather than string-matching a private template.

---

Summary

Solid feature. Architecture, defensive error handling, and test coverage are all strong. Two items worth addressing before merge:

1. Legacy processor signature contract: document or validate the expected (ctx, messages) signature so call sites do not silently wrap single-argument processors that crash at runtime.
2. CLAUDE.md pitfall entry for the import-time RuntimeError: a pydantic-ai version bump should not surface as a mysterious startup crash to oncall.

Everything else is non-blocking polish.

[claude]

Code Review: In-run history compaction via pydantic-ai history_processors

Overview

This PR wires a threshold-gated shrink_old_artifacts_processor into every pydantic-ai agent constructed through make_pydantic_ai_agent, closing the context-pressure gap where tool-call/return messages inside a long Agent.run() loop were accumulating unbounded. The implementation is thoughtful and overall quality is high. A few findings below, from a subtle test coverage gap to minor implementation notes.

---

Findings

1. test_thinking_only_modelresponse_is_not_emptied does not actually exercise the guard it is named for

opencontractserver/tests/test_history_processors.py

The fixture places only_thinking (600k-char ThinkingPart) in the older prefix, but no other shrinkable content exists there — the R0 tool return is only 50 chars, well under target_chars=4000. The _shrink_message call on only_thinking does enter the empty-parts guard (preventing an empty ModelResponse), but the guard returns (msg, 0, 0), so both thinking_parts_dropped and tool_returns_shrunk stay 0 for the entire older prefix. The processor then takes the early-return path:

if tool_returns_shrunk == 0 and thinking_parts_dropped == 0:
    logger.info(...)
    return messages   # actual return path in this test

The assertion passes, but because nothing changed at all — not because the guard preserved only_thinking while other content was being shrunk. A stronger fixture that also places a large ToolReturnPart in the older prefix would force the guard code path and give the test its stated meaning.

2. _shrink_tool_return_part stringifies non-string content with str() instead of json.dumps()

opencontractserver/llms/history_processors.py lines 519-520

ToolReturnPart.content can be a list or dict. str() on those produces Python repr-style output rather than valid JSON. The model may struggle to parse Python-style dicts. json.dumps(content, default=str) would be more appropriate and consistent with how tool outputs are serialized elsewhere in the codebase.

3. Final shrunk content can exceed target_chars by the trim-notice length; the validator and docstring do not mention this

opencontractserver/llms/history_processors.py lines 525-528

After truncating to target_chars, the trim-notice template (~40 chars) is appended, making the final content slightly longer than target_chars. Fine in practice, but the constant comment or module docstring should mention the overhead so callers who set a tight ceiling are not surprised.

4. _agent_capabilities in tests reaches into a pydantic-ai internal

opencontractserver/tests/test_pydantic_ai_factory.py

agent.root_capability.capabilities is not part of the public pydantic-ai API. It is well-documented here with upstream file pointers and a tripwire — the defensive error message is excellent. Just flagging it as a known fragility for the next pydantic-ai bump. No change needed; the documentation handles it.

5. Closure over user_msg_id / llm_msg_id in _on_in_run_shrink — worth a comment

opencontractserver/llms/agents/pydantic_ai_agents.py — _stream_core

The closure captures user_msg_id and llm_msg_id via Python late-binding. If either is reassigned later in _stream_core's body the callback would see the updated value on a subsequent in-loop shrink. Since _stream_core handles a single user turn this is almost certainly safe, but a brief comment confirming the IDs are stable for the lifetime of the closure would preempt the question during future refactors of the streaming loop.

---

What is Done Well

• Defensive layering: import-time @dataclass assertion + outer try/except + empty-parts guard is the right design. Silent no-ops from pydantic-ai internals changing shape would be far worse than a loud startup error.
• Telemetry design: separating the streaming ThoughtEvent path from the log-only non-streaming path is clean, and catching callback exceptions locally so a dead WebSocket cannot silently disable compaction is exactly right.
• Threshold reuse: sharing threshold_ratio between turn-level and in-run compaction keeps the kick-in semantics coherent. The comment explaining why to avoid splitting it into two ratios is worth its weight.
• CompactionConfig propagation fix: the regression test test_compaction_config_is_copied_to_deps cleanly pins the bug where only threshold_ratio was forwarded — a real correctness gap.
• Test coverage: 12 unit tests for the processor + 3 factory tests + 4 CompactionConfig field tests + 1 regression test is thorough. The log-level assertion in test_no_shrinkable_content_logs_info (INFO, not WARNING) is a nice touch for production log monitors.

---

Summary

Two items are worth addressing before merge:

1. test_thinking_only_modelresponse_is_not_emptied: strengthen the fixture so the empty-parts guard is the actual reason the ThinkingPart survives. Add a large ToolReturnPart alongside only_thinking in the older prefix so the processor has something to shrink while the guard prevents emptying the thinking-only response.
2. str() to json.dumps() for non-string ToolReturnPart.content: avoids sending Python repr output to the model for structured tool results.

Everything else is non-blocking.