Day 1: Research (5000 tokens context)
Day 2: Load research again (5000 tokens) + Plan (3000 tokens) = 8000 tokens
Day 3: Load research + plan (8000 tokens) + Implement (4000 tokens) = 12,000 tokens

Result: 60% context utilization → Context collapse

# Research Phase Output (5000 tokens)

## Background

[Detailed background...]

## Literature Review

[Full citations, quotes, analysis...]

## Technical Investigation

[Complete technical deep-dive...]

## Conclusions

[Detailed findings...]

# Research Bundle (1000 tokens)

**What we learned:**

- Background: [Key points only]
- Literature: [3 critical findings]
- Technical: [Essential architecture decisions]
- Conclusions: [2 sentence summary]

**Decisions made:**

- Use Kubernetes StatefulSets (not Deployments)
- PostgreSQL for persistence (not MySQL)

**Carry forward:**

- Namespace: production-db
- Replicas: 3
- Storage class: fast-ssd

Day 1: Research in detail (high context)
Day 2: Review research bundle (low context, reinforcement)
Day 3: Reference research bundle again (spaced repetition)

Result: Better retention with less context

Git repository:
├── research-phase/ (full 5000 tokens preserved)
├── plan-phase/ (full 3000 tokens preserved)
└── implement-phase/ (full 4000 tokens preserved)

Context bundles:
├── research-bundle.md (1000 tokens for loading)
├── plan-bundle.md (600 tokens for loading)
└── implement-bundle.md (800 tokens for loading)

# [Phase] Bundle

## What we did

[1-2 sentence summary]

## Key decisions

- Decision 1: [Choice + Why]
- Decision 2: [Choice + Why]

## Critical findings

- Finding 1: [Impact]
- Finding 2: [Impact]

## Parameters for next phase

- Param 1: value
- Param 2: value

## Next steps

- [ ] Step 1
- [ ] Step 2

class BundleGenerator:
    def compress(self, full_document, compression_ratio=5):
        # Extract structure
        sections = self.parse_sections(full_document)

        # Prioritize content
        critical = self.extract_critical_info(sections)
        decisions = self.extract_decisions(sections)
        parameters = self.extract_parameters(sections)

        # Build bundle
        bundle = f"""
        # {full_document.phase} Bundle

        ## Summary
        {self.summarize(sections, max_tokens=100)}

        ## Decisions
        {self.format_decisions(decisions)}

        ## Parameters
        {self.format_parameters(parameters)}

        ## Next Steps
        {self.extract_next_steps(sections)}
        """

        # Verify compression ratio
        original_tokens = count_tokens(full_document)
        bundle_tokens = count_tokens(bundle)
        actual_ratio = original_tokens / bundle_tokens

        assert actual_ratio >= compression_ratio, "Insufficient compression"

        return bundle

class MultiDayWorkflow:
    def execute_day_2(self):
        # Load compressed context from Day 1
        research_bundle = load_bundle('research-bundle.md')  # 1000 tokens

        # Fresh context for today's work
        plan_context = load_plan_templates()  # 2000 tokens

        # Total: 3000 tokens (15% utilization ✅)
        plan_result = plan_agent.execute(research_bundle, plan_context)

        # Save new bundle
        save_bundle('plan-bundle.md', compress(plan_result))

    def execute_day_3(self):
        # Load compressed contexts from Days 1-2
        research_bundle = load_bundle('research-bundle.md')  # 1000 tokens
        plan_bundle = load_bundle('plan-bundle.md')  # 600 tokens

        # Fresh context for today's work
        implement_context = load_relevant_code()  # 4000 tokens

        # Total: 5600 tokens (28% utilization ✅)
        implement_result = implement_agent.execute(
            research_bundle,
            plan_bundle,
            implement_context
        )

def validate_bundle(bundle):
    # 1. Compression ratio check
    if bundle.compression_ratio < 4:
        raise BundleError("Insufficient compression")

    # 2. Critical information check
    required_fields = ['decisions', 'parameters', 'next_steps']
    for field in required_fields:
        if field not in bundle:
            raise BundleError(f"Missing required field: {field}")

    # 3. Token limit check
    if count_tokens(bundle) > 2000:
        raise BundleError("Bundle exceeds 2000 token limit")

    # 4. Context preservation check
    if not verify_essential_context(bundle):
        raise BundleError("Essential context not preserved")

    return True

