Preserve active DoD in compaction

Status	File	+
M	`src/loader/runtime/compaction.py`	8
M	`src/loader/runtime/session.py`	115
M	`tests/test_compaction.py`	31
M	`tests/test_session_state.py`	53

src/loader/runtime/compaction.pymodified

      budget: SummaryCompressionBudget | None = None,
      previous_summary: str | None = None,
      current_task: str | None = None,
++    active_dod_summary: str | None = None,
      original_input_tokens: int | None = None,
  ) -> SessionCompactionResult | None:
      """Compact older messages into one continuation summary message."""
          removed_messages,
          previous_summary=previous_summary,
          current_task=current_task,
++        active_dod_summary=active_dod_summary,
      )
      compression = compress_summary(summary_text, budget=budget)
      summary_message = Message(
              "Continuation instructions:\n"
              "- Continue from the preserved recent messages.\n"
              "- Honor the active DoD, workflow mode, and permission mode.\n"
++            "- Treat active DoD/check-failing messages as authoritative over "
++            "older summaries or durable memory notes.\n"
              "- Do not ask the user to repeat already-captured context unless essential."
          ),
      )
      *,
      previous_summary: str | None = None,
      current_task: str | None = None,
++    active_dod_summary: str | None = None,
  ) -> str:
      """Build a structured session summary before compression."""
          lines.append(f"- Confirmed facts: {confirmed_facts}")
      if preferred_next_step:
          lines.append(f"- Preferred next step: {preferred_next_step}")
++    if active_dod_summary:
++        lines.append(f"- Active DoD: {active_dod_summary}")
      if previous_summary:
          lines.append("- Previously compacted context retained.")
      lines.append(f"- Newly compacted context: {len(messages)} earlier message(s) summarized.")
              "- Recent user requests:",
              "- Confirmed facts:",
              "- Preferred next step:",
++            "- Active DoD:",
              "- Previously compacted context:",
              "- Newly compacted context:",
          )

src/loader/runtime/session.pymodified

  from __future__ import annotations
  import json
++import re
  import secrets
  from collections.abc import Callable
  from dataclasses import dataclass, field
  MAX_ROTATED_FILES = 3
  _UNSET = object()
  _REQUEST_TOOL_PAYLOAD_SUMMARY_THRESHOLD = 240
++_ABSOLUTE_PATH_SUMMARY_PATTERN = re.compile(
++    r"(?P<path>/(?:Users|home|tmp|var|private)/[^\s:]+)"
++)
  def _project_request_tool_call(tool_call: ToolCall) -> ToolCall | None:
      return datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%SZ")
++def _compact_inline_text(value: Any, *, max_chars: int = 220) -> str:
++    text = " ".join(str(value or "").split())
++    if len(text) <= max_chars:
++        return text
++    return f"{text[: max_chars - 3]}..."
++
++
++def _shorten_summary_paths(value: str) -> str:
++    def replace(match: re.Match[str]) -> str:
++        path = Path(match.group("path"))
++        parent = path.parent.name
++        if parent:
++            return f".../{parent}/{path.name}"
++        return f".../{path.name}"
++
++    return _ABSOLUTE_PATH_SUMMARY_PATTERN.sub(replace, value)
++
++
++def _preview_sequence(values: Any, *, limit: int = 3) -> str:
++    if not isinstance(values, list):
++        return ""
++    items = [
++        _compact_inline_text(value, max_chars=140)
++        for value in values
++        if str(value or "").strip()
++    ]
++    if not items:
++        return ""
++    preview = ", ".join(items[:limit])
++    if len(items) > limit:
++        preview += ", ..."
++    return preview
++
++
++def _latest_failed_evidence_preview(
++    evidence_items: Any,
++    *,
++    limit: int = 3,
++) -> str:
++    if not isinstance(evidence_items, list):
++        return ""
++    for item in reversed(evidence_items):
++        if not isinstance(item, dict):
++            continue
++        if item.get("passed") is True or item.get("skipped") is True:
++            continue
++        text = "\n".join(
++            str(item.get(key) or "")
++            for key in ("output", "stdout", "stderr")
++            if item.get(key)
++        )
++        issue_lines: list[str] = []
++        for raw_line in text.splitlines():
++            line = _compact_inline_text(
++                _shorten_summary_paths(str(raw_line)),
++                max_chars=180,
++            )
++            if not line:
++                continue
++            lowered = line.lower()
++            if lowered.startswith(
++                (
++                    "exit code ",
++                    "html guide content quality issues:",
++                    "missing local html links:",
++                    "verification:",
++                )
++            ):
++                continue
++            issue_lines.append(line)
++            if len(issue_lines) >= limit:
++                break
++        if issue_lines:
++            return "; ".join(issue_lines)
++    return ""
++
++
++def _summarize_active_dod_snapshot(data: dict[str, Any]) -> str | None:
++    parts: list[str] = []
++    status = _compact_inline_text(data.get("status"))
++    if status:
++        parts.append(f"status={status}")
++    verification = _compact_inline_text(data.get("last_verification_result"))
++    if verification:
++        parts.append(f"last verification={verification}")
++    failed = _latest_failed_evidence_preview(data.get("evidence"))
++    if failed:
++        parts.append(f"latest failed verifier={failed}")
++    pending = _preview_sequence(data.get("pending_items"))
++    if pending:
++        parts.append(f"pending={pending}")
++    if not parts:
++        return None
++    return _compact_inline_text("; ".join(parts), max_chars=700)
++
++
  def _generate_session_id() -> str:
      timestamp = datetime.now(UTC).strftime("%Y%m%dT%H%M%SZ")
      return f"{timestamp}-{secrets.token_hex(4)}"
              keep_last_messages=self.compaction_keep_last_messages,
              previous_summary=self.compaction.summary if self.compaction else None,
              current_task=self.current_task,
