The Death of the Black Box: Architecting Self-Improving Persistent Development Workspaces with Agentic Context Systems

sequenceDiagram
    participant Dev as Developer
    participant IDE as IDE Plugin
    participant LLM as Stateless LLM
    
    Dev->>IDE: "Fix this bug"
    IDE->>LLM: prompt + file snippet
    LLM-->>IDE: code suggestion
    IDE-->>Dev: display suggestion
    Note over LLM: Session ends. All context lost.
    
    Dev->>IDE: "Why did you suggest that approach?"
    IDE->>LLM: new prompt (no memory of previous)
    LLM-->>IDE: generic answer

┌─────────────────────────────────────────────────────────┐
│                    USER INTERACTION LAYER                  │
│   (IDE Integration, CLI, Web Interface, Chat Protocol)     │
├─────────────────────────────────────────────────────────┤
│                    AGENT ORCHESTRATION                     │
│   (Planning, Task Decomposition, Tool Selection)          │
├─────────────────────────────────────────────────────────┤
│                    CONTEXT ENGINE                          │
│  ┌──────────┐ ┌──────────┐ ┌───────────┐ ┌──────────┐  │
│  │ Semantic │ │ Temporal │ │ Decision  │ │ Working  │  │
│  │  Index   │ │  Buffer  │ │  Memory   │ │  Memory  │  │
│  └──────────┘ └──────────┘ └───────────┘ └──────────┘  │
├─────────────────────────────────────────────────────────┤
│                    PERSISTENCE LAYER                       │
│   (Vector DB, Graph DB, File System, Event Log)           │
├─────────────────────────────────────────────────────────┤
│                    TOOL EXECUTION LAYER                    │
│   (File System, Shell, Test Runner, Lint, Git)            │
├─────────────────────────────────────────────────────────┤
│                    LLM INFERENCE LAYER                     │
│   (Model Router, Prompt Builder, Response Parser)         │
└─────────────────────────────────────────────────────────┘

class AgenticWorkspace:
    def run(self, user_request: str):
        # 1. Retrieve relevant context from persistent memory
        context = self.context_engine.retrieve(user_request)
        
        # 2. Formulate plan based on context + request
        plan = self.agent.plan(user_request, context)
        
        # 3. Execute actions (may involve multiple steps)
        for step in plan.steps:
            result = self.execute_step(step)
            
            # 4. Observe and update context
            self.context_engine.ingest(
                event=step,
                result=result,
                user_feedback=self.get_feedback(step)
            )
            
            # 5. Self-improvement: adjust based on outcomes
            self.improve(step, result)
        
        # 6. Persist session state
        self.context_engine.commit(session_id=self.session.id)

interface ContextEngine {
  // Retrieval
  retrieve(query: string, options: RetrievalOptions): Promise<ContextBundle>;
  
  // Ingestion
  ingest(event: WorkspaceEvent): Promise<IngestionResult>;
  
  // Session management
  beginSession(workspaceId: string): Promise<SessionContext>;
  commit(sessionId: string): Promise<void>;
  resume(sessionId: string): Promise<SessionContext>;
  
  // Decay and consolidation
  consolidate(): Promise<ConsolidationResult>;
  prune(options: PruneOptions): Promise<PruneResult>;
  
  // Query capabilities
  searchDecisions(query: string): Promise<DecisionRecord[]>;
  getProjectArchitecture(): Promise<ArchitectureMap>;
  getRecentErrors(): Promise<ErrorRecord[]>;
}

interface RetrievalOptions {
  // What to retrieve
  includeSemantic: boolean;
  includeEpisodic: boolean;
  includeProcedural: boolean;
  
  // How much
  maxTokens: number;
  relevanceThreshold: number;
  
  // Recency weighting
  recencyDecay: 'none' | 'linear' | 'exponential';
  
  // Scope
  fileScope?: string[];
  directoryScope?: string[];
}

interface ContextBundle {
  semantic: SemanticContext;    // Codebase understanding
  episodic: EpisodicContext;    // Past interactions
  procedural: ProceduralContext; // Learned preferences
  working: WorkingContext;      // Current session state
  tokenBudget: TokenBudget;     // How much was used vs available
}

