4.3: conversation compaction (text-only synthetic message)

Phase 4.3 — when a session's message history exceeds the configured
threshold (default 20 messages), the oldest portion gets summarised
into one synthetic assistant message and the most-recent
``keep_recent`` messages (default 10) are kept verbatim, INCLUDING
their original thinking blocks. Prevents unbounded context growth
on long sessions; sliced messages persist to
~/.tdpilot-api/history/<session>.jsonl for forensic recall.

Critical safety deviation from the plan's strawman:
The plan asked for a synthesized ``thinking`` block in the synthetic
message "to satisfy the API contract". That's actually the WRONG
shape — DeepSeek/Anthropic-compat thinking blocks carry a
cryptographic ``signature`` field signed by the API itself. A
synthesized signature wouldn't validate; the next API call would
HTTP 400. Text-only IS the safe path:

  * Compaction REPLACES the older turns entirely — no thinking
    blocks from those need echoing back.
  * Retained recent messages keep their ORIGINAL (validly-signed)
    thinking blocks, which echo back unchanged.
  * The synthetic message has no thinking block, so there's
    nothing for the API to validate against.

Documented in tdpilot_api_compaction.py module docstring.

td_component/tdpilot_api_compaction.py (NEW):
  - needs_compaction(messages, threshold) — bool predicate.
  - compact(messages, *, keep_recent) — pure function returning
    [synthetic_summary, *most_recent_N]. Does not mutate input;
    handles edge cases (keep_recent >= len, empty, mid-tool-chain
    slice point). Synthetic message: text-only, marker
    ``[[TDPILOT_COMPACTED_HISTORY]]`` plus a heuristic summary.
  - _summarise_old_turns — local heuristic (no LLM call): user
    prompt count + first/last sample, dedup'd tool names with
    counts, error count.
  - persist_history_chunk(messages, session_id, history_dir) —
    appends a JSONL record before each compaction. Best-effort:
    OS errors print a warning but never block the agent loop.
  - read_history(session_id) — re-reads the archive; used by
    a future ``td_history_recall`` tool / debug surfaces.
  - Compactor class — owns session_id + threshold + keep_recent +
    history_dir + persist toggle. ``maybe_compact(messages)`` is
    the agent-friendly entry; idempotent below threshold; tracks
    ``compactions_run`` for observability.

Agent integration (td_component/tdpilot_api_agent.py):
  - ``Agent.__init__`` gains ``compactor`` kwarg (Any, optional).
  - ``_loop`` calls ``compactor.maybe_compact(self.messages)`` ONCE
    at the top of every turn, before the model tier resolves.
    Wrapped in try/except — a crashing compactor must never break
    a turn.

Runtime wiring (td_component/tdpilot_api_runtime.py):
  - ``AgentRuntime._build_compactor()`` constructs the per-runtime
    Compactor from ``config["compaction_threshold"]`` (default 20)
    + ``config["compaction_keep_recent"]`` (default 10) +
    ``config["history_dir"]``. Reuses the tracer's session_id when
    available so forensic history files line up with the trace
    timeline.
  - Returns None when the compaction module isn't importable
    (older .tox builds) or threshold == 0 (explicit disable).
  - Wired into the Agent constructor.

Build script + extension imports updated.

Tests (tests/test_tdpilot_api_compaction.py): 26 cases.
  - Threshold semantics (below / at / zero-disables).
  - compact() shape: synthetic + recent slice; identity-preserving
    no-op below threshold; non-mutating.
  - **Synthetic message text-only contract** — no thinking block,
    no redacted_thinking, no signature on any block.
  - **Recent slice retains thinking blocks intact** including
    signatures (the validly-signed originals API will check).
  - Marker visible in synthetic text body.
  - Summary mentions user count + dedup'd tool names + handles
    no-tool turns + truncates 1000-char user texts.
  - persist_history_chunk writes JSONL, appends, round-trips.
  - read_history of missing file = []; never raises.
  - Compactor: disabled at threshold=0, fires above, no-op below,
    persistence-off mode skips disk, idempotent on re-run.
  - Agent integration: maybe_compact fires exactly once per
    _loop entry; a crashing compactor doesn't break the turn.
  - Runtime: ``compaction_threshold`` config plumbs through;
    threshold=0 yields None compactor.