++            active_dod_summary=self._active_dod_compaction_summary(),
              original_input_tokens=estimated_input_tokens,
+         )
          if result is None:
          self.persist()
          return result
++    def _active_dod_compaction_summary(self) -> str | None:
++        if not self.active_dod_path:
++            return None
++        path = Path(self.active_dod_path)
++        if not path.exists():
++            return None
++        try:
++            data = json.loads(path.read_text())
++        except (OSError, json.JSONDecodeError):
++            return None
++        if not isinstance(data, dict):
++            return None
++        return _summarize_active_dod_snapshot(data)
++
      def record_turn_usage(
          self,
          usage: dict[str, int],

tests/test_compaction.pymodified

+     ]
      assert result.messages[0].content.startswith("[COMPACTED CONTEXT]")
      assert "Continuation instructions:" in result.messages[0].content
++    assert (
++        "authoritative over older summaries or durable memory notes"
++        in result.messages[0].content
++    )
++
++
++def test_compact_session_messages_includes_active_dod_summary() -> None:
++    messages = [
++        Message(role=Role.USER, content="Create the generated guide."),
++        Message(role=Role.ASSISTANT, content="Wrote the guide files."),
++        Message(
++            role=Role.TOOL,
++            content="Observation [notepad_read]: Result: guide complete",
++        ),
++        Message(role=Role.ASSISTANT, content="I will finish."),
++        Message(role=Role.USER, content="Continue repairing."),
++    ]
++
++    result = compact_session_messages(
++        messages,
++        keep_last_messages=2,
++        current_task="Create the generated guide.",
++        active_dod_summary=(
++            "status=fixing; last verification=failed; "
++            "latest failed verifier=/tmp/guide/chapter.html: thin content"
++        ),
++    )
++
++    assert result is not None
++    assert "- Active DoD: status=fixing; last verification=failed;" in result.summary
++    assert "/tmp/guide/chapter.html: thin content" in result.summary
  def test_build_session_summary_skips_nested_compacted_context_content() -> None:

tests/test_session_state.pymodified

  from loader.agent.loop import Agent, AgentConfig, ReasoningConfig
  from loader.llm.base import CompletionResponse, Message, Role, ToolCall
  from loader.runtime.completion_trace import CompletionTraceEntry
++from loader.runtime.dod import (
++    DefinitionOfDoneStore,
++    VerificationEvidence,
++    create_definition_of_done,
++)
  from loader.runtime.evidence_provenance import EvidenceProvenance
  from loader.runtime.prompt_history import PromptSnapshot
  from loader.runtime.runtime_handle import RuntimeHandle
+     ]
++def test_session_compaction_summarizes_active_dod_failure(temp_dir: Path) -> None:
++    dod_store = DefinitionOfDoneStore(temp_dir)
++    dod = create_definition_of_done("Create a generated guide.")
++    dod.status = "fixing"
++    dod.last_verification_result = "failed"
++    dod.pending_items = ["Expand generated chapters to satisfy quality verification"]
++    dod.evidence.append(
++        VerificationEvidence(
++            command="python3 verify_html_quality.py",
++            passed=False,
++            output=(
++                "Exit code 1\n"
++                "HTML guide content quality issues:\n"
++                f"{temp_dir / 'guide' / 'chapters' / '05-load-balancing.html'}: "
++                "thin content (1500 text chars, expected at least 1758)\n"
++            ),
++        )
++    )
++    dod_path = dod_store.save(dod)
++    session = ConversationSession(
++        system_message_factory=_dummy_system,
++        few_shot_factory=_dummy_few_shots,
++        project_root=temp_dir,
++        messages=[
++            Message(role=Role.USER, content="Create the guide."),
++            Message(role=Role.ASSISTANT, content="Created draft files."),
++            Message(
++                role=Role.TOOL,
++                content="Observation [notepad_read]: Result: guide complete",
++            ),
++            Message(role=Role.ASSISTANT, content="Trying to finish."),
++            Message(role=Role.USER, content="Continue repairing."),
++        ],
++        active_dod_path=str(dod_path),
++        auto_compaction_input_tokens_threshold=1,
++        compaction_keep_last_messages=2,
++    )
++
++    result = session.maybe_compact()
++
++    assert result is not None
++    assert session.messages[0].content.startswith("[COMPACTED CONTEXT]")
++    assert "- Active DoD: status=fixing; last verification=failed" in result.summary
++    assert "05-load-balancing.html" in result.summary
++    assert "thin content" in result.summary
++    assert "authoritative over older summaries or durable memory notes" in result.summary
++
++
  def test_build_request_messages_omits_large_mutation_tool_calls_from_history(
      temp_dir: Path,
  ) -> None:

tenseleyflow/loader / `7f616be`

4 changed files