1# Filesystem Context Implementation Patterns
2 
3This reference provides detailed implementation patterns for filesystem-based context engineering.
4 
5## Pattern Catalog
6 
7### 1. Scratch Pad Manager
8 
9A centralized manager for handling large tool outputs and intermediate results.
10 
11```python
12import os
13import json
14from datetime import datetime
15from pathlib import Path
16 
17class ScratchPadManager:
18    """Manages temporary file storage for agent context offloading."""
19    
20    def __init__(self, base_path: str = "scratch", token_threshold: int = 2000):
21        self.base_path = Path(base_path)
22        self.base_path.mkdir(parents=True, exist_ok=True)
23        self.token_threshold = token_threshold
24        self.manifest = {}
25    
26    def should_offload(self, content: str) -> bool:
27        """Determine if content exceeds threshold for offloading."""
28        # Rough token estimate: 1 token ≈ 4 characters
29        estimated_tokens = len(content) // 4
30        return estimated_tokens > self.token_threshold
31    
32    def offload(self, content: str, source: str, summary: str = None) -> dict:
33        """Write content to file, return reference."""
34        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
35        filename = f"{source}_{timestamp}.txt"
36        file_path = self.base_path / filename
37        
38        file_path.write_text(content)
39        
40        reference = {
41            "type": "file_reference",
42            "path": str(file_path),
43            "source": source,
44            "timestamp": timestamp,
45            "size_chars": len(content),
46            "summary": summary or self._extract_summary(content)
47        }
48        
49        self.manifest[filename] = reference
50        return reference
51    
52    def _extract_summary(self, content: str, max_chars: int = 500) -> str:
53        """Extract first meaningful content as summary."""
54        lines = content.strip().split('\n')
55        summary_lines = []
56        char_count = 0
57        
58        for line in lines:
59            if char_count + len(line) > max_chars:
60                break
61            summary_lines.append(line)
62            char_count += len(line)
63        
64        return '\n'.join(summary_lines)
65    
66    def cleanup(self, max_age_hours: int = 24):
67        """Remove scratch files older than threshold."""
68        cutoff = datetime.now().timestamp() - (max_age_hours * 3600)
69        
70        for file_path in self.base_path.glob("*.txt"):
71            if file_path.stat().st_mtime < cutoff:
72                file_path.unlink()
73                if file_path.name in self.manifest:
74                    del self.manifest[file_path.name]
75```
76 
77### 2. Plan Persistence
78 
79Structured plan storage with progress tracking.
80 
81```python
82import yaml
83from dataclasses import dataclass, field, asdict
84from enum import Enum
85from typing import List, Optional
86 
87class StepStatus(Enum):
88    PENDING = "pending"
89    IN_PROGRESS = "in_progress"
90    COMPLETED = "completed"
91    BLOCKED = "blocked"
92    CANCELLED = "cancelled"
93 
94@dataclass
95class PlanStep:
96    id: int
97    description: str
98    status: StepStatus = StepStatus.PENDING
99    notes: Optional[str] = None
100 
101@dataclass
102class AgentPlan:
103    objective: str
104    steps: List[PlanStep] = field(default_factory=list)
105    status: str = "in_progress"
106    
107    def save(self, path: str = "scratch/current_plan.yaml"):
108        """Persist plan to filesystem."""
109        data = {
110            "objective": self.objective,
111            "status": self.status,
112            "steps": [
113                {
114                    "id": s.id,
115                    "description": s.description,
116                    "status": s.status.value,
117                    "notes": s.notes
118                }
119                for s in self.steps
120            ]
121        }
122        with open(path, 'w') as f:
123            yaml.dump(data, f, default_flow_style=False)
124    
125    @classmethod
126    def load(cls, path: str = "scratch/current_plan.yaml") -> "AgentPlan":
127        """Load plan from filesystem."""
128        with open(path, 'r') as f:
129            data = yaml.safe_load(f)
130        
131        plan = cls(objective=data["objective"], status=data.get("status", "in_progress"))
132        for step_data in data.get("steps", []):
133            plan.steps.append(PlanStep(
134                id=step_data["id"],
135                description=step_data["description"],
136                status=StepStatus(step_data["status"]),
137                notes=step_data.get("notes")
138            ))
139        return plan
140    
141    def current_step(self) -> Optional[PlanStep]:
142        """Get the first non-completed step."""
143        for step in self.steps:
144            if step.status != StepStatus.COMPLETED:
145                return step
146        return None
147    
148    def complete_step(self, step_id: int, notes: str = None):
149        """Mark step as completed."""
150        for step in self.steps:
151            if step.id == step_id:
152                step.status = StepStatus.COMPLETED
153                if notes:
154                    step.notes = notes
155                break
156```
157 
158### 3. Sub-Agent Workspace
159 
160File-based communication between agents.
161 
162```python
163from pathlib import Path
164from datetime import datetime
165import json
166 
167class AgentWorkspace:
168    """Manages file-based workspace for an agent."""
169    
170    def __init__(self, agent_id: str, base_path: str = "workspace/agents"):
171        self.agent_id = agent_id
172        self.path = Path(base_path) / agent_id
173        self.path.mkdir(parents=True, exist_ok=True)
174        
175        # Standard files
176        self.findings_file = self.path / "findings.md"
177        self.status_file = self.path / "status.json"
178        self.log_file = self.path / "activity.log"
179    
180    def write_finding(self, content: str, append: bool = True):
181        """Write or append a finding."""
182        mode = 'a' if append else 'w'
183        with open(self.findings_file, mode) as f:
184            if append:
185                f.write(f"\n---\n## {datetime.now().isoformat()}\n\n")
186            f.write(content)
187    
188    def update_status(self, status: str, progress: float = None, details: dict = None):
189        """Update agent status for coordinator visibility."""
190        status_data = {
191            "agent_id": self.agent_id,
192            "status": status,
193            "updated_at": datetime.now().isoformat(),
194            "progress": progress,
195            "details": details or {}
196        }
197        self.status_file.write_text(json.dumps(status_data, indent=2))
198    
199    def log(self, message: str):
200        """Append to activity log."""
201        with open(self.log_file, 'a') as f:
202            f.write(f"[{datetime.now().isoformat()}] {message}\n")
203    
204    def read_peer_findings(self, peer_id: str) -> str:
205        """Read findings from another agent's workspace."""
206        peer_path = self.path.parent / peer_id / "findings.md"
207        if peer_path.exists():
208            return peer_path.read_text()
209        return ""
210 
211 
212class CoordinatorWorkspace:
213    """Coordinator that reads from sub-agent workspaces."""
214    
215    def __init__(self, base_path: str = "workspace/agents"):
216        self.base_path = Path(base_path)
217    
218    def get_all_statuses(self) -> dict:
219        """Collect status from all sub-agents."""
220        statuses = {}
221        for agent_dir in self.base_path.iterdir():
222            if agent_dir.is_dir():
223                status_file = agent_dir / "status.json"
224                if status_file.exists():
225                    statuses[agent_dir.name] = json.loads(status_file.read_text())
226        return statuses
227    
228    def aggregate_findings(self) -> str:
229        """Combine all agent findings into synthesis."""
230        findings = []
231        for agent_dir in self.base_path.iterdir():
232            if agent_dir.is_dir():
233                findings_file = agent_dir / "findings.md"
234                if findings_file.exists():
235                    findings.append(f"# {agent_dir.name}\n\n{findings_file.read_text()}")
236        return "\n\n".join(findings)
237```
238 
239### 4. Dynamic Skill Loader
240 
241Load skill content on demand.
242 
243```python
244from pathlib import Path
245from typing import List, Optional
246import yaml
247 
248@dataclass
249class SkillMetadata:
250    name: str
251    description: str
252    path: str
253    triggers: List[str] = field(default_factory=list)
254 
255class SkillLoader:
256    """Manages dynamic loading of agent skills."""
257    
258    def __init__(self, skills_path: str = "skills"):
259        self.skills_path = Path(skills_path)
260        self.skill_index = self._build_index()
261    
262    def _build_index(self) -> dict:
263        """Build index of available skills from SKILL.md frontmatter."""
264        index = {}
265        for skill_dir in self.skills_path.iterdir():
266            if skill_dir.is_dir():
267                skill_file = skill_dir / "SKILL.md"
268                if skill_file.exists():
269                    metadata = self._parse_frontmatter(skill_file)
270                    if metadata:
271                        index[metadata.name] = metadata
272        return index
273    
274    def _parse_frontmatter(self, path: Path) -> Optional[SkillMetadata]:
275        """Extract YAML frontmatter from skill file."""
276        content = path.read_text()
277        if content.startswith('---'):
278            end = content.find('---', 3)
279            if end > 0:
280                frontmatter = yaml.safe_load(content[3:end])
281                return SkillMetadata(
282                    name=frontmatter.get('name', path.parent.name),
283                    description=frontmatter.get('description', ''),
284                    path=str(path),
285                    triggers=frontmatter.get('triggers', [])
286                )
287        return None
288    
289    def get_static_context(self) -> str:
290        """Generate minimal static context listing available skills."""
291        lines = ["Available skills (load with read_file when relevant):"]
292        for name, meta in self.skill_index.items():
293            lines.append(f"- {name}: {meta.description[:100]}")
294        return "\n".join(lines)
295    
296    def load_skill(self, name: str) -> str:
297        """Load full skill content."""
298        if name in self.skill_index:
299            return Path(self.skill_index[name].path).read_text()
300        raise ValueError(f"Unknown skill: {name}")
301    
302    def find_relevant_skills(self, query: str) -> List[str]:
303        """Find skills that might be relevant to a query."""
304        query_lower = query.lower()
305        relevant = []
306        for name, meta in self.skill_index.items():
307            if any(trigger in query_lower for trigger in meta.triggers):
308                relevant.append(name)
309            elif name.replace('-', ' ') in query_lower:
310                relevant.append(name)
311        return relevant
312```
313 
314### 5. Terminal Output Persistence
315 
316Capture and persist terminal sessions.
317 
318```python
319import subprocess
320from pathlib import Path
321from datetime import datetime
322import re
323 
324class TerminalCapture:
325    """Captures and persists terminal output for agent access."""
326    
327    def __init__(self, terminals_path: str = "terminals"):
328        self.terminals_path = Path(terminals_path)
329        self.terminals_path.mkdir(parents=True, exist_ok=True)
330        self.session_counter = 0
331    
332    def run_command(self, command: str, capture: bool = True) -> dict:
333        """Run command and optionally capture output to file."""
334        self.session_counter += 1
335        
336        result = subprocess.run(
337            command,
338            shell=True,
339            capture_output=True,
340            text=True
341        )
342        
343        output = {
344            "command": command,
345            "exit_code": result.returncode,
346            "stdout": result.stdout,
347            "stderr": result.stderr,
348            "timestamp": datetime.now().isoformat()
349        }
350        
351        if capture:
352            output["file"] = self._persist_output(output)
353        
354        return output
355    
356    def _persist_output(self, output: dict) -> str:
357        """Write output to terminal file."""
358        filename = f"{self.session_counter}.txt"
359        file_path = self.terminals_path / filename
360        
361        content = f"""---
362command: {output['command']}
363exit_code: {output['exit_code']}
364timestamp: {output['timestamp']}
365---
366 
367=== STDOUT ===
368{output['stdout']}
369 
370=== STDERR ===
371{output['stderr']}
372"""
373        file_path.write_text(content)
374        return str(file_path)
375    
376    def grep_terminals(self, pattern: str, context_lines: int = 3) -> List[dict]:
377        """Search all terminal outputs for pattern."""
378        matches = []
379        regex = re.compile(pattern, re.IGNORECASE)
380        
381        for term_file in self.terminals_path.glob("*.txt"):
382            content = term_file.read_text()
383            lines = content.split('\n')
384            
385            for i, line in enumerate(lines):
386                if regex.search(line):
387                    start = max(0, i - context_lines)
388                    end = min(len(lines), i + context_lines + 1)
389                    matches.append({
390                        "file": str(term_file),
391                        "line_number": i + 1,
392                        "context": '\n'.join(lines[start:end])
393                    })
394        
395        return matches
396```
397 
398### 6. Self-Modification Guard
399 
400Safe pattern for agent self-learning.
401 
402```python
403import yaml
404from pathlib import Path
405from datetime import datetime
406from typing import Any
407 
408class PreferenceStore:
409    """Guarded storage for agent-learned preferences."""
410    
411    MAX_ENTRIES = 100
412    MAX_VALUE_LENGTH = 1000
413    
414    def __init__(self, path: str = "agent/preferences.yaml"):
415        self.path = Path(path)
416        self.path.parent.mkdir(parents=True, exist_ok=True)
417        self.preferences = self._load()
418    
419    def _load(self) -> dict:
420        """Load preferences from file."""
421        if self.path.exists():
422            return yaml.safe_load(self.path.read_text()) or {}
423        return {}
424    
425    def _save(self):
426        """Persist preferences to file."""
427        self.path.write_text(yaml.dump(self.preferences, default_flow_style=False))
428    
429    def remember(self, key: str, value: Any, source: str = "user"):
430        """Store a preference with validation."""
431        # Validate key
432        if not key or len(key) > 100:
433            raise ValueError("Invalid key length")
434        
435        # Validate value
436        value_str = str(value)
437        if len(value_str) > self.MAX_VALUE_LENGTH:
438            raise ValueError(f"Value exceeds max length of {self.MAX_VALUE_LENGTH}")
439        
440        # Check entry limit
441        if len(self.preferences) >= self.MAX_ENTRIES and key not in self.preferences:
442            raise ValueError(f"Max entries ({self.MAX_ENTRIES}) reached")
443        
444        # Store with metadata
445        self.preferences[key] = {
446            "value": value,
447            "source": source,
448            "updated_at": datetime.now().isoformat()
449        }
450        self._save()
451    
452    def recall(self, key: str, default: Any = None) -> Any:
453        """Retrieve a preference."""
454        entry = self.preferences.get(key)
455        if entry:
456            return entry["value"]
457        return default
458    
459    def list_all(self) -> dict:
460        """Get all preferences for context injection."""
461        return {k: v["value"] for k, v in self.preferences.items()}
462    
463    def forget(self, key: str):
464        """Remove a preference."""
465        if key in self.preferences:
466            del self.preferences[key]
467            self._save()
468```
469 
470## Integration Example
471 
472Combining patterns in an agent harness:
473 
474```python
475class FilesystemContextAgent:
476    """Agent with filesystem-based context management."""
477    
478    def __init__(self):
479        self.scratch = ScratchPadManager()
480        self.skills = SkillLoader()
481        self.preferences = PreferenceStore()
482        self.workspace = AgentWorkspace("main_agent")
483    
484    def handle_tool_output(self, tool_name: str, output: str) -> str:
485        """Process tool output, offloading if necessary."""
486        if self.scratch.should_offload(output):
487            ref = self.scratch.offload(output, source=tool_name)
488            return f"[{tool_name} output saved to {ref['path']}. Summary: {ref['summary'][:200]}]"
489        return output
490    
491    def get_system_prompt(self) -> str:
492        """Build system prompt with dynamic skill references."""
493        base_prompt = "You are a helpful assistant."
494        skill_context = self.skills.get_static_context()
495        user_prefs = self.preferences.list_all()
496        
497        pref_section = ""
498        if user_prefs:
499            pref_section = "\n\nUser preferences:\n" + "\n".join(
500                f"- {k}: {v}" for k, v in user_prefs.items()
501            )
502        
503        return f"{base_prompt}\n\n{skill_context}{pref_section}"
504```
505 
506## File Organization Best Practices
507 
508```
509project/
510├── scratch/                    # Ephemeral working files
511│   ├── tool_outputs/          # Large tool results
512│   │   └── search_20260107.txt
513│   └── plans/                 # Active task plans
514│       └── current_plan.yaml
515├── workspace/                  # Agent workspaces
516│   └── agents/
517│       ├── research_agent/
518│       │   ├── findings.md
519│       │   └── status.json
520│       └── code_agent/
521│           ├── findings.md
522│           └── status.json
523├── agent/                      # Agent configuration
524│   ├── preferences.yaml       # Learned preferences
525│   └── patterns.md           # Discovered patterns
526├── skills/                     # Loadable skills
527│   └── {skill-name}/
528│       └── SKILL.md
529├── terminals/                  # Terminal output
530│   ├── 1.txt
531│   └── 2.txt
532└── history/                    # Chat history archives
533    └── session_001.txt
534```
535 
536## Token Accounting Metrics
537 
538Track these metrics to validate filesystem patterns:
539 
5401. **Static context ratio**: tokens in static context / total tokens
5412. **Dynamic load rate**: how often skills/files are loaded per task
5423. **Offload savings**: tokens saved by writing to files vs keeping in context
5434. **Retrieval precision**: percentage of loaded content actually used
544 
545Target benchmarks:
546- Static context ratio < 20%
547- Offload savings > 50% for tool-heavy workflows
548- Retrieval precision > 70% (loaded content is relevant)
549 
550