class TokenBudgetAllocator:
    def __init__(self, total_budget: int = 128000):
        self.total_budget = total_budget
        self.reserved_system = int(total_budget * 0.10)  # System prompt
        self.reserved_output = int(total_budget * 0.20)  # Model output space
        self.available = total_budget - self.reserved_system - self.reserved_output
        
    def allocate(self, context_bundle: ContextBundle, 
                 query: str) -> AllocatedContext:
        
        scores = {
            'semantic': self._score_semantic(context_bundle.semantic, query),
            'episodic': self._score_episodic(context_bundle.episodic, query),
            'procedural': self._score_procedural(context_bundle.procedural, query),
            'working': 1.0,  # Always include working memory
        }
        
        # Weighted allocation based on relevance and type
        allocations = self._distribute_budget(scores, self.available)
        
        return AllocatedContext(
            semantic=self._truncate_to_tokens(
                context_bundle.semantic, allocations['semantic']
            ),
            episodic=self._truncate_to_tokens(
                context_bundle.episodic, allocations['episodic']
            ),
            procedural=self._truncate_to_tokens(
                context_bundle.procedural, allocations['procedural']
            ),
            working=context_bundle.working,
        )
    
    def _score_semantic(self, ctx: SemanticContext, query: str) -> float:
        # Higher score if the semantic context directly relates to the query
        return ctx.relevance_score(query) * 0.8 + 0.2  # Floor at 0.2

from typing import Optional
from dataclasses import dataclass
from enum import Enum


class NodeType(Enum):
    FILE = "file"
    FUNCTION = "function"
    CLASS = "class"
    MODULE = "module"
    DECISION = "decision"
    ERROR = "error"
    PREFERENCE = "preference"
    PATTERN = "pattern"


class EdgeType(Enum):
    IMPORTS = "imports"
    CALLS = "calls"
    EXTENDS = "extends"
    IMPLEMENTS = "implements"
    DECIDED_IN = "decided_in"  # Decision -> File
    CAUSED_BY = "caused_by"     # Error -> Code
    RESOLVED_BY = "resolved_by" # Error -> Fix
    LEARNED_FROM = "learned_from" # Preference -> Interaction


@dataclass
class MemoryNode:
    id: str
    type: NodeType
    content: str
    embedding: list[float]
    metadata: dict
    created_at: float
    last_accessed: float
    access_count: int = 0
    confidence: float = 1.0  # How confident are we this is still valid?


@dataclass
class MemoryEdge:
    source: str
    target: str
    type: EdgeType
    weight: float = 1.0
    confidence: float = 1.0

class HybridMemoryStore:
    def __init__(self, vector_db, graph_db):
        self.vector_db = vector_db  # e.g., Qdrant, Weaviate, pgvector
        self.graph_db = graph_db    # e.g., Neo4j, NebulaGraph
        
    async def query(self, request: MemoryQuery) -> MemoryResults:
        # Phase 1: Vector search for semantic matches
        semantic_results = await self.vector_db.search(
            vector=request.query_embedding,
            filter=request.filters,
            limit=request.limit * 3  # Over-fetch for ranking
        )
        
        # Phase 2: Graph expansion for structural context
        expanded = await self.graph_db.expand(
            node_ids=[r.id for r in semantic_results[:10]],
            depth=request.graph_depth,
            edge_types=request.preferred_edges
        )
        
        # Phase 3: Reciprocal rank fusion
        fused = self._reciprocal_rank_fusion(
            semantic_results, expanded, k=60
        )
        
        # Phase 4: Decay adjustment
        for result in fused:
            age_days = (time.time() - result.last_accessed) / 86400
            decay_factor = self._calculate_decay(result, age_days)
            result.score *= decay_factor
        
        return MemoryResults(
            items=fused[:request.limit],
            token_count=self._count_tokens(fused[:request.limit])
        )
    
    def _calculate_decay(self, result: MemoryNode, age_days: float) -> float:
        """
        Different memory types decay at different rates.
        Procedural memories decay very slowly.
        Episodic memories decay moderately.
        Semantic memories decay only if the code changes.
        """
        decay_rates = {
            NodeType.PROCEDURE: 0.001,  # 99.9% retained per day
            NodeType.EPISODIC: 0.01,    # 99% retained per day  
            NodeType.SEMANTIC: 0.005,   # Code changes override this
            NodeType.DECISION: 0.002,   # Decisions are sticky
        }
        base_rate = decay_rates.get(result.type, 0.01)
        return base_rate ** age_days