Day 1: Research (15% context)
Day 2: Load research + Plan (40% context) ⚠️
Day 3: Load research + plan + Implement (65% context) ❌ Collapse
Result: Abandoned, started over

Day 1: Research (15% context) → Save bundle (1500 tokens)
Day 2: Load research bundle (8%) + Plan (20%) = 28% ✅ → Save plan bundle (800 tokens)
Day 3: Load bundles (10%) + Implement (25%) = 35% ✅
Day 4: Load bundles (12%) + Test (18%) = 30% ✅
Day 5: Load bundles (12%) + Document (15%) = 27% ✅
Day 6: Load bundles (12%) + Deploy (20%) = 32% ✅
Day 7: Load bundles (12%) + Monitor (10%) = 22% ✅

Result: Completed successfully, never exceeded 40%

research-bundle.md (Day 1 output)
└── plan-bundle.md (Day 2 output, references research-bundle)
    └── implement-bundle.md (Day 3 output, references plan-bundle)
        └── deploy-bundle.md (Day 4 output, references implement-bundle)

# Day 1
research_bundle = create_bundle(research_result)

# Day 2
plan_bundle = create_bundle(plan_result)
combined_bundle_day2 = merge_bundles([research_bundle, plan_bundle])

# Day 3
implement_bundle = create_bundle(implement_result)
combined_bundle_day3 = merge_bundles([research_bundle, plan_bundle, implement_bundle])

# Each day: New bundle + merged context bundle (auto-compressed)

class BundleLoader:
    def load_with_detail(self, bundle, detail_level='summary'):
        # Load compressed bundle
        summary = load_bundle(bundle)

        # If more detail needed, fetch from Git
        if detail_level == 'full':
            return load_from_git(bundle.source_commit)

        # If specific section needed, fetch just that
        elif detail_level == 'section':
            return load_section_from_git(bundle.source_commit, section)

        # Default: return compressed summary
        return summary

┌─────────────────────────────────────────────────────────────┐
│                    AGENT MEMORY ARCHITECTURE                 │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  ┌─────────────────┐  ┌─────────────────┐  ┌─────────────┐ │
│  │   RAG Layer     │  │  Graph Layer    │  │  Historical │ │
│  │   (Semantic)    │  │ (Relationships) │  │   Layer     │ │
│  ├─────────────────┤  ├─────────────────┤  ├─────────────┤ │
│  │ pgvector/Qdrant │  │     Neo4j       │  │ PostgreSQL  │ │
│  │                 │  │                 │  │             │ │
│  │ "Find similar   │  │ "What calls     │  │ "Has this   │ │
│  │  code to this"  │  │  this function?"│  │  failed     │ │
│  │                 │  │                 │  │  before?"   │ │
│  │ Semantic search │  │ Graph traversal │  │ Time-series │ │
│  │ over content    │  │ over structure  │  │ patterns    │ │
│  └─────────────────┘  └─────────────────┘  └─────────────┘ │
│                                                             │
└─────────────────────────────────────────────────────────────┘

❌ WRONG: Store everything in pgvector
- Relationships lost (no "who calls what")
- Time patterns lost (no "has this failed before")
- Query efficiency suffers

✅ RIGHT: Use the appropriate layer
- RAG: Content search (embeddings)
- Graph: Structure search (relationships)
- Historical: Pattern search (time-series)

# Hybrid retrieval pipeline
retrieval:
  # Step 1: Semantic search
  rag:
    store: pgvector
    query: 'SELECT * FROM embeddings ORDER BY embedding <-> $query_embedding LIMIT 10'

  # Step 2: Expand via relationships
  graph:
    store: neo4j
    query: |
      MATCH (chunk:Chunk)-[:MENTIONS]->(entity)
      WHERE chunk.id IN $rag_results
      MATCH (entity)<-[:MENTIONS]-(related:Chunk)
      RETURN related

  # Step 3: Check historical patterns
  historical:
    store: postgresql
    query: 'SELECT failure_rate, last_failure FROM deployment_history WHERE service_id = $entity_id'

  # Step 4: Merge and rank
  merge: hybrid

