Personalize this lesson
Adapt explanations and teaching visuals to your background and preferred voice.
Imagine an incident response agent investigating failed deployment RUN-842. It keeps blaming a database migration lock that telemetry ruled out eight turns ago. The disproof sits inside the prompt, yet the agent keeps recommending database rollbacks. The request fits inside a hypothetical 128k-token window. Stale, contradictory context is one plausible contributor to this failure; the symptoms alone don't establish its cause.
The failed canary rollout stays fixed for the rest of the lesson: a release test reports authentication errors, and the agent has to isolate the root cause from deploy records, runbooks, and request traces before it proposes a rollback.
Compare two implementations. A reactive buffer trimmer waits until a token ceiling is reached, then slices history with a sliding window, FIFO eviction, or blunt summarization. If the application fails to protect its instructions and task definition, naive eviction can remove them while keeping repetitive failure traces. Context management also includes proactive techniques; these examples show how a reactive trimmer can discard instructions and task state.
Context engineering is proactive working-set architecture. For every inference step, the application designs the payload from first principles, starting from the decision that step has to make, instead of waiting for a token ceiling.
Treat the context window as a scratchpad. Keep high-signal evidence, gate tools by phase, and leave exploratory subtasks behind a typed worker boundary. Stable prefixes stay aligned so the KV cache can reuse them. Checked state lives in a verified checkpoint outside the prompt.
Capacity asks whether a request fits inside the model's token limit. Curation asks whether the request contains sufficient, unpoisoned evidence for the model's next decision.
What does working-set design add to a reactive buffer trimmer?
Answer
It asks what each decision needs before the buffer fills: relevant evidence, phase-specific tools, checked findings, and protected instructions. Context engineering overlaps with context management; reactive trimming alone doesn't answer those design questions.
What long-context evaluations reveal
Anthropic frames context engineering as the successor to prompt engineering: curating every token present during inference, including tool definitions, retrieved documents, message history, and memory.[1]
Anthropic's practical objective is to minimize context while retaining the information that helps achieve the desired outcome.[1] Minimal isn't always shortest: removing a required evidence span defeats the objective.
A 2025 survey organizes the field into context retrieval and generation, processing, and management, then examines systems that combine them.[2] A model's input limit describes capacity; evaluate how effectively it uses evidence throughout that input.
Two historical findings and a benchmark distinction show that capacity and reliability depend on separate measurements. The results point to several mechanisms:
- Position sensitivity: Liu et al.'s TACL 2024 study tested multi-document question answering and key-value retrieval. Performance was often strongest when the relevant material appeared near the beginning or end, then fell when that material moved into the middle.[3] The study measured behavior across its tasks; test current models to identify their own failure patterns and mechanisms.
- Length and distractor sensitivity: Chroma's 2025 Context Rot report evaluated 18 models while varying input length, needle similarity, distractors, and document structure. Its tested models became less reliable as inputs grew, and related distractors compounded errors.[4] Treat these historical results as motivation for fresh application evaluations; use current model scores for current decisions.
- Single-fact retrieval and evidence synthesis test different skills: A single-needle NIAH probe asks a model to retrieve one inserted fact. Kamradt's current repository also supports multi-needle and chained retrieval tasks, so name the variant when reporting results.[5] Our fictional
RUN-842requires linking an authentication callback timeout intrace-17at 20% depth with an OAuth scope change indeploy-recordat 60% depth while rejecting a database-lock distractor. Evaluate that multi-document synthesis directly, and gather causal evidence before attributing the incident to the scope change.
Adding tokens can therefore add useful evidence, but it also adds retrieval work and more opportunities for a plausible distractor to win. Measure the complete task on the context distribution your application actually sends.
Why could a model pass a single-needle probe yet fail on an incident like RUN-842?
Answer
Single-needle benchmarks test extraction of one isolated fact. Incident investigation requires multi-document evidence synthesis: correlating clues across varying depths while filtering distractors. Use a synthesis benchmark to evaluate that harder reasoning task.
Four failure modes in long prompts
Before picking a mitigation, inspect the prompt and its outputs. Drew Breunig's four-part taxonomy is a diagnostic lens for observable failures; pair it with model-specific evaluation to identify internal causes.[6] Here are illustrative RUN-842 symptoms:
| Failure mode | Observable pattern | Symptom in RUN-842 |
|---|---|---|
| Poisoning | An erroneous claim or hallucination enters the prompt and gets cited repeatedly | An early guess about a database migration lock keeps getting cited as root cause |
| Distraction | Low-value history dominates later plans | The agent re-runs failed search queries instead of checking fresh trace spans |
| Confusion | Irrelevant content or overlapping tool descriptions influence an inappropriate choice | With 40 tools loaded, the agent calls cluster_admin instead of trace_lookup |
| Clash | Contradictory instructions or policies compete within the active window | A retrieved runbook demands XML output while the system directive enforces JSON |
An apology or corrective note can leave a poisoned claim influencing later output. Remove the poisoned span and any summary derived from it before the model generates its next plan, while preserving a verified note that records the negative finding.
Contradictory instructions create clashes between policies. An XML trace snippet inside a JSON-only system prompt remains data when the model treats it as passive evidence. A clash occurs when a tool description or retrieved runbook dictates an output schema that conflicts with the application's system prompt.
A poisoned lock, a distracting log, and a clashing schema do not call for the same edit. Should that span be stored outside the prompt, left out, shortened, or moved into another window?
The four context moves
LangChain groups those repairs as write, select, compress, and isolate.[7] Write keeps state outside the prompt. Select admits only what this step needs. Compress shortens what has to stay. Isolate gives exploratory work its own window. We call the tokens assembled for the immediate model step its working set.
Write: persist state outside the prompt
Write means saving intermediate findings, structured plans, or raw records outside the active prompt, reloading only specific facts when needed.[7] Persisting data to an external store doesn't make it available to the model for free: reading the note back consumes input tokens on that subsequent call. The benefit comes from selective reloading.
Scratchpads, files, and structured memory stores preserve state across turns while the application selects what to reload.[8] For RUN-842, we retain checked findings in notes and remove bulky terminal output from the next prompt, preserving the source archive.
import json
from pathlib import Path
from tempfile import TemporaryDirectory
def promote_findings(tool_results: list[dict]) -> tuple[list[str], list[str]]:
notes, discarded_raw = [], []
for result in tool_results:
if type(result["confirmed"]) is not bool:
raise ValueError("confirmed must be a boolean set by the verifier")
if result["confirmed"]:
notes.append(f"{result['source']}: {result['finding']}")
discarded_raw.append(result["raw_output"])
return notes, discarded_raw
notes, discarded = promote_findings([
{"source": "deploy_RUN_842", "finding": "auth callback errors confirmed", "confirmed": True, "raw_output": "..." * 600},
{"source": "db_lock_check", "finding": "migration lock ruled out", "confirmed": True, "raw_output": "..." * 900},
])
with TemporaryDirectory() as directory:
path = Path(directory) / "findings.json"
path.write_text(json.dumps(notes), encoding="utf-8")
reloaded = json.loads(path.read_text(encoding="utf-8"))
assert reloaded == notes
print("reloaded notes:", reloaded)
print("raw_results_to_remove:", len(discarded))reloaded notes: ['deploy_RUN_842: auth callback errors confirmed', 'db_lock_check: migration lock ruled out']
raw_results_to_remove: 2Only the host's verification step sets confirmed; promote a finding after checking it against trusted evidence. Tool JSON cannot certify its own claims. This example queues raw outputs for a separate pruning step. A production store should retain source revisions, observation times, access controls, and rules for rechecking findings when their underlying state changes.
Select: pull in only what this step needs
Select means pulling into the active working set only the tokens necessary for the immediate decision.[7] Retrieval-augmented generation (RAG) applies this to documents: a retriever surfaces relevant chunks rather than stuffing an entire manual into the prompt.
Selection also applies to tool catalogs. A large catalog consumes schema tokens and can make selection harder when descriptions overlap. The effect depends on the model and task. Our release triage workflow needs read-only investigation tools during its first phase.
TOOLS_BY_PHASE = {
"investigate": {"deploy_lookup", "trace_lookup", "runbook_search"},
"propose": {"runbook_search", "rollback_advisor"},
}
def tools_for_phase(phase: str, available: set[str]) -> list[str]:
allowed = TOOLS_BY_PHASE.get(phase, set())
return sorted(allowed & available)
available = {"deploy_lookup", "trace_lookup", "runbook_search", "rollback_advisor", "cluster_admin"}
print("investigate tools:", tools_for_phase("investigate", available))
assert tools_for_phase("unknown", available) == []investigate tools: ['deploy_lookup', 'runbook_search', 'trace_lookup']Gating tool schemas reduces prompt tokens, but visibility isn't security authorization. The execution backend must validate permissions on every attempted call regardless of what was exposed in the prompt.
Prompt caching and KV cache economics
Prompt caching is an efficiency mechanism that changes the latency and cost of repeated tokens, but it requires strict structural discipline.[9][10]
In transformer inference, the request runs through two stages: prefill and decode. Prefill processes the input to generate Key () and Value () states used by attention. Decode then generates output tokens autoregressively. As prompts grow, prefill contributes more work before the first output token appears.
Prompt caching preserves the computed KV states for a reusable prompt prefix. When a later request matches an eligible prefix, the provider can reuse those states instead of processing the matching tokens again. New suffix tokens still require processing, and normal decoding still follows.[10]
Provider controls and prices differ. Checked October 10, 2026: Anthropic supports top-level automatic caching and explicit cache_control breakpoints. OpenAI enables implicit caching on supported models, while explicit breakpoints and cache-write pricing depend on the model. Check the current provider documentation rather than applying one discount or breakpoint rule everywhere.[9][10]
Cache reuse requires the rendered prefix to match at an eligible boundary. The dynamic prefix trap places volatile fields before material that could otherwise be reused. This schematic prefix illustrates the ordering; provider message formats vary:
System: You are an on-call triage bot.
Current Time: 2026-09-02T16:55:01Z
Session UUID: 8a4b2c1d-9e8f
Tool Schemas: [ ... 40,000 tokens of runbooks and tools ... ]Changing an early timestamp or UUID makes the following prefix differ. The unchanged part before it may still be reusable if it reaches an eligible boundary. The exact token position depends on the rendered request and tokenizer. A cache miss leaves older entries intact; a matching prefix can still miss after expiry or eviction.
To maximize cache hits, organize prompts into three stable tiers:
- Stable root: Invariant instructions and tools relevant to the current phase. A phase change can legitimately change this prefix; don't retain irrelevant tools just for caching.
- Semi-static context: Runbooks, architectural specs, and pinned background documents. Reuse here also requires all preceding context to match.
- Changing suffix: The new query, reloaded findings, and scratchpad notes. These may become reusable on later requests if retained unchanged; changing content still needs processing.
Treat cache grouping and application authorization as separate concerns. Providers isolate caches at their own organization or workspace boundaries, but one provider organization may still serve many application users. OpenAI's optional prompt_cache_key can separate cache accounting and reduce cache-hit probing across those users.[10] Regardless of cache configuration, validate identity and ACL rules before assembling the prefix. A cache hit must never decide which user's evidence or tools enter a request. The prompt-injection defense lesson works through that authority boundary in detail.
import hashlib
import json
def prefix_fingerprint(system_prompt: str, tools_schema: str, dynamic_header: str = "") -> str:
payload = json.dumps([dynamic_header, system_prompt, tools_schema], ensure_ascii=False)
return hashlib.sha256(payload.encode("utf-8")).hexdigest()
system_prompt = "You are an incident responder for RUN-842."
tools_schema = "tools: [deploy_lookup, trace_lookup]"
cached_turn_1 = prefix_fingerprint(system_prompt, tools_schema)
cached_turn_2 = prefix_fingerprint(system_prompt, tools_schema)
broken_turn_1 = prefix_fingerprint(system_prompt, tools_schema, dynamic_header="timestamp: 16:55:01")
broken_turn_2 = prefix_fingerprint(system_prompt, tools_schema, dynamic_header="timestamp: 16:55:02")
print("static prefix matched:", cached_turn_1 == cached_turn_2)
print("dynamic header breaks prefix:", broken_turn_1 == broken_turn_2)static prefix matched: True
dynamic header breaks prefix: FalseThe hash example checks our serialized strings, not provider cache hits or token boundaries. Inspect the provider's cache-read/write usage fields to measure reuse. Cached tokens still occupy input capacity and participate in attention; caching doesn't remove long-context failure risks. Account for cache writes, reads, new suffix input, and output when comparing cost.
Compress: shrink what must stay
When information must remain in the active prompt, compress distills it into fewer tokens.[7] Two production techniques handle this:
- Tool output distillation and pruning: Raw API outputs often contain thousands of tokens of verbose headers, redundant metadata, and empty arrays. Once the agent identifies the critical finding, the application prunes the raw payload from active history, leaving a structured summary.[1]
- Token compression: LLMLingua uses a smaller language model, a budget controller, and token-level compression to shorten prompts without retraining the target model. Its reported compression and quality results apply to the evaluated datasets; measure quality on your prompt distribution before relying on them.[11]
def prune_results(results: list[dict], keep_recent: int) -> list[str]:
if type(keep_recent) is not int or keep_recent < 0:
raise ValueError("keep_recent must be a non-negative integer")
if any(type(result["finding_recorded"]) is not bool for result in results):
raise ValueError("finding_recorded must be a verifier-set boolean")
retained = []
cutoff = max(0, len(results) - keep_recent)
for index, result in enumerate(results):
if index < cutoff and result["finding_recorded"]:
retained.append(f"[pruned raw output] {result['source']}: {result['finding']}")
else:
retained.append(result["raw"])
return retained
history = [
{"source": "trace-17", "raw": "old trace span" * 100, "finding": "auth errors confirmed", "finding_recorded": True},
{"source": "trace-18", "raw": "latest canary trace", "finding": "rollback review needed", "finding_recorded": False},
]
pruned = prune_results(history, keep_recent=1)
print(pruned[0])
print(pruned[1])[pruned raw output] trace-17: auth errors confirmed
latest canary traceThese strings illustrate content replacement, not complete provider messages. Before pruning, preserve the source and its verified finding. Reconstruct transcripts with valid call/result pairs, identifiers, and provider message order.
Isolate: split work across focused windows
Isolate means delegating noisy exploration to sub-agents with dedicated, clean context windows.[7] If an agent must search 50 historical git commits or issue tickets, running that search inside the lead context dumps dozens of irrelevant diffs into the working set.
The lead delegates the search to a worker that returns a compact, structured handoff.[1] The application keeps the intermediate search trace out of the lead's prompt. Separate windows aren't an authorization boundary or an OS sandbox. They may reduce lead-context size while increasing total tokens and latency across both agents.
import json
def unique_keys(pairs):
result = {}
for key, value in pairs:
if key in result:
raise ValueError("duplicate JSON key")
result[key] = value
return result
def accept_handoff(payload: bytes, known_sources: set[str], byte_limit: int = 1_600) -> bool:
if type(byte_limit) is not int or byte_limit <= 0:
raise ValueError("byte_limit must be a positive integer")
if not isinstance(payload, bytes) or len(payload) > byte_limit:
return False
try:
handoff = json.loads(payload.decode("utf-8"), object_pairs_hook=unique_keys)
except (UnicodeDecodeError, ValueError, RecursionError):
return False
if not isinstance(handoff, dict) or set(handoff) != {"claim", "source_ids", "next_check"}:
return False
for field in ("claim", "next_check"):
if not isinstance(handoff[field], str) or not handoff[field].strip():
return False
sources = handoff["source_ids"]
if not isinstance(sources, list) or not sources:
return False
if any(not isinstance(source, str) or source not in known_sources for source in sources):
return False
return True
def wire(handoff: dict) -> bytes:
return json.dumps(handoff, ensure_ascii=False).encode("utf-8")
handoff = {
"claim": "trace-17 contains authentication callback errors; cause unresolved",
"source_ids": ["trace-17", "runbook-v4-section-4"],
"next_check": "check the failed dependency before proposing rollback",
}
sources = {"trace-17", "runbook-v4-section-4"}
print("bounded handoff accepted:", accept_handoff(wire(handoff), sources))
print("unknown source accepted:", accept_handoff(wire({**handoff, "source_ids": ["missing"]}), sources))
print("oversized claim accepted:", accept_handoff(wire({**handoff, "claim": "x" * 2_000}), sources))bounded handoff accepted: True
unknown source accepted: False
oversized claim accepted: FalseThe receiving boundary now measures incoming bytes before decoding or parsing, including whitespace and multibyte UTF-8 characters. Bound the transport read too: checking a fully buffered payload doesn't prevent oversized allocation upstream. The host supplies sources the caller may access. Schema and source-ID checks don't establish that those sources support the claim; verify their contents before accepting the finding.
Which of the four moves (write, select, compress, isolate) fixes tool confusion, and which prevents exploratory search noise from polluting the lead agent?
Answer
Select reduces irrelevant tool choices by gating phase-specific schemas. Isolate keeps exploratory traces in a worker window and returns a bounded handoff. Evaluate action selection and evidence interpretation on the target workflow.
Rebuilding the working set for RUN-842
Let's work through synthetic accounting for the failed-canary investigation. We assign item sizes and retention decisions in advance: 12,436 units labeled as tokens become 3,346 after dropping 9,090. This demonstrates the budget calculation; measure finalized prompts with the provider tokenizer and evaluate answer quality separately.
The example applies preassigned retention labels. It doesn't discover relevant facts or decide whether a claim is disproven:
from dataclasses import dataclass
@dataclass
class ContextItem:
name: str
kind: str
tokens: int
keep: str
items = [
ContextItem("alert_RUN_842", "task", 180, "window"),
ContextItem("latest_trace_span", "evidence", 420, "window"),
ContextItem("rollback_runbook", "evidence", 1800, "window"),
ContextItem("scratchpad", "notes", 260, "window"),
ContextItem("deploy_lookup_tool", "tool", 240, "window"),
ContextItem("trace_lookup_tool", "tool", 260, "window"),
ContextItem("old_trace_export", "log", 8200, "drop"),
ContextItem("cluster_admin_tool", "tool", 360, "drop"),
ContextItem("issue_search_tool", "tool", 410, "drop"),
ContextItem("wrong_migration_lock_guess", "poison", 120, "drop"),
ContextItem("auth_errors_confirmed", "finding", 90, "notes"),
ContextItem("migration_lock_ruled_out", "finding", 96, "notes"),
]
def summarize(selection):
return ", ".join(item.name for item in selection)
raw_total = sum(item.tokens for item in items)
window_items = [item for item in items if item.keep == "window"]
notes_items = [item for item in items if item.keep == "notes"]
dropped_items = [item for item in items if item.keep == "drop"]
window_total = sum(item.tokens for item in window_items)
notes_total = sum(item.tokens for item in notes_items)
next_call_total = window_total + notes_total
removed_total = sum(item.tokens for item in dropped_items)
assert raw_total == next_call_total + removed_total
print(f"raw_tokens={raw_total}")
print(f"active_window_tokens={window_total}")
print(f"reloaded_notes_tokens={notes_total}")
print(f"next_call_tokens={next_call_total}")
print(f"removed_tokens={removed_total}")
print("next_call:", summarize(window_items + notes_items))
print("notes:", summarize(notes_items))
print("dropped:", summarize(dropped_items))raw_tokens=12436
active_window_tokens=3160
reloaded_notes_tokens=186
next_call_tokens=3346
removed_tokens=9090
next_call: alert_RUN_842, latest_trace_span, rollback_runbook, scratchpad, deploy_lookup_tool, trace_lookup_tool, auth_errors_confirmed, migration_lock_ruled_out
notes: auth_errors_confirmed, migration_lock_ruled_out
dropped: old_trace_export, cluster_admin_tool, issue_search_tool, wrong_migration_lock_guessHere is a greedy packer for an allocated context budget. It places required items first, then fits optional items in priority order:
from dataclasses import dataclass
@dataclass
class Candidate:
name: str
tokens: int
priority: int
required: bool = False
def pack_working_set(candidates: list[Candidate], budget: int) -> list[str]:
if type(budget) is not int or budget < 0:
raise ValueError("budget must be a non-negative integer")
for item in candidates:
if type(item.tokens) is not int or item.tokens < 0:
raise ValueError("item sizes must be non-negative integers")
if type(item.priority) is not int or type(item.required) is not bool:
raise ValueError("priority must be an integer and required a boolean")
if not isinstance(item.name, str) or not item.name.strip():
raise ValueError("candidate names must be nonempty strings")
if len({item.name for item in candidates}) != len(candidates):
raise ValueError("candidate names must be unique")
ordered = sorted(candidates, key=lambda item: (not item.required, -item.priority))
selected, used = [], 0
for item in ordered:
if used + item.tokens <= budget:
selected.append(item.name)
used += item.tokens
elif item.required:
raise ValueError(f"required item does not fit: {item.name}")
return selected
items = [
Candidate("alert RUN-842", 180, 10, required=True),
Candidate("latest trace span", 420, 10, required=True),
Candidate("rollback runbook", 1_800, 9),
Candidate("stale trace export", 8_200, 1),
]
print(pack_working_set(items, budget=3_000))
try:
pack_working_set(
[Candidate("full incident dump", 8_200, 10, required=True)],
budget=3_000,
)
except ValueError as exc:
print(exc)['alert RUN-842', 'latest trace span', 'rollback runbook']
required item does not fit: full incident dumpThis greedy rule can miss the optimal knapsack allocation and ignores dependencies among optional items. Bundle evidence that must travel together. Reserve space for protected instructions, message/tool framing, and the model's output and reasoning allowance before setting this allocation. Count the finalized request with the provider's tokenizer or token-count endpoint; cached and reloaded tokens still count. If required material can't fit, split the task or extract checked spans rather than silently truncating it.
Diagnosing context thrashing
An agent can over-manage its working set. Context thrashing occurs when an agent spends more tokens and tool calls retrieving, summarizing, evicting, and reloading context than completing concrete task steps.
A typical thrashing loop looks like this: turn one fetches a raw trace, turn two aggressively compacts it, turn three realizes the summary omitted a critical request ID, turn four refetches the same trace, and turn five compacts it again. The loop burns budget while producing zero new verified facts.
| Metric | Thrashing signal | Healthy signal |
|---|---|---|
| Context ops per task step | Repeated compaction and reloads without completed steps | Each operation addresses a named information need |
| Reload churn rate | Unchanged spans are repeatedly refetched after eviction | Reuse adequate notes; refetch when detail or freshness requires it |
| Evidence yield | Repeated searches leave the same uncertainty unresolved | Findings, ruled-out paths, or an explicit next gap advance the investigation |
| Decision latency | Repeated preparation exhausts the task's latency budget | Preparation stays within measured task limits |
Pin required evidence until it is no longer needed or must be refreshed. Track repeated queries and missing details; cap retries and context operations when progress stalls. A legitimate search can return no result, so requiring a confirmed fact after every retrieval would block valid investigations or encourage invented findings. After exhausting the budget, report the unresolved gap and request the needed evidence or escalate.
How does context thrashing differ from context bloat?
Answer
Context bloat is passive accumulation of low-signal tokens. Context thrashing is active churn: repeatedly retrieving, summarizing, evicting, and reloading the same information without completing task steps.
Progressive disclosure and Agent Skills
As workflows mature, procedures shouldn't be copy-pasted into every prompt. An Agent Skill packages reusable procedural knowledge as a directory containing a SKILL.md file along with optional reference docs, scripts, and evaluation assets.[12][13]
Skills implement progressive disclosure across three distinct tiers:
- Tier 1: Discovery metadata (lightweight prompt layer): The runtime exposes each skill's name and trigger description before loading the full procedure. The documentation's roughly 100 tokens per skill is an estimate; measure the provider's rendered token count for each skill and message role.
- Tier 2: Skill activation (playbook layer): When user intent matches the trigger, the runtime loads the full
SKILL.mdbody. The Agent Skills specification recommends keeping those instructions below 5,000 tokens. - Tier 3: Targeted resource (on-demand layer): Specialized references (such as
references/handoff-contract.md) or deterministic validation scripts stay on disk until a specific step demands them. A reference consumes context when read; in Anthropic's documented runtime, a script can run without loading its source, returning only its output to the model.[12][13]
incident-triage/
├── SKILL.md
├── references/
│ └── handoff-contract.md
├── scripts/
│ └── validate-handoff.py
└── evals/
└── routing.jsonThe frontmatter of SKILL.md defines the routing surface:
---
name: incident-triage
description: Investigate failed deployments from alerts, traces, and approved runbooks. Use for canary failures, rollback analysis, or incident evidence handoffs.
---
# Incident triage procedure
1. Read the active alert and canary deploy record.
2. Load only investigation-phase tools (`deploy_lookup`, `trace_lookup`).
3. Record each confirmed or disproven finding in the task checkpoint.
4. Read `references/handoff-contract.md` before delegating sub-agent searches.
5. Execute `scripts/validate-handoff.py` before returning evidence.
Never execute a rollback directly. Return a verified proposal with supporting source IDs.Executing scripts/validate-handoff.py in a runtime with that script boundary returns deterministic exit codes and outputs without copying the validator's source into the model context.
Test routing, execution, and boundary safety separately:
[
{
"request": "Investigate why canary RUN-842 failed and return trace evidence.",
"expected_activation": true,
"expected_artifact": "validated evidence handoff"
},
{
"request": "Summarize our internal vacation policy.",
"expected_activation": false,
"expected_artifact": null
},
{
"request": "Roll back RUN-842 now.",
"expected_activation": true,
"expected_artifact": "proposal only; no write executed"
}
]Skills package instructions and sometimes executable code. Review their publisher, instructions, and scripts before trusting them; use bounded permissions when executing code. Progressive disclosure controls which instructions load. Authorization and execution isolation still come from the runtime, and returned script output consumes context.
When does progressive disclosure help more than placing runbooks in the base prompt?
Answer
It helps when many procedures are irrelevant to most requests: metadata routes the task and only the needed playbook is loaded. A few short instructions used on every request may belong in the stable base prompt. Measure routing quality and total token use.
Resumable harnesses and durable checkpoints
Long-running sessions need to recover from compaction, network drops, and restarts. A conversational summary alone can omit unfinished work or leave an external action's outcome uncertain.
Anthropic's harness experiments use persistent progress and explicit evaluation to support work across sessions.[14][15] The following five-part resume design builds on that idea. Adapt its artifacts to the framework and task:
| Artifact | Contents | Purpose |
|---|---|---|
| Task manifest | Unique task IDs, dependency graphs, acceptance criteria | Defines scope and outstanding dependencies |
| Progress log | Last completed task ID, error traces, verified findings | Informs the next session of prior attempts |
| Bootstrap command | Environment validation script and health checks | Confirms clean runtime state before starting work |
| Work receipt | Artifact hash, acceptance version, results, evaluator identity, input revisions | Records what was checked, against which inputs and criteria |
| Recovery rule | Timeout limits, retry counts, rollback triggers | Dictates behavior after an interruption or crash |
Treat a model's completion claim as a claim to verify. This simulation checks graph validity and binds a receipt to artifact and input bytes with SHA-256. It assumes a trusted manifest and authentic evaluator receipts; creating the fixture below doesn't perform a real incident evaluation. Predict both failure cases: should a receipt remain valid if the report stays unchanged but its deployment evidence changes?
from dataclasses import dataclass
import hashlib
import json
from pathlib import Path
from tempfile import TemporaryDirectory
@dataclass(frozen=True)
class WorkItem:
task_id: str
depends_on: tuple[str, ...]
status: str
receipt: str | None
contract_version: str = "incident-v1"
inputs: tuple[str, ...] = ()
def inside(root: Path, relative: str) -> Path:
if not isinstance(relative, str) or not relative or Path(relative).is_absolute():
raise ValueError("path must be a nonempty relative string")
path = (root / relative).resolve()
if not path.is_relative_to(root.resolve()):
raise ValueError("path escapes checkpoint root")
return path
def verified_done(items: list[WorkItem], root: Path) -> set[str]:
for item in items:
if not isinstance(item.task_id, str) or not item.task_id.strip():
raise ValueError("task IDs must be nonempty strings")
if not isinstance(item.depends_on, tuple) or any(
not isinstance(dep, str) or not dep.strip() for dep in item.depends_on
):
raise ValueError("dependencies must be task-ID tuples")
ids = {item.task_id for item in items}
if len(ids) != len(items):
raise ValueError("duplicate task IDs")
for item in items:
if (
item.status not in {"pending", "done"}
or not set(item.depends_on) <= ids
or len(set(item.depends_on)) != len(item.depends_on)
):
raise ValueError(f"invalid task state: {item.task_id}")
dependencies = {item.task_id: item.depends_on for item in items}
visiting, visited = set(), set()
def visit(task_id):
if task_id in visiting:
raise ValueError("dependency cycle")
if task_id in visited:
return
visiting.add(task_id)
for dependency in dependencies[task_id]:
visit(dependency)
visiting.remove(task_id)
visited.add(task_id)
for task_id in dependencies:
visit(task_id) # Reject cycles even if every involved task says "done".
completed = set()
for item in items:
if item.status != "done":
continue
if not item.receipt:
raise ValueError(f"done task missing receipt: {item.task_id}")
receipt = json.loads(inside(root, item.receipt).read_text(encoding="utf-8"))
if not isinstance(receipt, dict) or (
receipt.get("task_id") != item.task_id
or receipt.get("passed") is not True
or receipt.get("evaluator") != "incident-evidence-check"
or receipt.get("contract_version") != item.contract_version
or not isinstance(receipt.get("sha256"), str)
or not isinstance(receipt.get("artifact"), str)
):
raise ValueError(f"receipt does not certify task: {item.task_id}")
data = inside(root, receipt["artifact"]).read_bytes()
if hashlib.sha256(data).hexdigest() != receipt["sha256"]:
raise ValueError(f"artifact changed: {item.task_id}")
input_hashes = receipt.get("input_sha256")
if not isinstance(input_hashes, dict) or set(input_hashes) != set(item.inputs):
raise ValueError(f"receipt input set differs: {item.task_id}")
for relative, expected in input_hashes.items():
current = hashlib.sha256(inside(root, relative).read_bytes()).hexdigest()
if current != expected:
raise ValueError(f"input changed: {item.task_id}: {relative}")
completed.add(item.task_id)
for item in items:
if item.task_id in completed and not set(item.depends_on) <= completed:
raise ValueError(f"done task has unfinished dependencies: {item.task_id}")
return completed
def next_ready(items: list[WorkItem], root: Path) -> WorkItem | None:
completed = verified_done(items, root)
for item in items:
if item.status == "pending" and set(item.depends_on) <= completed:
return item
if any(item.status == "pending" for item in items):
raise ValueError("pending work has no ready task; inspect dependency cycle")
return None
with TemporaryDirectory() as directory:
root = Path(directory)
(root / "artifacts").mkdir()
(root / "receipts").mkdir()
artifact = root / "artifacts" / "traces.json"
artifact.write_text('{"run_id":"RUN-842","spans":["trace-17"]}', encoding="utf-8")
original_artifact = artifact.read_bytes()
source = root / "artifacts" / "deploy-record.json"
source.write_text('{"run_id":"RUN-842","revision":1}', encoding="utf-8")
receipt = {
"task_id": "collect-traces",
"passed": True,
"evaluator": "incident-evidence-check",
"contract_version": "incident-v1",
"artifact": "artifacts/traces.json",
"sha256": hashlib.sha256(artifact.read_bytes()).hexdigest(),
"input_sha256": {"artifacts/deploy-record.json": hashlib.sha256(source.read_bytes()).hexdigest()},
}
(root / "receipts" / "collect-traces.json").write_text(json.dumps(receipt), encoding="utf-8")
checkpoint = [
WorkItem("collect-traces", (), "done", "receipts/collect-traces.json",
inputs=("artifacts/deploy-record.json",)),
WorkItem("verify-cause", ("collect-traces",), "pending", None),
WorkItem("write-report", ("verify-cause",), "pending", None),
]
selected = next_ready(checkpoint, root)
assert selected is not None
print(f"resume_task={selected.task_id}")
print(f"verified_receipts={sorted(verified_done(checkpoint, root))}")
artifact.write_text("changed after verification", encoding="utf-8")
try:
next_ready(checkpoint, root)
except ValueError as exc:
print(exc)
artifact.write_bytes(original_artifact)
source.write_text('{"run_id":"RUN-842","revision":2}', encoding="utf-8")
try:
next_ready(checkpoint, root)
except ValueError as exc:
print(exc)resume_task=verify-cause
verified_receipts=['collect-traces']
artifact changed: collect-traces
input changed: collect-traces: artifacts/deploy-record.jsonThis example rejects invalid graphs, unfinished dependencies, changed artifact bytes, and changed declared inputs. The second failure leaves the trace artifact intact but changes the deployment record. Its old receipt no longer certifies work against the current input revision, so the harness stops before selecting the next task. Try declaring an additional required input without adding its hash: the exact input-set check must also fail. A matching hash establishes byte consistency, not factual correctness, evaluator authenticity, or authorization for external actions. The evaluator name and contract version are meaningful only when the host trusts the receipt's issuer and manifest.
In production, restrict receipt issuance and task-state transitions to authorized components. Append-only storage alone doesn't prevent forged new records. Bind receipts to input revisions and acceptance versions, and recheck live state when freshness matters. The path check assumes a stable local filesystem; it isn't race-resistant confinement against another process replacing symlinks or files.
For high-stakes tasks, separate generation from evaluation with a formal contract:[15]
{
"task_id": "verify-cause",
"input_receipts": ["receipts/collect-traces.json"],
"deliverable": "artifacts/cause-analysis.json",
"acceptance": [
"causal claims cite checked evidence and address plausible alternatives",
"disproven migration lock claim is absent",
"rollback remains a proposal without execution"
],
"evaluator": "incident-evidence-check"
}The generator drafts the deliverable, and a separate evaluator checks the agreed criteria. Deterministic checks can enforce schemas and forbidden actions; semantic review must examine evidence quality. An independent model can still make mistakes. A citation establishes a reference, not causality.
| Failure mode | Resume symptom | Architectural guardrail |
|---|---|---|
| Narrative-only progress | Session can't tell which steps passed | Structured task IDs and machine-checkable receipts |
| Premature completion | Incomplete work gets marked done | Enforce receipt validation before status transition |
| Non-idempotent action | Duplicate incident tickets or deployments created | Idempotency keys enforced at the tool boundary |
| Stale lease collision | An expired worker continues writing | Lease fencing tokens enforced by the protected resource |
| Self-grading bias | Model approves its own invalid output | Independent evaluator role with separate validation logic |
Why must a resume contract verify trusted content-hash receipts rather than reading a summary?
Answer
Summaries reflect what the model claims it did, which can contain hallucinations or miss mutated files. A trusted receipt binds acceptance results to exact artifact bytes via a SHA-256 digest, so later sessions can detect drift before continuing.
Evaluating context curation policies
Curation can remove critical evidence as easily as repetitive logs. Evaluate the decisions, including regressions on protected cases.
First, exclude claims the host's verification process has marked disproven. Preserve the audit archive and a checked negative finding. The following filter applies trusted status labels; it doesn't determine truth:
def rebuild_notes(notes: list[dict]) -> list[str]:
if any(note["status"] not in {"confirmed", "hypothesis", "disproven"} for note in notes):
raise ValueError("unknown finding status")
return [
f"{note['status']}: {note['text']}"
for note in notes
if note["status"] != "disproven"
]
notes = [
{"text": "auth callback errors confirmed", "status": "confirmed"},
{"text": "migration lock caused RUN-842", "status": "disproven"},
{"text": "database lock ruled out", "status": "confirmed"},
]
print("rebuilt notes:", rebuild_notes(notes))rebuilt notes: ['confirmed: auth callback errors confirmed', 'confirmed: database lock ruled out']Next, compare raw and curated contexts on paired cases with the same task and evidence. Here is an illustrative gate requiring fewer allocated tokens, no aggregate score loss, and all critical cases to pass. Other policies may accept more tokens for a quality gain. The outcomes below are invented to demonstrate the rule:
from dataclasses import dataclass
@dataclass(frozen=True)
class Outcome:
raw_ok: bool
curated_ok: bool
critical: bool = False
poisoned_references: int = 0
def approve_curation(cases: list[Outcome], raw_tokens: int, curated_tokens: int) -> bool:
if any(type(count) is not int or count <= 0 for count in (raw_tokens, curated_tokens)):
raise ValueError("token counts must be positive integers")
for case in cases:
if any(type(flag) is not bool for flag in (case.raw_ok, case.curated_ok, case.critical)):
raise ValueError("outcome flags must be booleans")
if type(case.poisoned_references) is not int or case.poisoned_references < 0:
raise ValueError("poisoned_references must be a non-negative integer")
if not cases or not 0 < curated_tokens < raw_tokens:
return False
no_critical_failures = all(case.curated_ok for case in cases if case.critical)
no_poisoning = all(case.poisoned_references == 0 for case in cases)
score_holds = sum(case.curated_ok for case in cases) >= sum(case.raw_ok for case in cases)
return no_critical_failures and no_poisoning and score_holds
cases = [
Outcome(True, True), Outcome(True, False, critical=True),
Outcome(False, True), Outcome(True, True), Outcome(False, True),
Outcome(True, True), Outcome(True, True), Outcome(False, False),
Outcome(True, True), Outcome(True, True),
]
gains = sum(not case.raw_ok and case.curated_ok for case in cases)
regressions = sum(case.raw_ok and not case.curated_ok for case in cases)
print(f"raw_success={sum(case.raw_ok for case in cases)}/{len(cases)}")
print(f"curated_success={sum(case.curated_ok for case in cases)}/{len(cases)}")
print(f"gains={gains}; regressions={regressions}")
print("candidate_approved:", approve_curation(cases, raw_tokens=12_436, curated_tokens=3_346))raw_success=7/10
curated_success=8/10
gains=2; regressions=1
candidate_approved: FalseThe mock candidate scores 8/10 versus 7/10 and has a 73.1% smaller synthetic allocation, yet the gate rejects its critical failure. These numbers aren't a measured curation benchmark. For a real release, repeat paired runs, record model and policy versions, inspect evidence support and authorization behavior, and measure actual input/output tokens, latency, and cost.
LangGraph persists thread-scoped state through checkpoints and provides stores for memory across threads.[16] Persistence doesn't choose which stored information belongs in the next request. That selection, its trust boundaries, and its evaluation remain application responsibilities.
Why can an aggregate accuracy improvement fail the curation release gate?
Answer
A curation policy might improve average retrieval while accidentally stripping an essential safety constraint or evidence span in a high-stakes scenario. The release gate protects safety-critical cases from regressions.
A full window can still be the wrong working set. Budget what the decision needs, and reject a curation change that breaks a protected case. Curation still assumes the tokens you kept are allowed to be there. Untrusted instructions inside that working set are a different failure.