import ast
import hashlib
from pathlib import Path


class CodebaseIndexer:
    def __init__(self, memory_store: HybridMemoryStore, embedding_model):
        self.store = memory_store
        self.embedder = embedding_model
        
    async def index_repository(self, root: Path):
        files = self._discover_files(root)
        
        # Phase 1: Parse AST and extract structure
        for file_path in files:
            tree = self._parse_ast(file_path)
            if tree:
                await self._index_ast_nodes(tree, file_path)
        
        # Phase 2: Build dependency graph
        await self._build_dependency_graph(files)
        
        # Phase 3: Generate semantic summaries
        for file_path in files:
            summary = await self._generate_summary(file_path)
            await self.store.upsert(
                node=MemoryNode(
                    id=self._stable_id(file_path),
                    type=NodeType.FILE,
                    content=summary,
                    embedding=self.embedder.encode(summary),
                    metadata={'path': str(file_path), 'hash': self._file_hash(file_path)}
                )
            )
        
    def _parse_ast(self, path: Path) -> ast.AST | None:
        try:
            source = path.read_text()
            return ast.parse(source)
        except (SyntaxError, UnicodeDecodeError):
            return None
    
    def _index_ast_nodes(self, tree: ast.AST, file_path: Path):
        """Extract functions, classes, and their signatures for graph indexing."""
        for node in ast.walk(tree):
            if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
                signature = self._extract_signature(node)
                docstring = ast.get_docstring(node) or ""
                yield MemoryNode(
                    id=f"{file_path}:{node.name}",
                    type=NodeType.FUNCTION,
                    content=f"{node.name}: {signature}\n{docstring}",
                    ...
                )
            elif isinstance(node, ast.ClassDef):
                methods = [n.name for n in node.body if isinstance(n, ast.FunctionDef)]
                yield MemoryNode(
                    id=f"{file_path}:{node.name}",
                    type=NodeType.CLASS,
                    content=f"class {node.name}({', '.join(methods)})",
                    ...
                )

class FeedbackType(Enum):
    EXPLICIT_ACCEPT = "explicit_accept"      # User accepts suggestion
    EXPLICIT_REJECT = "explicit_reject"      # User rejects suggestion  
    MODIFICATION = "modification"             # User modifies suggestion
    OVERRIDE = "override"                    # User writes completely different code
    SILENT_ACCEPT = "silent_accept"          # User doesn't change suggestion
    CORRECTION = "correction"                # User explicitly corrects a fact
    ESCALATION = "escalation"                # User asks for more context/different approach


class FeedbackSignal:
    """Captured feedback that feeds into self-improvement."""
    type: FeedbackType
    original_suggestion: str
    user_response: str
    context: ContextBundle
    timestamp: float
    
    @property
    def edit_distance(self) -> float:
        """How much did the user change our suggestion?"""
        return difflib.SequenceMatcher(
            None, self.original_suggestion, self.user_response
        ).ratio()

class PreferenceLearner:
    def __init__(self, min_samples: int = 5, confidence_threshold: float = 0.75):
        self.preferences: dict[str, LearnedPreference] = {}
        self.min_samples = min_samples
        self.confidence_threshold = confidence_threshold
        
    def observe(self, signal: FeedbackSignal):
        # Extract preference candidates from this interaction
        candidates = self._extract_preference_candidates(signal)
        
        for candidate in candidates:
            key = self._preference_key(candidate)
            
            if key not in self.preferences:
                self.preferences[key] = LearnedPreference(
                    dimension=key,
                    votes=[candidate.value],
                    confidence=0.0
                )
            else:
                self.preferences[key].votes.append(candidate.value)
                self._update_confidence(self.preferences[key])
    
    def get_active_preferences(self) -> list[LearnedPreference]:
        return [
            p for p in self.preferences.values()
            if p.confidence >= self.confidence_threshold
            and len(p.votes) >= self.min_samples
        ]
    
    def _extract_preference_candidates(self, signal: FeedbackSignal) -> list[PreferenceCandidate]:
        candidates = []
        
        if signal.type == FeedbackType.MODIFICATION:
            # Analyze what the user changed
            diff = self._analyze_diff(signal.original_suggestion, signal.user_response)
            
            if diff.naming_changes:
                candidates.append(PreferenceCandidate(
                    dimension="naming_convention",
                    value=diff.naming_pattern
                ))
            
            if diff.structural_changes:
                candidates.append(PreferenceCandidate(
                    dimension="code_structure",
                    value=diff.structural_pattern
                ))
            
            if diff.style_changes:
                candidates.append(PreferenceCandidate(
                    dimension="code_style",
                    value=diff.style_pattern
                ))
        
        return candidates

