1"""
2Filesystem Context Manager -- composable utilities for filesystem-based context engineering.
3 
4Provides three core patterns for managing agent context through the filesystem:
51. ScratchPadManager -- offload large tool outputs to files, return compact references
62. AgentPlan / PlanStep -- persist plans to disk so agents survive context window refreshes
73. ToolOutputHandler -- automatic offload-or-inline decision for tool outputs
8 
9Use when:
10    - Tool outputs exceed ~2000 tokens and would bloat the context window
11    - Agents need plan persistence across long-horizon, multi-turn tasks
12    - Building agent systems that offload intermediate results to files
13 
14Example (library usage)::
15 
16    from filesystem_context import ScratchPadManager, ToolOutputHandler
17 
18    handler = ToolOutputHandler(ScratchPadManager(base_path="scratch"))
19    result = handler.process_output("web_search", large_output_string)
20 
21Example (CLI demo)::
22 
23    python filesystem_context.py
24"""
25 
26from __future__ import annotations
27 
28import json
29import os
30import shutil
31from dataclasses import dataclass, field
32from datetime import datetime
33from pathlib import Path
34from typing import Any, Dict, List, Optional
35 
36__all__: list[str] = [
37    "ScratchPadManager",
38    "PlanStep",
39    "AgentPlan",
40    "ToolOutputHandler",
41]
42 
43 
44# =============================================================================
45# Pattern 1: Scratch Pad Manager
46# =============================================================================
47 
48 
49class ScratchPadManager:
50    """Manage temporary file storage for offloading large tool outputs.
51 
52    Use when: tool outputs exceed a token threshold and would bloat the
53    context window. Writes content to a scratch directory and returns a
54    compact reference the agent can include in context instead.
55    """
56 
57    def __init__(self, base_path: str = "scratch", token_threshold: int = 2000) -> None:
58        self.base_path: Path = Path(base_path)
59        self.base_path.mkdir(parents=True, exist_ok=True)
60        self.token_threshold: int = token_threshold
61 
62    def estimate_tokens(self, content: str) -> int:
63        """Return a rough token estimate (~4 characters per token).
64 
65        Use when: deciding whether content should be offloaded before
66        writing it to disk.
67        """
68        return len(content) // 4
69 
70    def should_offload(self, content: str) -> bool:
71        """Return True if *content* exceeds the configured token threshold.
72 
73        Use when: making an inline-vs-offload decision for a tool output.
74        """
75        return self.estimate_tokens(content) > self.token_threshold
76 
77    def offload(self, content: str, source: str) -> Dict[str, Any]:
78        """Write *content* to a timestamped scratch file and return a reference dict.
79 
80        Use when: a tool output has been determined to exceed the threshold
81        and should be persisted to disk.
82 
83        Returns a dict with keys: path, source, tokens_saved, summary.
84        """
85        timestamp: str = datetime.now().strftime("%Y%m%d_%H%M%S_%f")
86        filename: str = f"{source}_{timestamp}.txt"
87        file_path: Path = self.base_path / filename
88 
89        file_path.write_text(content)
90 
91        # Extract summary from first meaningful lines
92        lines: list[str] = content.strip().split("\n")[:5]
93        summary: str = "\n".join(lines)
94        if len(summary) > 300:
95            summary = summary[:300] + "..."
96 
97        return {
98            "path": str(file_path),
99            "source": source,
100            "tokens_saved": self.estimate_tokens(content),
101            "summary": summary,
102        }
103 
104    def format_reference(self, ref: Dict[str, Any]) -> str:
105        """Format a reference dict as a compact string for context inclusion.
106 
107        Use when: constructing the replacement message that goes into context
108        in place of the full tool output.
109        """
110        return (
111            f"[Output from {ref['source']} saved to {ref['path']}. "
112            f"~{ref['tokens_saved']} tokens. "
113            f"Summary: {ref['summary'][:200]}]"
114        )
115 
116    def cleanup(self, max_age_seconds: int = 3600) -> int:
117        """Remove scratch files older than *max_age_seconds*.
118 
119        Use when: ending a session or when the scratch directory has grown
120        large enough to slow directory listings.
121 
122        Returns the number of files removed.
123        """
124        removed: int = 0
125        now: float = datetime.now().timestamp()
126        for f in self.base_path.iterdir():
127            if f.is_file() and (now - f.stat().st_mtime) > max_age_seconds:
128                f.unlink()
129                removed += 1
130        return removed
131 
132 
133# =============================================================================
134# Pattern 2: Plan Persistence
135# =============================================================================
136 
137 
138@dataclass
139class PlanStep:
140    """Individual step in an agent plan.
141 
142    Use when: building a plan that will be persisted to disk for later
143    re-reading across context window boundaries.
144    """
145 
146    id: int
147    description: str
148    status: str = "pending"  # pending | in_progress | completed | blocked
149    notes: Optional[str] = None
150 
151 
152@dataclass
153class AgentPlan:
154    """Persistent plan that survives context window limitations.
155 
156    Use when: an agent needs to track a multi-step objective across turns
157    or context refreshes. Write the plan to disk so the agent can re-read
158    it at any point, even after summarization or context window refresh.
159    """
160 
161    objective: str
162    steps: List[PlanStep] = field(default_factory=list)
163    created_at: str = field(default_factory=lambda: datetime.now().isoformat())
164 
165    def to_dict(self) -> Dict[str, Any]:
166        """Serialize the plan to a plain dict suitable for JSON output."""
167        return {
168            "objective": self.objective,
169            "created_at": self.created_at,
170            "steps": [
171                {
172                    "id": s.id,
173                    "description": s.description,
174                    "status": s.status,
175                    "notes": s.notes,
176                }
177                for s in self.steps
178            ],
179        }
180 
181    def save(self, path: str = "scratch/current_plan.json") -> None:
182        """Persist plan to *path* as JSON.
183 
184        Use when: a plan has been created or updated and must survive a
185        potential context refresh.
186        """
187        Path(path).parent.mkdir(parents=True, exist_ok=True)
188        with open(path, "w") as f:
189            json.dump(self.to_dict(), f, indent=2)
190        print(f"Plan saved to {path}")
191 
192    @classmethod
193    def load(cls, path: str = "scratch/current_plan.json") -> AgentPlan:
194        """Load a plan from *path*.
195 
196        Use when: resuming work in a new context window or after
197        summarization -- re-read the plan to restore task awareness.
198        """
199        with open(path, "r") as f:
200            data: Dict[str, Any] = json.load(f)
201 
202        plan = cls(objective=data["objective"])
203        plan.created_at = data.get("created_at", "")
204 
205        for step_data in data.get("steps", []):
206            plan.steps.append(
207                PlanStep(
208                    id=step_data["id"],
209                    description=step_data["description"],
210                    status=step_data["status"],
211                    notes=step_data.get("notes"),
212                )
213            )
214        return plan
215 
216    def current_step(self) -> Optional[PlanStep]:
217        """Return the first non-completed step, or None if all are done.
218 
219        Use when: determining what to work on next after re-reading a plan.
220        """
221        for step in self.steps:
222            if step.status not in ("completed", "cancelled"):
223                return step
224        return None
225 
226    def complete_step(self, step_id: int, notes: Optional[str] = None) -> None:
227        """Mark step *step_id* as completed, optionally attaching *notes*.
228 
229        Use when: an agent finishes a plan step and needs to record
230        progress before persisting the updated plan.
231        """
232        for step in self.steps:
233            if step.id == step_id:
234                step.status = "completed"
235                if notes:
236                    step.notes = notes
237                return
238        raise ValueError(f"Step {step_id} not found")
239 
240    def progress_summary(self) -> str:
241        """Generate a compact progress string for context injection.
242 
243        Use when: the agent needs a one-line status to include in context
244        without re-reading the full plan.
245        """
246        completed: int = sum(1 for s in self.steps if s.status == "completed")
247        total: int = len(self.steps)
248        current: Optional[PlanStep] = self.current_step()
249 
250        summary: str = f"Objective: {self.objective}\n"
251        summary += f"Progress: {completed}/{total} steps completed\n"
252        if current:
253            summary += f"Current step: [{current.id}] {current.description}"
254        else:
255            summary += "All steps completed."
256 
257        return summary
258 
259 
260# =============================================================================
261# Pattern 3: Tool Output Handler
262# =============================================================================
263 
264 
265class ToolOutputHandler:
266    """Automatically decide whether to inline or offload tool outputs.
267 
268    Use when: building an agent loop that processes heterogeneous tool
269    outputs -- some small enough to inline, others requiring offload.
270    """
271 
272    def __init__(self, scratch_pad: Optional[ScratchPadManager] = None) -> None:
273        self.scratch_pad: ScratchPadManager = scratch_pad or ScratchPadManager()
274 
275    def process_output(self, tool_name: str, output: str) -> str:
276        """Return *output* directly if small, or a file reference if large.
277 
278        Use when: handling a tool's return value in an agent loop. Pass
279        the result into context; offloading happens transparently.
280        """
281        if self.scratch_pad.should_offload(output):
282            ref: Dict[str, Any] = self.scratch_pad.offload(output, source=tool_name)
283            return self.scratch_pad.format_reference(ref)
284        return output
285 
286 
287# =============================================================================
288# Demonstration
289# =============================================================================
290 
291 
292def _demo_scratch_pad() -> None:
293    """Demonstrate the scratch pad offloading pattern."""
294    print("=" * 60)
295    print("DEMO: Scratch Pad for Tool Output Offloading")
296    print("=" * 60)
297 
298    scratch = ScratchPadManager(base_path="demo_scratch", token_threshold=100)
299 
300    # Small output stays in context
301    small_output: str = "API returned: {'status': 'ok', 'data': [1, 2, 3]}"
302    print(f"\nSmall output ({scratch.estimate_tokens(small_output)} tokens):")
303    print(f"  Should offload: {scratch.should_offload(small_output)}")
304 
305    # Large output gets offloaded
306    large_output: str = """
307Search Results for "context engineering":
308 
3091. Context Engineering: The Art of Curating LLM Context
310   URL: https://example.com/article1
311   Snippet: Context engineering is the discipline of managing what information
312   enters the language model's context window. Unlike prompt engineering which
313   focuses on instruction crafting, context engineering addresses the holistic
314   curation of all information...
315 
3162. Building Production Agents with Effective Context Management
317   URL: https://example.com/article2
318   Snippet: Production agent systems require sophisticated context management
319   strategies. This includes compression, caching, and strategic partitioning
320   of work across sub-agents with isolated contexts...
321 
3223. The Lost-in-Middle Problem and How to Avoid It
323   URL: https://example.com/article3
324   Snippet: Research shows that language models exhibit U-shaped attention
325   patterns, with information in the middle of long contexts receiving less
326   attention than content at the beginning or end...
327 
328... (imagine 50 more results) ...
329"""
330 
331    print(f"\nLarge output ({scratch.estimate_tokens(large_output)} tokens):")
332    print(f"  Should offload: {scratch.should_offload(large_output)}")
333 
334    if scratch.should_offload(large_output):
335        ref = scratch.offload(large_output, source="web_search")
336        print(f"\nOffloaded to: {ref['path']}")
337        print(f"Tokens saved: {ref['tokens_saved']}")
338        print(f"\nReference for context:\n{scratch.format_reference(ref)}")
339 
340 
341def _demo_plan_persistence() -> None:
342    """Demonstrate the plan persistence pattern."""
343    print("\n" + "=" * 60)
344    print("DEMO: Plan Persistence for Long-Horizon Tasks")
345    print("=" * 60)
346 
347    plan = AgentPlan(objective="Refactor authentication module")
348    plan.steps = [
349        PlanStep(id=1, description="Audit current auth endpoints"),
350        PlanStep(id=2, description="Design new token validation flow"),
351        PlanStep(id=3, description="Implement changes"),
352        PlanStep(id=4, description="Write tests"),
353        PlanStep(id=5, description="Deploy and monitor"),
354    ]
355 
356    print("\nInitial plan:")
357    print(plan.progress_summary())
358 
359    plan.save("demo_scratch/current_plan.json")
360 
361    # Simulate completing first step
362    plan.complete_step(1, notes="Found 12 endpoints, 3 need updates")
363    plan.steps[1].status = "in_progress"
364 
365    print("\nAfter completing step 1:")
366    print(plan.progress_summary())
367 
368    plan.save("demo_scratch/current_plan.json")
369 
370    # Simulate loading from file (as if in new context)
371    print("\n--- Simulating context refresh ---")
372    loaded_plan = AgentPlan.load("demo_scratch/current_plan.json")
373    print("\nPlan loaded from file:")
374    print(loaded_plan.progress_summary())
375 
376 
377def _demo_tool_handler() -> None:
378    """Demonstrate the integrated tool output handler."""
379    print("\n" + "=" * 60)
380    print("DEMO: Integrated Tool Output Handler")
381    print("=" * 60)
382 
383    handler = ToolOutputHandler(
384        scratch_pad=ScratchPadManager(base_path="demo_scratch", token_threshold=50)
385    )
386 
387    outputs: list[tuple[str, str]] = [
388        ("calculator", "42"),
389        ("file_read", "Error: File not found"),
390        (
391            "database_query",
392            """
393            Results (250 rows):
394            | id | name | email | created_at | status |
395            |----|------|-------|------------|--------|
396            | 1  | John | j@e.c | 2024-01-01 | active |
397            | 2  | Jane | j@e.c | 2024-01-02 | active |
398            ... (248 more rows) ...
399        """,
400        ),
401    ]
402 
403    for tool_name, output in outputs:
404        processed: str = handler.process_output(tool_name, output)
405        print(f"\n{tool_name}:")
406        print(f"  Original length: {len(output)} chars")
407        print(f"  Processed: {processed[:100]}...")
408 
409 
410def _cleanup_demo() -> None:
411    """Remove demo files created during the demonstration."""
412    demo_path = Path("demo_scratch")
413    if demo_path.exists():
414        shutil.rmtree(demo_path)
415        print("\nDemo files cleaned up.")
416 
417 
418if __name__ == "__main__":
419    _demo_scratch_pad()
420    _demo_plan_persistence()
421    _demo_tool_handler()
422 
423    print("\n" + "=" * 60)
424    print("Cleaning up demo files...")
425    _cleanup_demo()
426