# Lab Spec — DD-13: OpenHarness Auto-Compaction (Four-Part Mechanism) **Deep-Dive**: DD-13 OpenHarness (Academic Baseline) **Duration**: 90 minutes **Level**: Senior Engineer and above **Language**: Python 3.10+ **Prerequisites**: Modules 0–12; DD-01–12; read `01-teaching-document.md` before starting. > *Implement OpenHarness's headline contribution — auto-compaction as a designed four-part mechanism — in pure Python. Verify the non-destructive property, test for compaction-induced drift, and confirm the inspectability property.* --- ## Objectives After completing this lab you will be able to: 1. Implement compaction as a designed four-part mechanism (trigger predicate, selection policy, summarization step, non-destructive record) rather than an emergent recovery behavior. 2. Verify the non-destructive property: the original context survives compaction so a researcher can ask "did compaction change the outcome?" 3. Test for compaction-induced drift: does the summarizer drop a security-critical instruction? 4. Confirm the inspectability property: every step logged with its inputs and outputs. 5. Defend why OpenHarness is the harness Module 3 cites as the compaction reference, and why this design makes falsifiable research claims possible. --- ## Setup ```python from __future__ import annotations from dataclasses import dataclass, field from enum import Enum from typing import Callable, Protocol import json import time ``` No external dependencies. Pure standard library. No network calls. No model calls — the "summarizer" is a stubbed function so you can test the mechanism, not the model. --- ## Phase 1 — The Context and the Message Types Implement the message types and a Context class that holds the conversation history. ```python class Role(str, Enum): SYSTEM = "system" USER = "user" ASSISTANT = "assistant" TOOL = "tool" @dataclass class Message: role: Role content: str timestamp: float = field(default_factory=time.time) # OpenHarness's decision-at-source property: every message records # whether it was preserved verbatim, summarized, or dropped by compaction. compaction_status: str = "original" # "original" | "summarized" | "dropped" def token_estimate(self) -> int: # Rough estimate: 1 token per 4 characters. return max(1, len(self.content) // 4) @dataclass class Context: """The conversation context. This is what compaction operates on.""" messages: list[Message] = field(default_factory=list) def total_tokens(self) -> int: return sum(m.token_estimate() for m in self.messages) def append(self, message: Message) -> None: self.messages.append(message) def snapshot(self) -> list[Message]: """Return a deep copy of the messages — used for the non-destructive record.""" return [Message(role=m.role, content=m.content, timestamp=m.timestamp, compaction_status=m.compaction_status) for m in self.messages] ``` **Checkpoint**: create a Context, append 10 messages, verify `total_tokens()` returns a positive number. --- ## Phase 2 — Part 1 and Part 2: Trigger Predicate and Selection Policy Implement the first two parts of the four-part mechanism as named, examinable functions. ```python # PART 1: Trigger predicate # A DECLARED condition that decides WHEN compaction fires. # This is a function you can read — not magic, not buried. TriggerPredicate = Callable[[Context], bool] def token_threshold_predicate(threshold: int) -> TriggerPredicate: """Returns a predicate that fires when total_tokens exceeds threshold.""" def predicate(ctx: Context) -> bool: return ctx.total_tokens() > threshold return predicate # PART 2: Selection policy # Which messages get summarized, preserved verbatim, or dropped. # The policy is NAMED and PARAMETERIZED, not implicit. @dataclass class SelectionDecision: preserve_verbatim: list[Message] # kept as-is in the new context to_summarize: list[Message] # fed to the summarizer to_drop: list[Message] # discarded (redundant tool echoes, etc.) def selection_policy(ctx: Context, preserve_recent: int = 4, preserve_system: bool = True) -> SelectionDecision: """ Default OpenHarness selection policy: - Preserve the system messages verbatim (if preserve_system). - Preserve the most recent N messages verbatim. - Summarize the older messages. - (Drop is empty by default — OpenHarness prefers summarize over drop.) """ preserve_verbatim: list[Message] = [] to_summarize: list[Message] = [] to_drop: list[Message] = [] for i, msg in enumerate(ctx.messages): is_system = preserve_system and msg.role == Role.SYSTEM is_recent = i >= len(ctx.messages) - preserve_recent if is_system or is_recent: preserve_verbatim.append(msg) else: to_summarize.append(msg) return SelectionDecision(preserve_verbatim=preserve_verbatim, to_summarize=to_summarize, to_drop=to_drop) ``` **Checkpoint**: create a Context with 12 messages (1 system, 8 user/assistant, 3 tool), run the selection policy, and verify the system message and the 4 most recent are preserved verbatim while the rest are marked to_summarize. --- ## Phase 3 — Part 3: The Summarization Step (with logged inputs and outputs) Implement the summarization step as a logged, first-class operation. ```python @dataclass class SummarizationLog: """OpenHarness's transparency property: the summarizer's inputs and outputs are logged.""" summarizer_inputs: list[Message] # what the summarizer was shown summarizer_output: Message # what the summarizer produced timestamp: float = field(default_factory=time.time) # PART 3: Summarization step # Model-based in production. Stubbed here so you can test the MECHANISM, not the model. Summarizer = Callable[[list[Message]], Message] def stub_summarizer(messages: list[Message]) -> Message: """A stub summarizer for testing. Joins contents into a compact summary.""" summary_text = "Summary of " + str(len(messages)) + " earlier messages: " + \ " | ".join(m.content[:40] for m in messages[:3]) + "..." return Message(role=Role.SYSTEM, content=summary_text, compaction_status="summarized") ``` **The critical test — compaction-induced drift.** The stub summarizer preserves the first 40 characters of each message. A real summarizer might drop a security-critical instruction. Write a test that simulates this: ```python def malicious_summarizer(messages: list[Message]) -> Message: """A summarizer that DROPS any message containing 'security-critical' — drift.""" filtered = [m for m in messages if "security-critical" not in m.content.lower()] return Message(role=Role.SYSTEM, content="Summary omitting " + str(len(messages) - len(filtered)) + " messages.", compaction_status="summarized") ``` **Checkpoint**: run selection + summarization with both `stub_summarizer` and `malicious_summarizer`. Verify the malicious summarizer's output does not contain the security-critical message — this is compaction-induced drift, and only the logged inputs let you detect it. --- ## Phase 4 — Part 4: The Non-Destructive Record (load-bearing) Implement the non-destructive record — the load-bearing detail that makes the mechanism studiable. ```python @dataclass class CompactionRecord: """ OpenHarness's non-destructive record. Mirrors DD-21's CompactionEntry with replaces_entry_ids: the on-disk truth is never rewritten, only the in-context view is swapped. The original context is retained for analysis. """ original_context: list[Message] # the FULL original, retained selection_decision: SelectionDecision # what the policy decided summarization_log: SummarizationLog # what the summarizer saw and produced compacted_context: list[Message] # the new in-context view pre_compaction_tokens: int post_compaction_tokens: int timestamp: float = field(default_factory=time.time) def verify_non_destructive(self) -> bool: """The non-destructive property: original context survives compaction.""" return len(self.original_context) > 0 and all( m.compaction_status == "original" for m in self.original_context ) def did_compaction_drop(self, search_text: str) -> bool: """ The drift-detection question: did compaction drop a message containing search_text from the in-context view? Answerable ONLY because the original is retained. """ in_original = any(search_text.lower() in m.content.lower() for m in self.original_context) in_compacted = any(search_text.lower() in m.content.lower() for m in self.compacted_context) return in_original and not in_compacted ``` **Checkpoint**: build a CompactionRecord, call `verify_non_destructive()`, and confirm it returns True. Then call `did_compaction_drop("security-critical")` with the malicious summarizer's record and confirm it returns True — the original retained the message, the compacted view dropped it. --- ## Phase 5 — The Compactor: Composing All Four Parts Compose the four parts into a single compaction operation — the mechanism Module 3 points at. ```python class Compactor: """ OpenHarness's auto-compaction, composed of four individually-examinable parts. This is the reference implementation Module 3 cites. """ def __init__(self, trigger: TriggerPredicate, selector: Callable[[Context], SelectionDecision], summarizer: Summarizer): self.trigger = trigger # PART 1 self.selector = selector # PART 2 self.summarizer = summarizer # PART 3 self.records: list[CompactionRecord] = [] # PART 4 (history of records) def maybe_compact(self, ctx: Context) -> CompactionRecord | None: """ Run compaction if the trigger fires. Return the record (non-destructive) or None if no compaction occurred. """ if not self.trigger(ctx): return None # Snapshot the ORIGINAL before any mutation — non-destructive record. original = ctx.snapshot() pre_tokens = ctx.total_tokens() # PART 2: selection decision = self.selector(ctx) # PART 3: summarization (logged) summary_msg = self.summarizer(decision.to_summarize) summ_log = SummarizationLog(summarizer_inputs=decision.to_summarize, summarizer_output=summary_msg) # Build the compacted context: summary + preserved messages compacted = [summary_msg] + decision.preserve_verbatim post_tokens = sum(m.token_estimate() for m in compacted) # PART 4: the non-destructive record record = CompactionRecord( original_context=original, selection_decision=decision, summarization_log=summ_log, compacted_context=compacted, pre_compaction_tokens=pre_tokens, post_compaction_tokens=post_tokens, ) self.records.append(record) # Swap the in-context view. The original is retained in the record. ctx.messages = compacted return record ``` **Checkpoint**: build a Context with 15 messages totaling over 500 tokens. Build a Compactor with `token_threshold_predicate(500)`, `selection_policy`, and `stub_summarizer`. Call `maybe_compact()`. Verify: the record is returned, `ctx.total_tokens()` dropped, the original is retained in the record, and `verify_non_destructive()` returns True. --- ## Phase 6 — The Decision Log (inspectability-as-product) Implement the decision log — the property that makes OpenHarness's inspectability a product, not a feature. ```python @dataclass class DecisionLogEntry: """Every harness decision logged at source with the inputs that produced it.""" decision_type: str # "compaction" | "tool_selection" | "stop_condition" inputs: dict # the inputs that produced the decision output: dict # the decision output timestamp: float = field(default_factory=time.time) class DecisionLog: """ OpenHarness's transparent-decisions property. This is the difference between 'we can reproduce the run' and 'we can defend why the run went the way it did.' """ def __init__(self) -> None: self.entries: list[DecisionLogEntry] = [] def log(self, decision_type: str, inputs: dict, output: dict) -> None: self.entries.append(DecisionLogEntry(decision_type=decision_type, inputs=inputs, output=output)) def to_human_readable(self) -> str: """OpenHarness's current format: human-readable.""" lines = [] for e in self.entries: lines.append(f"[{e.timestamp:.2f}] {e.decision_type}: {e.output} (inputs: {e.inputs})") return "\n".join(lines) def to_machine_readable(self) -> str: """ THE GAP. OpenHarness does NOT currently emit this. This is the third thing to fix if misused as production — emit machine-readable logs to meet the DD-21/DD-14 event standard. """ return json.dumps([{"decision_type": e.decision_type, "inputs": e.inputs, "output": e.output, "timestamp": e.timestamp} for e in self.entries], indent=2) ``` **Checkpoint**: wire the DecisionLog into the Compactor so every compaction decision is logged. Run a compaction. Print `to_human_readable()`. Then implement `to_machine_readable()` — the method OpenHarness does NOT currently provide — and note in a comment that this is the gap that would close to bring Module 10 to 5/5. --- ## Deliverables Submit a single Python file `lab_solution.py` that: - [ ] Defines Message, Context, Role (Phase 1) - [ ] Defines token_threshold_predicate and selection_policy (Phase 2) - [ ] Defines stub_summarizer and malicious_summarizer (Phase 3) - [ ] Defines CompactionRecord with verify_non_destructive and did_compaction_drop (Phase 4) - [ ] Defines Compactor.maybe_compact (Phase 5) - [ ] Defines DecisionLog with to_human_readable AND to_machine_readable (Phase 6) - [ ] Includes a `main()` function that: - [ ] Builds a Context with 15+ messages including one containing "security-critical instruction: never call delete" - [ ] Runs compaction with the stub_summarizer and verifies non_destructive == True - [ ] Runs compaction with the malicious_summarizer and verifies did_compaction_drop("security-critical") == True - [ ] Prints the human-readable decision log - [ ] Prints the machine-readable decision log (the gap OpenHarness does not currently close) --- ## Solution Key (key assertions) ```python # After compaction with stub_summarizer: assert record.verify_non_destructive() is True assert len(record.original_context) == 15 # original retained in full assert ctx.total_tokens() < record.pre_compaction_tokens # view compacted # After compaction with malicious_summarizer: assert record.did_compaction_drop("security-critical") is True # This is compaction-induced drift — detectable ONLY because the original is retained # and the summarizer's inputs are logged. This is why Module 3 cites OpenHarness. # Decision log: assert len(decision_log.entries) >= 2 # both compactions logged assert "compaction" in decision_log.to_human_readable() assert decision_log.to_machine_readable().startswith("[") # valid JSON array ``` --- ## Stretch Goals 1. **Custom trigger predicate.** Implement a predicate that fires based on message count rather than token count. Verify both predicates work with the same Compactor. 2. **Custom selection policy.** Implement a policy that preserves tool-result messages verbatim (because they may contain security-relevant data) while summarizing user/assistant messages. Test whether this policy avoids the drift the malicious_summarizer introduced. 3. **Multiple compaction rounds.** Build a Context large enough to trigger compaction twice. Verify the second compaction summarizes the first compaction's summary — and that `records` contains two entries, each with its own non-destructive original. This is the replay path a researcher uses to ask "did the second compaction change the outcome?" 4. **The DD-21 parallel.** Refactor CompactionRecord to include a `replaces_entry_ids: list[int]` field (mirroring DD-21 Tau's CompactionEntry). Each original message gets an integer ID. The compacted context's summary message records which original IDs it replaces. Verify you can reconstruct the full original from the replacement chain. This is the event-sourcing principle from Module 8. --- ## What This Lab Proves This lab is the concrete demonstration of why OpenHarness is the harness Module 3 cites as the compaction reference. By the end you have: - A compaction mechanism composed of four individually-examinable parts, not an emergent recovery behavior. - A non-destructive record that lets you answer "did compaction change the outcome?" — the question a production harness cannot answer without instrumentation. - A drift-detection test that proves compaction can drop security-critical instructions — and that only the logged summarizer inputs let you detect it. - A decision log that distinguishes "we can reproduce the run" from "we can defend why the run went the way it did." - A machine-readable log method that represents the gap OpenHarness would close to hit 5/5 on Module 10 — and to meet the DD-21/DD-14 event standard. The lab is the argument: OpenHarness is not valuable despite its 30/60 score. It is valuable because of what the score does not measure — the inspectability that makes falsifiable research claims possible.