class SelfImprovementEngine:
    async def process_feedback_batch(self, signals: list[FeedbackSignal]):
        # Level 1: Immediate adjustments
        immediate_adjustments = self._derive_prompt_adjustments(signals)
        await self.context_engine.update_procedural_memory(immediate_adjustments)
        
        # Level 2: Pattern updates
        pattern_updates = self._analyze_patterns(signals)
        for update in pattern_updates:
            if update.is_reinforcement:
                await self.graph_db.boost_node(update.node_id, weight=update.strength)
            elif update.is_weakening:
                await self.graph_db.weaken_node(update.node_id, weight=update.strength)
            elif update.is_new_pattern:
                await self.graph_db.create_edge(update.edge)
        
        # Level 3: Schedule model adaptation if enough data
        if len(signals) >= self.adaptation_threshold:
            await self._schedule_model_adaptation(signals)

# config/workspace.yaml
storage:
  vector_db:
    provider: qdrant
    collection: workspace_memory
    dimensions: 1536  # OpenAI embedding dimension
    distance: cosine
    shards: 4
    replicas: 2
    
  graph_db:
    provider: neo4j
    database: workspace_graph
    connection_pool:
      min_size: 5
      max_size: 20
    
  event_log:
    provider: append_only_log  # e.g., Kafka, or local log-structured storage
    retention_days: 90
    compression: zstd
    
  cache:
    provider: redis
    ttl_seconds: 3600
    max_memory_mb: 2048

class WorkspaceSession:
    """A complete session with full context persistence."""
    
    def __init__(self, workspace_id: str, context_engine: ContextEngine):
        self.workspace_id = workspace_id
        self.engine = context_engine
        self.history: list[InteractionRecord] = []
        self.active_files: set[str] = set()
        self.token_usage = TokenBudget(128000)
        
    async def handle_request(self, user_input: str) -> str:
        # 1. Load persistent context
        context = await self.engine.retrieve(
            query=user_input,
            options=RetrievalOptions(
                include_semantic=True,
                include_episodic=True,
                include_procedural=True,
                max_tokens=self.token_usage.remaining,
                recency_decay='exponential'
            )
        )
        
        # 2. Check for relevant past decisions
        related_decisions = await self.engine.search_decisions(user_input)
        
        # 3. Build the augmented prompt
        prompt = self._build_prompt(user_input, context, related_decisions)
        
        # 4. Execute with the model
        response = await self._execute_with_tools(prompt)
        
        # 5. Persist the interaction
        await self.engine.ingest(WorkspaceEvent(
            type='interaction',
            input=user_input,
            output=response,
            context_used=context,
            session_id=self.session_id
        ))
        
        # 6. Update working memory
        self.history.append(InteractionRecord(user_input, response))
        
        return response
        
    def _build_prompt(self, user_input, context, decisions) -> str:
        parts = []
        
        # System prompt with learned preferences
        parts.append(self._system_prompt_with_preferences(context.procedural))
        
        # Relevant architectural context
        if context.semantic:
            parts.append("## Project Architecture\n")
            for item in context.semantic.top_items:
                parts.append(f"- {item.summary}")
        
        # Past decisions that might be relevant
        if decisions:
            parts.append("\n## Related Decisions\n")
            for decision in decisions[:5]:
                parts.append(
                    f"- {decision.title}: {decision.rationale} "
                    f"(made {decision.date}, confidence: {decision.confidence})"
                )
        
        # Recent episodic context
        if context.episodic:
            parts.append("\n## Recent Activity\n")
            for event in context.episodic.recent_events:
                parts.append(f"- [{event.type}] {event.summary}")
        
        # Current task
        parts.append(f"\n## Current Request\n{user_input}")
        
        return "\n".join(parts)