Agent Invocation (stateless):
  Input:  Event + Context (loaded from external memory)
  Output: Response + Actions (persisted to external memory)

  Agent itself: Pure function, no internal state

# Agent execution model
class StatelessAgent:
    def execute(self, event: Event, memory: MemorySystem) -> AgentResult:
        # Load context from external memory
        context = memory.load(
            session_id=event.session_id,
            layers=['rag', 'graph', 'historical']
        )

        # Agent does its work (pure function)
        result = self.process(event, context)

        # Persist results to external memory
        memory.store(
            session_id=event.session_id,
            layer='historical',
            data=result.observations
        )

        return result

class MemorySystem:
    """External memory backing for stateless agents"""

    def __init__(self, postgresql: PostgreSQL, neo4j: Neo4j, pgvector: PgVector):
        self.historical = postgresql  # Time-series patterns
        self.graph = neo4j            # Relationships
        self.rag = pgvector           # Semantic search

    def load(self, session_id: str, layers: List[str]) -> Context:
        """Load context from specified memory layers"""
        context = {}
        if 'historical' in layers:
            context['history'] = self.historical.query_session(session_id)
        if 'graph' in layers:
            context['relationships'] = self.graph.query_session(session_id)
        if 'rag' in layers:
            context['similar'] = self.rag.query_session(session_id)
        return Context(**context)

    def store(self, session_id: str, layer: str, data: Any):
        """Persist data to specified memory layer"""
        if layer == 'historical':
            self.historical.insert(session_id, data)
        elif layer == 'graph':
            self.graph.upsert(session_id, data)
        elif layer == 'rag':
            self.rag.upsert(session_id, data)

prep → queued → held → launching → active → complete|failed|aborted

class TransitionReason(Enum):
    USER_REQUEST = "user_request"      # User initiated
    SCHEDULER = "scheduler"            # Scheduler decision
    AGENT_READY = "agent_ready"        # Agent confirmed launch
    COMPLETION = "completion"          # Work finished successfully
    FAILURE = "failure"                # Error occurred
    TIMEOUT = "timeout"                # Deadline exceeded
    QUOTA_EXCEEDED = "quota_exceeded"  # Budget/limit hit
    RATE_LIMIT = "rate_limit"          # Too many requests

from enum import Enum
from dataclasses import dataclass
from datetime import datetime

class MissionPhase(Enum):
    PREP = "prep"
    QUEUED = "queued"
    HELD = "held"
    LAUNCHING = "launching"
    ACTIVE = "active"
    COMPLETE = "complete"
    FAILED = "failed"
    ABORTED = "aborted"

@dataclass
class StateTransition:
    from_state: MissionPhase
    to_state: MissionPhase
    reason: TransitionReason
    timestamp: datetime
    details: dict

class MissionStateMachine:
    VALID_TRANSITIONS = {
        MissionPhase.PREP: [MissionPhase.QUEUED, MissionPhase.ABORTED],
        MissionPhase.QUEUED: [MissionPhase.HELD, MissionPhase.LAUNCHING, MissionPhase.ABORTED],
        MissionPhase.HELD: [MissionPhase.QUEUED, MissionPhase.ABORTED],
        MissionPhase.LAUNCHING: [MissionPhase.ACTIVE, MissionPhase.FAILED, MissionPhase.ABORTED],
        MissionPhase.ACTIVE: [MissionPhase.COMPLETE, MissionPhase.FAILED, MissionPhase.ABORTED],
    }

    def transition(self, mission_id: str, to_state: MissionPhase, reason: TransitionReason):
        current = self.get_state(mission_id)

        # Validate transition
        if to_state not in self.VALID_TRANSITIONS.get(current, []):
            raise InvalidTransition(f"Cannot transition from {current} to {to_state}")

        # Log transition with reason
        transition = StateTransition(
            from_state=current,
            to_state=to_state,
            reason=reason,
            timestamp=datetime.utcnow(),
            details={"mission_id": mission_id}
        )
        self.log_transition(transition)

        # Update state
        self.set_state(mission_id, to_state)

        return transition