Pytest 1121 passing (up from 1104). Lints + format clean. Three
td_component sources changed plus one new module — standalone
.tox needs rebuilding before compaction fires live. Live
verification requires a 20+ message conversation; recommend a
follow-up session that runs an extended chat against the
rebuilt .tox to confirm DeepSeek accepts the post-compaction
prefix without a 400.

Author: dreamrec

td_component/build_tdpilot_api_tox.py

@@ -202,6 +202,7 @@ def _load_legacy_module():
     ("tdpilot_api_batch", "textDAT", "td_component/tdpilot_api_batch.py"),
     ("tdpilot_api_recovery", "textDAT", "td_component/tdpilot_api_recovery.py"),
     ("tdpilot_api_tracing", "textDAT", "td_component/tdpilot_api_tracing.py"),
+    ("tdpilot_api_compaction", "textDAT", "td_component/tdpilot_api_compaction.py"),
     ("tdpilot_api_chat_html", "textDAT", "td_component/tdpilot_api_chat.html"),
     ("tdpilot_api_web_callbacks", "textDAT", "td_component/tdpilot_api_web_callbacks.py"),
     ("mcp_webserver_callbacks", "textDAT", "td_component/mcp_webserver_callbacks.py"),

td_component/tdpilot_api_agent.py

@@ -198,6 +198,10 @@ def __init__(
         # it's free to vary turn-to-turn (memory_save propagates here,
         # not in the system prompt).
         dynamic_context_provider: Callable[[], list[dict]] | None = None,
+        # Phase 4.3 — conversation compaction. Optional; when set, the
+        # agent calls ``compactor.maybe_compact(self.messages)`` at the
+        # top of each ``_loop`` iteration. Set None to disable.
+        compactor: Any | None = None,
         # Callbacks — all optional. Receive primitive args.
         on_text: Callable[[str], None] = _noop,
         on_tool_call: Callable[[str, dict], None] = _noop,
@@ -242,6 +246,7 @@ def __init__(
         self.request_timeout = request_timeout
 
         self.dynamic_context_provider = dynamic_context_provider
+        self.compactor = compactor
 
         self.on_text = on_text
         self.on_tool_call = on_tool_call
@@ -337,6 +342,19 @@ def _resolve_model(self, user_text: str) -> str:
         return self.model if score >= 2 else self.flash_model
 
     def _loop(self) -> str | None:
+        # Phase 4.3 — compact the conversation history if it has grown
+        # past the threshold. Runs ONCE at turn start, BEFORE the
+        # model's tier is resolved (the model decision works on the
+        # last user message which is preserved in the recent slice).
+        # The compactor is responsible for forensic persistence
+        # before slicing, so a "lost detail" debug session can recover
+        # the original messages from ~/.tdpilot-api/history/.
+        if self.compactor is not None:
+            try:
+                self.messages = self.compactor.maybe_compact(self.messages)
+            except Exception as exc:  # noqa: BLE001 — compaction must never break a turn
+                print(f"[tdpilot_API/agent] compaction failed: {exc}")
+
         # Pick the model ONCE at turn start. Stays pinned for the entire
         # tool-use chain (mid-turn switching busts DeepSeek's auto-cache
         # AND risks flash failing to finish what pro started). Resolve

td_component/tdpilot_api_compaction.py

@@ -0,0 +1,322 @@
+"""TDPilot API — conversation compaction (Phase 4.3).
+
+When a session's message history exceeds the configured threshold,
+the oldest portion gets summarised into one synthetic assistant
+message and the most-recent ``keep_recent`` messages are kept
+verbatim. Prevents unbounded context growth on long sessions.
+
+Critical thinking-block contract
+================================
+
+DeepSeek's Anthropic-compatible endpoint REQUIRES that any
+``type: thinking`` content blocks emitted by a previous turn be
+echoed back in the next turn's message history. The error on
+mismatch is::
+
+    HTTP 400: The content[].thinking in the thinking mode must
+    be passed back to the API.
+
+The synthetic message we produce here is **text-only — no
+``thinking`` block**. Why this is safe:
+
+  * Compaction REPLACES the older turns entirely (the original
+    user / tool_use / tool_result / assistant messages, including
+    any thinking blocks, are sliced out of the history).
+  * The retained recent messages still carry their ORIGINAL,
+    API-issued thinking blocks (with valid ``signature`` fields).
+    Those echo back unchanged.
+  * The new synthetic message has no thinking block, so there's
+    nothing for the API to validate.
+
+The implementation plan's strawman called for "a synthesized
+thinking block to satisfy the contract" — that's what we'd LIKE
+but can't do safely. ``signature`` fields are signed by the API
+itself; a fabricated signature would 400. Text-only is the
+defensive path the plan's risk note explicitly anticipated.
+
+Forensic preservation
+=====================
+
+Before each compaction, the to-be-removed messages are appended
+to ``~/.tdpilot-api/history/<session_id>.jsonl`` (one JSON
+record per compaction event, the record containing the entire
+sliced batch). Forensic users can reload via ``read_history``.
+"""
+
+from __future__ import annotations
+
+import json
+import time
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any
+
+DEFAULT_HISTORY_DIR = Path.home() / ".tdpilot-api" / "history"
+DEFAULT_THRESHOLD = 20
+DEFAULT_KEEP_RECENT = 10
+
+# Marker the synthesized message carries in its text body so future
+# tooling can recognise compaction summaries (e.g. for re-expansion
+# from the on-disk forensic file).
+COMPACTION_MARKER = "[[TDPILOT_COMPACTED_HISTORY]]"
+
+
+# ---------------------------------------------------------------------------
+# Public API
+# ---------------------------------------------------------------------------
+
+
+def needs_compaction(messages: list[dict], threshold: int = DEFAULT_THRESHOLD) -> bool:
+    """``True`` when the message count is at or above the threshold AND
+    the threshold is positive (zero/negative disables compaction).
+    """
+    if threshold <= 0:
+        return False
+    return len(messages) >= threshold
+
+
+def compact(
+    messages: list[dict],
+    *,
+    keep_recent: int = DEFAULT_KEEP_RECENT,
+) -> list[dict]:
+    """Return a new message list = ``[synthetic_summary, *most_recent_N]``.
+
+    Pure function — does NOT mutate the input. Does NOT touch the
+    filesystem (forensic persistence is the caller's job; see
+    ``persist_history_chunk`` below). Always preserves the most-recent
+    ``keep_recent`` messages verbatim, including any original
+    ``thinking`` content blocks they carry.
+
+    Edge cases:
+      - ``keep_recent`` >= ``len(messages)``: return a copy of
+        ``messages`` unchanged (nothing to compact away).
+      - ``len(messages) <= 1``: return a copy unchanged (degenerate).
+      - The slice point lands mid-tool-chain (assistant tool_use
+        WITHOUT its matching tool_result): we still slice. The
+        synthetic message replaces the gap; the API contract only
+        requires thinking-block echo, not tool-call pairing across
+        the historical boundary.
+    """
+    if keep_recent < 0:
+        keep_recent = 0
+    if len(messages) <= max(1, keep_recent):
+        return list(messages)
+    cut = len(messages) - keep_recent
+    older = messages[:cut]
+    recent = messages[cut:]
+    summary_text = _summarise_old_turns(older)
+    synthetic = {
+        "role": "assistant",
+        "content": [
+            {
+                "type": "text",
+                "text": COMPACTION_MARKER + "\n\n" + summary_text,
+            }
+        ],
+    }
+    return [synthetic, *recent]
+
+
+def persist_history_chunk(
+    messages: list[dict],
+    session_id: str,
+    *,
+    history_dir: Path | None = None,
+) -> Path:
+    """Append one JSONL record to ``<history_dir>/<session_id>.jsonl``
+    capturing the to-be-compacted messages. Returns the file path.
+
+    The record is one JSON line: ``{ts, session_id, message_count,
+    messages}``. Each call appends; the file grows monotonically so a
+    long session with multiple compactions has a readable timeline.
+
+    Best-effort: filesystem errors print a one-line warning and
+    return the path that WOULD have been used. The caller never
+    blocks on persistence failure — context window is the priority.
+    """
+    target_dir = Path(history_dir) if history_dir else DEFAULT_HISTORY_DIR
+    target_path = target_dir / f"{session_id}.jsonl"
+    record = {
+        "ts": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.%fZ"),
+        "session_id": session_id,
+        "message_count": len(messages),
+        "messages": messages,
+    }
+    try:
+        target_dir.mkdir(parents=True, exist_ok=True)
+        with target_path.open("a", encoding="utf-8") as fh:
+            fh.write(json.dumps(record, ensure_ascii=False, default=str) + "\n")
+    except OSError as exc:
+        print(f"[tdpilot_API/compaction] history persist failed: {exc}")
+    return target_path
+
+
+def read_history(
+    session_id: str,
+    *,
+    history_dir: Path | None = None,
+) -> list[dict]:
+    """Re-read every persisted compaction chunk for ``session_id``.
+
+    Returns a list of records (each with ``ts`` /
+    ``message_count`` / ``messages``). Empty list if no file or any
+    parse error — never raises. Used by future ``td_history_recall``
+    tools / debug UIs to reconstruct full session detail after
+    compaction.
+    """
+    target_dir = Path(history_dir) if history_dir else DEFAULT_HISTORY_DIR
+    target_path = target_dir / f"{session_id}.jsonl"
+    if not target_path.is_file():
+        return []
+    out: list[dict] = []
+    try:
+        with target_path.open("r", encoding="utf-8") as fh:
+            for line in fh:
+                line = line.strip()
+                if not line:
+                    continue
+                try:
+                    out.append(json.loads(line))
+                except (TypeError, ValueError):
+                    continue
+    except OSError:
+        return []
+    return out
+
+
+# ---------------------------------------------------------------------------
+# Internal — local-heuristic summarisation
+# ---------------------------------------------------------------------------
+
+
+def _summarise_old_turns(messages: list[dict]) -> str:
+    """Build a text summary of the sliced-away messages.
+
+    Pure local heuristic — no LLM call. Captures:
+      - User-message count + a sample of the first / last user goals.
+      - Tool calls invoked, deduplicated and sorted (most-frequent
+        first), with per-tool counts. Token-frugal: just names.
+      - Notable error events.
+
+    The model reading this gets enough signal to maintain rough
+    continuity without paying full-fidelity context cost. If higher
+    fidelity is needed mid-session, the agent can call a future
+    ``td_history_recall(query)`` tool that hits the on-disk archive.
+    """
+    user_texts: list[str] = []
+    tool_counts: dict[str, int] = {}
+    error_count = 0
+
+    for msg in messages:
+        role = msg.get("role", "")
+        content = msg.get("content", [])
+        if not isinstance(content, list):
+            continue
+        if role == "user":
+            for block in content:
+                if not isinstance(block, dict):
+                    continue
+                if block.get("type") == "text":
+                    text = (block.get("text") or "").strip()
+                    if text and not text.startswith("[[TDPILOT_"):
+                        user_texts.append(text)
+                elif block.get("type") == "tool_result":
+                    if block.get("is_error"):
+                        error_count += 1
+        elif role == "assistant":
+            for block in content:
+                if not isinstance(block, dict):
+                    continue
+                if block.get("type") == "tool_use":
+                    name = block.get("name", "")
+                    if name:
+                        tool_counts[name] = tool_counts.get(name, 0) + 1
+
+    parts: list[str] = []
+    parts.append(f"This summary replaces {len(messages)} compacted messages from earlier in the session.")
+    parts.append(f"User prompts: {len(user_texts)}.")
+    if user_texts:
+        first = user_texts[0]
+        if len(first) > 200:
+            first = first[:197] + "..."
+        parts.append(f"First user goal: {first!r}")
+        if len(user_texts) > 1:
+            last = user_texts[-1]
+            if len(last) > 200:
+                last = last[:197] + "..."
+            parts.append(f"Latest user goal before this point: {last!r}")
+    if tool_counts:
+        ranked = sorted(tool_counts.items(), key=lambda kv: (-kv[1], kv[0]))
+        rendered = ", ".join(f"{n}×{c}" if c > 1 else n for n, c in ranked)
+        parts.append(f"Tools used (deduped): {rendered}.")
+    else:
+        parts.append("No tool calls in this slice (text-only conversation).")
+    if error_count:
+        parts.append(f"Tool errors encountered: {error_count}.")
+    parts.append(
+        "Recent turns are kept verbatim below. If you need detail from "
+        "the compacted slice, call td_history_recall (when available) "
+        "or ask the user to re-state the relevant context."
+    )
+    return "\n".join(parts)
+
+
+# ---------------------------------------------------------------------------
+# Compactor — pairs compact() with persist_history_chunk + tracking
+# ---------------------------------------------------------------------------
+
+
+class Compactor:
+    """Per-session compactor. Owns threshold + keep_recent + the
+    on-disk archive path, and exposes a ``maybe_compact`` method the
+    Agent calls at the top of each ``_loop`` iteration.
+
+    Re-entrant safe — repeated calls with the same messages are
+    a no-op once below the threshold.
+
+    Stats:
+      - ``compactions_run``: count of times this Compactor reduced
+        the history. Surfaced in observability traces (Phase 4.1).
+    """
+
+    def __init__(
+        self,
+        *,
+        session_id: str,
+        threshold: int = DEFAULT_THRESHOLD,
+        keep_recent: int = DEFAULT_KEEP_RECENT,
+        history_dir: Path | None = None,
+        persist: bool = True,
+    ) -> None:
+        self.session_id = session_id
+        self.threshold = max(0, int(threshold))
+        self.keep_recent = max(0, int(keep_recent))
+        self.history_dir = Path(history_dir) if history_dir else DEFAULT_HISTORY_DIR
+        self.persist = bool(persist)
+        self.compactions_run = 0
+        self.last_compaction_ts: float = 0.0
+
+    @property
+    def enabled(self) -> bool:
+        return self.threshold > 0 and self.keep_recent < self.threshold
+
+    def maybe_compact(self, messages: list[dict]) -> list[dict]:
+        """Return a possibly-compacted copy of ``messages``. If no
+        compaction was needed, returns the input unchanged
+        (same-object identity preserved for caller-friendliness).
+        """
+        if not self.enabled:
+            return messages
+        if not needs_compaction(messages, self.threshold):
+            return messages
+        cut = len(messages) - self.keep_recent
+        if cut <= 0:
+            return messages
+        older = messages[:cut]
+        if self.persist and older:
+            persist_history_chunk(older, self.session_id, history_dir=self.history_dir)
+        compacted = compact(messages, keep_recent=self.keep_recent)
+        self.compactions_run += 1
+        self.last_compaction_ts = time.monotonic()
+        return compacted

td_component/tdpilot_api_extension.py

@@ -114,6 +114,7 @@ def _ensure_module_path(self) -> None:
             "tdpilot_api_batch",
             "tdpilot_api_recovery",
             "tdpilot_api_tracing",
+            "tdpilot_api_compaction",
             "mcp_webserver_callbacks",
         ):
             child = self.owner.op(name)