class MemoryConsolidator:
    """Runs periodically to keep memory efficient."""
    
    async def consolidate(self, workspace_id: str):
        # Step 1: Identify clusters of related episodic memories
        episodes = await self.store.get_episodic_memories(
            workspace_id, older_than=timedelta(days=7)
        )
        clusters = self._cluster_episodes(episodes)
        
        # Step 2: Summarize each cluster into a higher-level memory
        for cluster in clusters:
            if len(cluster) >= 3:  # Only consolidate meaningful clusters
                summary = await self._summarize_cluster(cluster)
                
                # Create a consolidated memory node
                consolidated = MemoryNode(
                    id=f"consolidated_{hashlib.md5(cluster.id).hexdigest()}",
                    type=NodeType.DECISION,
                    content=summary.text,
                    embedding=self.embedder.encode(summary.text),
                    metadata={
                        'source_episodes': [e.id for e in cluster],
                        'consolidation_date': time.time(),
                        'importance': summary.importance_score
                    }
                )
                
                # Link it in the graph
                await self.store.upsert(consolidated)
                for episode in cluster:
                    await self.store.create_edge(
                        MemoryEdge(
                            source=consolidated.id,
                            target=episode.id,
                            type=EdgeType.COMPOSED_OF
                        )
                    )
        
        # Step 3: Prune low-value episodic memories
        pruned = await self.store.prune_episodic(
            workspace_id,
            criteria=PruneCriteria(
                min_access_count=1,
                max_age_days=30,
                exclude_if_linked_to_consolidated=True
            )
        )
        
        return ConsolidationResult(
            episodes_processed=len(episodes),
            clusters_created=len(clusters),
            memories_pruned=pruned.count,
            storage_saved_mb=pruned.storage_saved
        )

class ToolExecutor:
    def __init__(self, workspace: WorkspaceConfig):
        self.tools = {
            'read_file': FileReadTool(workspace),
            'write_file': FileWriteTool(workspace),
            'edit_file': FileEditTool(workspace),
            'run_command': ShellTool(workspace, allowed_commands=workspace.allowed),
            'run_tests': TestRunnerTool(workspace),
            'git_status': GitStatusTool(),
            'git_diff': GitDiffTool(),
            'search_files': FileSearchTool(workspace),
        }
        
    async def execute(self, tool_call: ToolCall) -> ToolResult:
        tool = self.tools.get(tool_call.name)
        if not tool:
            return ToolResult(error=f"Unknown tool: {tool_call.name}")
        
        # Apply safety checks
        if not self._is_safe(tool_call):
            return ToolResult(error="Action blocked by safety policy")
        
        # Execute with timeout
        try:
            result = await asyncio.wait_for(
                tool.execute(tool_call.arguments),
                timeout=tool_call.timeout or 30.0
            )
            return ToolResult(success=True, output=result)
        except asyncio.TimeoutError:
            return ToolResult(error="Operation timed out")
        except Exception as e:
            return ToolResult(error=str(e))


class AgentLoop:
    """The core agentic reasoning loop with tool use."""
    
    async def execute_task(self, task: str, context: ContextBundle) -> TaskResult:
        messages = self._initialize_messages(task, context)
        max_iterations = 15
        
        for iteration in range(max_iterations):
            # Model decides next action
            response = await self.llm.chat(messages, tools=self.tool_definitions)
            
            if response.has_tool_calls:
                for tool_call in response.tool_calls:
                    result = await self.tool_executor.execute(tool_call)
                    
                    # Add tool result to conversation
                    messages.append(ToolResultMessage(
                        tool_call_id=tool_call.id,
                        result=result
                    ))
                    
                    # Feed result back into context engine
                    await self.context_engine.ingest(WorkspaceEvent(
                        type='tool_execution',
                        tool=tool_call.name,
                        arguments=tool_call.arguments,
                        result=result
                    ))
            elif response.is_final_answer:
                return TaskResult(
                    answer=response.content,
                    iterations=iteration + 1,
                    tools_used=self._count_tools_used(messages)
                )
            
            messages.append(response)
        
        return TaskResult(error="Maximum iterations reached")