// Core session nodes
(:Session {id, started_at, phase, bundle_path})
(:Agent {name, version, type})
(:Event {id, type, timestamp, payload})
(:Decision {id, description, rationale, timestamp})
(:Artifact {id, path, type, created_at})

// Relationships
(Session)-[:EXECUTED_BY]->(Agent)
(Session)-[:TRIGGERED_BY]->(Event)
(Session)-[:MADE_DECISION]->(Decision)
(Session)-[:PRODUCED]->(Artifact)
(Session)-[:CONTINUES]->(Session)  // Session chaining
(Decision)-[:LED_TO]->(Decision)   // Decision dependencies
(Artifact)-[:USED_IN]->(Session)   // Artifact reuse

// Find all artifacts from sessions that led to this one
MATCH (current:Session {id: $session_id})
MATCH (current)-[:CONTINUES*]->(previous:Session)
MATCH (previous)-[:PRODUCED]->(artifact:Artifact)
RETURN artifact

// Find decisions that affected this session
MATCH (s:Session {id: $session_id})
MATCH (s)-[:CONTINUES*]->(prev:Session)-[:MADE_DECISION]->(d:Decision)
RETURN d ORDER BY d.timestamp DESC

// Find sessions that used similar artifacts (for pattern mining)
MATCH (current:Session {id: $session_id})-[:PRODUCED]->(a:Artifact)
MATCH (other:Session)-[:USED_IN]->(a)
WHERE other.id <> current.id
RETURN other, count(a) as shared_artifacts
ORDER BY shared_artifacts DESC

class Neo4jSessionStore:
    def create_session(self, session_id: str, agent: str, event: dict) -> None:
        """Create new session node linked to triggering event"""
        self.graph.run("""
            MERGE (s:Session {id: $session_id})
            SET s.started_at = datetime(), s.phase = 'prep'

            MERGE (a:Agent {name: $agent})
            MERGE (e:Event {id: $event_id})
            SET e.type = $event_type, e.payload = $payload

            MERGE (s)-[:EXECUTED_BY]->(a)
            MERGE (s)-[:TRIGGERED_BY]->(e)
        """, session_id=session_id, agent=agent, **event)

    def continue_session(self, new_session_id: str, previous_session_id: str) -> None:
        """Link new session to previous (for multi-day work)"""
        self.graph.run("""
            MATCH (prev:Session {id: $previous_id})
            MERGE (new:Session {id: $new_id})
            MERGE (new)-[:CONTINUES]->(prev)
        """, previous_id=previous_session_id, new_id=new_session_id)

    def record_decision(self, session_id: str, decision: dict) -> None:
        """Record a decision made during the session"""
        self.graph.run("""
            MATCH (s:Session {id: $session_id})
            CREATE (d:Decision {
                id: $decision_id,
                description: $description,
                rationale: $rationale,
                timestamp: datetime()
            })
            MERGE (s)-[:MADE_DECISION]->(d)
        """, session_id=session_id, **decision)

    def load_context(self, session_id: str) -> dict:
        """Load accumulated context from session chain"""
        result = self.graph.run("""
            MATCH (current:Session {id: $session_id})
            OPTIONAL MATCH (current)-[:CONTINUES*]->(prev:Session)
            OPTIONAL MATCH (prev)-[:MADE_DECISION]->(d:Decision)
            OPTIONAL MATCH (prev)-[:PRODUCED]->(a:Artifact)
            RETURN prev, collect(DISTINCT d) as decisions, collect(DISTINCT a) as artifacts
        """, session_id=session_id)

        return {
            "previous_sessions": [r["prev"] for r in result],
            "decisions": flatten([r["decisions"] for r in result]),
            "artifacts": flatten([r["artifacts"] for r in result])
        }