class SafetyPolicy:
    def __init__(self, workspace: WorkspaceConfig):
        self.allowed_paths = workspace.allowed_paths
        self.denied_paths = workspace.denied_paths
        self.max_file_size = workspace.max_file_size_mb
        self.require_confirmation = workspace.actions_requiring_confirmation
        
    def evaluate(self, tool_call: ToolCall) -> SafetyDecision:
        # Path safety
        if tool_call.name in ('write_file', 'edit_file'):
            target = tool_call.arguments.get('path', '')
            if any(target.startswith(d) for d in self.denied_paths):
                return SafetyDecision(block=True, reason="Path is denied")
            if not any(target.startswith(a) for a in self.allowed_paths):
                return SafetyDecision(block=True, reason="Path not in workspace")
        
        # Command safety
        if tool_call.name == 'run_command':
            cmd = tool_call.arguments.get('command', '')
            if self._is_destructive(cmd):
                return SafetyDecision(
                    block=False,
                    require_confirmation=True,
                    reason="Potentially destructive command"
                )
        
        return SafetyDecision(block=False)

class WorkspaceObservability:
    def __init__(self):
        self.metrics = {
            'context_hit_rate': Counter(),     # % of queries that find relevant context
            'preference_accuracy': Counter(),  # % of suggestions matching learned prefs
            'iteration_count': Histogram(),    # Tool iterations per task
            'token_efficiency': Histogram(),   # Useful tokens / total tokens
            'memory_utilization': Gauge(),     # Storage usage vs capacity
            'consolidation_ratio': Gauge(),    # Memories consolidated / total
        }
        
    def track_interaction(self, interaction: InteractionRecord):
        self.metrics['iteration_count'].observe(interaction.iterations)
        
        if interaction.context_relevance_score > 0.5:
            self.metrics['context_hit_rate'].inc()
        else:
            self.metrics['context_hit_rate'].inc()  # Still count as a query
        
        self.metrics['token_efficiency'].observe(
            interaction.useful_tokens / interaction.total_tokens
        )

interface ContextNode {
  id: string; // UUID
  timestamp: number; // Unix epoch
  artifactHash: string; // SHA-256 of the associated code/test output
  agentDecision: {
    reasoning: string;
    confidenceScore: number; // 0.0 to 1.0
    alternativesConsidered: string[];
  };
  dependencies: string[]; // IDs of parent nodes
  feedbackLoop?: {
    outcome: 'success' | 'failure' | 'timeout';
    correctiveAction?: string;
  };
}

class AgenticContextManager:
    def __init__(self, graph_store):
        self.graph_store = graph_store
        self.embedding_model = load_model('sentence-transformer')

    def get_relevant_context(self, current_task: str, k: int = 5) -> List[ContextNode]:
        """
        Retrieves the most relevant historical context nodes for a given task.
        Uses semantic similarity to find past scenarios that were similar to the current problem.
        """
        current_embedding = self.embedding_model.encode(current_task)
        
        # Query the graph store for nodes with high similarity
        # We prioritize nodes with 'success' outcomes and high confidence scores
        candidates = self.graph_store.query(
            embedding=current_embedding,
            filter={
                'outcome': 'success',
                'confidenceScore': {'$gt': 0.8}
            },
            limit=k
        )
        
        return candidates

    def record_outcome(self, node_id: str, outcome: str, feedback: str = None):
        """
        Updates the graph with the result of an agent's action.
        If the outcome is a failure, this triggers a 'corrective action' flag.
        """
        node = self.graph_store.get_node(node_id)
        if outcome == 'failure':
            # The system learns from mistakes by marking this path as suboptimal
            node.feedbackLoop = {
                'outcome': outcome,
                'correctiveAction': feedback or "Review assumptions regarding external dependencies"
            }
        else:
            node.feedbackLoop = {'outcome': outcome}
            
        self.graph_store.update_node(node)