Major updates and improvements to BZZZ system

- Updated configuration and deployment files
- Improved system architecture and components
- Enhanced documentation and testing
- Fixed various issues and added new features

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
anthonyrawlins
2025-09-17 18:06:57 +10:00
parent 4e6140de03
commit f5f96ba505
71 changed files with 664 additions and 3823 deletions

View File

Before

Width:  |  Height:  |  Size: 39 KiB

After

Width:  |  Height:  |  Size: 39 KiB

View File

Before

Width:  |  Height:  |  Size: 117 KiB

After

Width:  |  Height:  |  Size: 117 KiB

View File

@@ -0,0 +1,71 @@
FROM ubuntu:24.04
RUN apt-get update && apt-get install -y --no-install-recommends \
ca-certificates tzdata curl locales gettext-base systemd systemd-sysv && \
rm -rf /var/lib/apt/lists/*
# Configure systemd for container use
RUN cd /lib/systemd/system/sysinit.target.wants/ && \
ls | grep -v systemd-tmpfiles-setup | xargs rm -f && \
rm -f /lib/systemd/system/multi-user.target.wants/* && \
rm -f /etc/systemd/system/*.wants/* && \
rm -f /lib/systemd/system/local-fs.target.wants/* && \
rm -f /lib/systemd/system/sockets.target.wants/*udev* && \
rm -f /lib/systemd/system/sockets.target.wants/*initctl* && \
rm -f /lib/systemd/system/basic.target.wants/* && \
rm -f /lib/systemd/system/anaconda.target.wants/*
# Create bzzz directories
RUN mkdir -p /opt/bzzz /opt/bzzz/.bzzz /etc/systemd/system
# BZZZ binary
COPY ./build/bzzz /opt/bzzz/bzzz
RUN chmod +x /opt/bzzz/bzzz
# Config template
COPY ./config.yml.tmpl /opt/bzzz/.bzzz/config.yml.tmpl
# Create systemd service file
RUN cat > /etc/systemd/system/bzzz.service << 'EOF'
[Unit]
Description=BZZZ P2P Task Coordination System
After=network.target
Wants=network-online.target
[Service]
Type=simple
User=root
WorkingDirectory=/opt/bzzz
ExecStart=/opt/bzzz/bzzz -config /opt/bzzz/.bzzz/config.yml
Restart=always
RestartSec=10
StandardOutput=journal
StandardError=journal
SyslogIdentifier=bzzz
[Install]
WantedBy=multi-user.target
EOF
# Enable the service
RUN systemctl enable bzzz.service
# Create startup script that renders config and starts systemd
RUN cat > /opt/bzzz/startup.sh << 'EOF'
#!/bin/bash
set -e
# Render config from template
envsubst < /opt/bzzz/.bzzz/config.yml.tmpl > /opt/bzzz/.bzzz/config.yml
# Start systemd
exec /lib/systemd/systemd
EOF
RUN chmod +x /opt/bzzz/startup.sh
# Working directory
WORKDIR /opt/bzzz
# Use systemd as init system
ENTRYPOINT ["/opt/bzzz/startup.sh"]

View File

@@ -0,0 +1,69 @@
# Minimal BZZZ Docker container without systemd
# Uses multi-stage build for smaller final image
FROM golang:1.21-alpine AS builder
WORKDIR /build
COPY go.mod go.sum ./
RUN go mod download
COPY . .
RUN CGO_ENABLED=0 GOOS=linux go build -ldflags='-w -s -extldflags "-static"' -o bzzz .
# Final minimal image
FROM alpine:3.18
# Install only essential packages
RUN apk --no-cache add \
ca-certificates \
tzdata \
curl
# Create non-root user for security
RUN addgroup -g 1000 bzzz && \
adduser -u 1000 -G bzzz -s /bin/sh -D bzzz
# Create required directories
RUN mkdir -p /app/data /app/config /app/logs && \
chown -R bzzz:bzzz /app
# Copy binary from builder stage
COPY --from=builder /build/bzzz /app/bzzz
RUN chmod +x /app/bzzz
# Copy config template
COPY dockerize/config.minimal.yml.tmpl /app/config/config.yml.tmpl
# Create entrypoint script that handles config generation
RUN cat > /app/entrypoint.sh << 'EOF'
#!/bin/sh
set -e
# Generate config from template if it doesn't exist
if [ ! -f /app/config/config.yml ]; then
echo "🔧 Generating configuration from template..."
envsubst < /app/config/config.yml.tmpl > /app/config/config.yml
fi
# Ensure proper ownership
chown -R bzzz:bzzz /app/data /app/config /app/logs
echo "🚀 Starting BZZZ..."
exec "$@"
EOF
RUN chmod +x /app/entrypoint.sh
# Switch to non-root user
USER bzzz
WORKDIR /app
# Health check
HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \
CMD curl -f http://localhost:8081/health || exit 1
# Expose ports
EXPOSE 8080 8081 9000-9100
# Set entrypoint and default command
ENTRYPOINT ["/app/entrypoint.sh"]
CMD ["/app/bzzz", "--config", "/app/config/config.yml"]

View File

@@ -0,0 +1,29 @@
#!/bin/bash
set -e
# Build script for minimal BZZZ container
echo "🐳 Building minimal BZZZ container..."
# Set build context to parent directory (BZZZ root)
cd "$(dirname "$0")/.."
# Build the minimal container
docker build -f dockerize/Dockerfile.minimal -t bzzz:minimal .
echo "✅ BZZZ minimal container built successfully!"
echo ""
echo "Usage:"
echo " # Start with default configuration:"
echo " docker-compose -f dockerize/docker-compose.minimal.yml up -d"
echo ""
echo " # Start with custom environment:"
echo " cp dockerize/bzzz.minimal.env.example dockerize/bzzz.minimal.env"
echo " # Edit dockerize/bzzz.minimal.env with your settings"
echo " docker-compose -f dockerize/docker-compose.minimal.yml --env-file dockerize/bzzz.minimal.env up -d"
echo ""
echo " # Check logs:"
echo " docker-compose -f dockerize/docker-compose.minimal.yml logs -f"
echo ""
echo " # Check health:"
echo " curl http://localhost:8081/health"

View File

@@ -0,0 +1,14 @@
# Human-readable hint in rendered config header
NODE_HINT=swarm
# Unique id (defaults to $HOSTNAME if unset)
# AGENT_ID=
# Ollama endpoint (per-node host)
OLLAMA_BASE_URL=http://host.docker.internal:11434
# Log level: debug|info|warn|error
LOG_LEVEL=info
# UCXL storage path inside the container (persisted under /var/lib/bzzz)
UCXL_DIR=/var/lib/bzzz/ucxl

View File

@@ -0,0 +1,32 @@
# BZZZ Minimal Container Configuration
# Copy this file to bzzz.minimal.env and customize as needed
# Basic Agent Configuration
BZZZ_AGENT_ID=bzzz-docker-01
BZZZ_SPECIALIZATION=general_developer
BZZZ_MAX_TASKS=3
# Network Ports (adjust if ports are already in use)
BZZZ_P2P_PORT=9000
BZZZ_API_PORT=8080
BZZZ_HEALTH_PORT=8081
# Logging
LOG_LEVEL=info
# DEBUG=1 # Uncomment to enable debug logging
# DHT and P2P Settings
BZZZ_DHT_ENABLED=true
# BZZZ_BOOTSTRAP_PEERS= # Comma-separated list of bootstrap peers
# AI Configuration
OLLAMA_ENDPOINT=http://host.docker.internal:11434
# Licensing (if required)
# LICENSE_EMAIL=your.email@example.com
# LICENSE_KEY=your-license-key-here
CLUSTER_ID=docker-cluster
# Optional: Override default resource limits in docker-compose.minimal.yml
# MEMORY_LIMIT=1G
# CPU_LIMIT=1.0

View File

@@ -0,0 +1,91 @@
# BZZZ Configuration for Container Deployment
# Environment variables will be substituted at runtime
agent:
id: "${BZZZ_AGENT_ID}"
specialization: "${BZZZ_SPECIALIZATION}"
max_tasks: ${BZZZ_MAX_TASKS}
capabilities:
- "general_development"
- "task_coordination"
- "p2p_collaboration"
models:
- "llama3.1:8b"
- "codellama:7b"
role: "" # Will be auto-assigned based on specialization
expertise: []
reports_to: ""
network:
p2p:
listen_port: ${BZZZ_P2P_PORT}
bind_address: "0.0.0.0"
api:
port: ${BZZZ_API_PORT}
bind_address: "0.0.0.0"
health:
port: ${BZZZ_HEALTH_PORT}
bind_address: "0.0.0.0"
# DHT configuration for peer discovery
v2:
dht:
enabled: ${BZZZ_DHT_ENABLED}
bootstrap_peers: [] # Will be populated from BZZZ_BOOTSTRAP_PEERS env var
# AI configuration
ai:
ollama:
endpoint: "${OLLAMA_ENDPOINT}"
timeout: "30s"
# UCXL protocol configuration
ucxl:
enabled: true
server:
enabled: true
port: 8082
base_path: ""
storage:
directory: "/tmp/bzzz-ucxi-storage"
resolution:
cache_ttl: "1h"
# Licensing configuration (if required)
license:
email: "${LICENSE_EMAIL}"
license_key: "${LICENSE_KEY}"
cluster_id: "${CLUSTER_ID}"
organization_name: ""
kaching_url: "https://kaching.chorus.services"
is_active: false
grace_period_hours: 72
license_type: ""
max_nodes: 1
# Binary type for specialized behavior
binary_type: "agent"
# Repository integration (disabled in container mode)
repository:
provider: ""
url: ""
token: ""
webhook_url: ""
# Security settings optimized for containers
security:
enable_auth: false
auth_token: ""
# Storage paths for container environment
storage:
data_directory: "/app/data"
config_directory: "/app/config"
log_directory: "/app/logs"
# Logging configuration for containers (stdout/stderr)
logging:
level: "${LOG_LEVEL}"
format: "structured" # Better for container log collection
output: "stdout" # Force stdout for container compatibility

View File

@@ -0,0 +1,136 @@
# BZZZ Configuration for ${NODE_HINT:-container}
whoosh_api:
base_url: "https://whoosh.home.deepblack.cloud"
api_key: ""
timeout: 30s
retry_count: 3
agent:
id: "${AGENT_ID:-${HOSTNAME}}"
capabilities: ["general"]
poll_interval: 30s
max_tasks: 2
models: []
specialization: ""
model_selection_webhook: ""
default_reasoning_model: ""
sandbox_image: ""
role: ""
system_prompt: ""
reports_to: []
expertise: []
deliverables: []
collaboration:
preferred_message_types: []
auto_subscribe_to_roles: []
auto_subscribe_to_expertise: []
response_timeout_seconds: 0
max_collaboration_depth: 0
escalation_threshold: 0
custom_topic_subscriptions: []
github:
token_file: ""
user_agent: "BZZZ-Agent/1.0"
timeout: 30s
rate_limit: true
assignee: ""
p2p:
service_tag: "bzzz-peer-discovery"
bzzz_topic: "bzzz/coordination/v1"
hmmm_topic: "hmmm/meta-discussion/v1"
discovery_timeout: 10s
escalation_webhook: ""
escalation_keywords: []
conversation_limit: 10
logging:
level: "${LOG_LEVEL:-info}"
format: "text"
output: "stdout"
structured: false
slurp:
enabled: false
base_url: ""
api_key: ""
timeout: 30s
retry_count: 3
max_concurrent_requests: 10
request_queue_size: 100
v2:
enabled: false
protocol_version: "2.0.0"
uri_resolution:
cache_ttl: 5m0s
max_peers_per_result: 5
default_strategy: "best_match"
resolution_timeout: 30s
dht:
enabled: false
bootstrap_peers: []
mode: "auto"
protocol_prefix: "/bzzz"
bootstrap_timeout: 30s
discovery_interval: 1m0s
auto_bootstrap: false
semantic_addressing:
enable_wildcards: true
default_agent: "any"
default_role: "any"
default_project: "any"
enable_role_hierarchy: true
feature_flags:
uri_protocol: false
semantic_addressing: false
dht_discovery: false
advanced_resolution: false
ucxl:
enabled: false
server:
port: 8081
base_path: "/bzzz"
enabled: false
resolution:
cache_ttl: 5m0s
enable_wildcards: true
max_results: 50
storage:
type: "filesystem"
directory: "${UCXL_DIR:/var/lib/bzzz/ucxl}"
max_size: 104857600
p2p_integration:
enable_announcement: false
enable_discovery: false
announcement_topic: "bzzz/ucxl/announcement/v1"
discovery_timeout: 30s
security:
admin_key_shares:
threshold: 3
total_shares: 5
election_config:
heartbeat_timeout: 5s
discovery_timeout: 30s
election_timeout: 15s
max_discovery_attempts: 6
discovery_backoff: 5s
minimum_quorum: 3
consensus_algorithm: "raft"
split_brain_detection: true
conflict_resolution: "highest_uptime"
key_rotation_days: 90
audit_logging: false
audit_path: ""
ai:
ollama:
endpoint: "${OLLAMA_BASE_URL:-http://host.docker.internal:11434}"
timeout: 30s
models: []
openai:
api_key: ""
endpoint: "https://api.openai.com/v1"

View File

@@ -0,0 +1,112 @@
version: "3.9"
services:
bzzz-minimal:
image: bzzz:minimal
build:
context: ..
dockerfile: dockerize/Dockerfile.minimal
environment:
# Basic BZZZ configuration
- BZZZ_AGENT_ID=${BZZZ_AGENT_ID:-bzzz-docker-01}
- BZZZ_SPECIALIZATION=${BZZZ_SPECIALIZATION:-general_developer}
- BZZZ_MAX_TASKS=${BZZZ_MAX_TASKS:-3}
# Network configuration
- BZZZ_P2P_PORT=${BZZZ_P2P_PORT:-9000}
- BZZZ_API_PORT=${BZZZ_API_PORT:-8080}
- BZZZ_HEALTH_PORT=${BZZZ_HEALTH_PORT:-8081}
# Logging configuration
- LOG_LEVEL=${LOG_LEVEL:-info}
- DEBUG=${DEBUG:-}
# DHT and P2P settings
- BZZZ_DHT_ENABLED=${BZZZ_DHT_ENABLED:-true}
- BZZZ_BOOTSTRAP_PEERS=${BZZZ_BOOTSTRAP_PEERS:-}
# AI/Ollama configuration
- OLLAMA_ENDPOINT=${OLLAMA_ENDPOINT:-http://host.docker.internal:11434}
# Licensing (if required)
- LICENSE_EMAIL=${LICENSE_EMAIL:-}
- LICENSE_KEY=${LICENSE_KEY:-}
- CLUSTER_ID=${CLUSTER_ID:-docker-cluster}
# Persist data across container restarts
volumes:
- bzzz_data:/app/data
- bzzz_config:/app/config
- type: bind
source: /tmp/bzzz-ucxi-storage
target: /tmp/bzzz-ucxi-storage
- type: bind
source: /tmp/hcfs-workspaces
target: /tmp/hcfs-workspaces
# Network ports
ports:
- "${BZZZ_API_PORT:-8080}:8080" # HTTP API
- "${BZZZ_HEALTH_PORT:-8081}:8081" # Health check
- "${BZZZ_P2P_PORT:-9000}:9000" # P2P communication
# Container resource limits
deploy:
mode: replicated
replicas: 1
update_config:
order: start-first
parallelism: 1
failure_action: rollback
restart_policy:
condition: on-failure
delay: 10s
max_attempts: 3
resources:
limits:
cpus: "1.0"
memory: 1G
reservations:
cpus: "0.25"
memory: 256M
# Network configuration
networks:
- bzzz_net
# Host resolution for connecting to host services
extra_hosts:
- "host.docker.internal:host-gateway"
# Logging configuration for container runtime
logging:
driver: "json-file"
options:
max-size: "10m"
max-file: "3"
labels: "service=bzzz"
# Health check
healthcheck:
test: ["CMD", "curl", "-f", "http://localhost:8081/health"]
interval: 30s
timeout: 10s
retries: 3
start_period: 10s
# Named volumes for persistence
volumes:
bzzz_data:
driver: local
bzzz_config:
driver: local
# Network for BZZZ communication
networks:
bzzz_net:
driver: overlay
attachable: true
ipam:
driver: default
config:
- subnet: 10.200.0.0/24

View File

@@ -0,0 +1,62 @@
version: "3.9"
services:
bzzz:
image: bzzz:latest
env_file:
- bzzz.env
# Persist identity/state per node
volumes:
- type: bind
source: /var/lib/bzzz
target: /var/lib/bzzz
- type: bind
source: /tmp/bzzz-ucxl-storage
target: /tmp/bzzz-ucxl-storage
- type: bind
source: /tmp/bzzz-ucxi-storage
target: /tmp/bzzz-ucxi-storage
- type: bind
source: /tmp/hcfs-workspaces
target: /tmp/hcfs-workspaces
# If you later enable ucxl.server.enabled: true and need to expose it:
ports:
- target: 8081
published: 8081
protocol: tcp
mode: host
deploy:
mode: replicated
replicas: 3
update_config:
order: start-first
parallelism: 1
failure_action: rollback
restart_policy:
condition: on-failure
resources:
limits:
cpus: "1.0"
memory: 2G
reservations:
cpus: "0.25"
memory: 512M
placement:
preferences:
- spread: node.id
networks:
- bzzz_net
# Lets the container resolve the node's host at host.docker.internal
extra_hosts:
- "host.docker.internal:host-gateway"
networks:
bzzz_net:
driver: overlay
attachable: true
# driver_opts:
# encrypted: "true"

View File

@@ -0,0 +1,48 @@
package main
import (
"fmt"
"os"
"time"
)
// ContainerLogger provides structured logging for containers
// All output goes to stdout/stderr for container runtime collection
type ContainerLogger struct {
name string
}
// NewContainerLogger creates a new container-friendly logger
func NewContainerLogger(name string) *ContainerLogger {
return &ContainerLogger{name: name}
}
// Info logs informational messages to stdout
func (l *ContainerLogger) Info(msg string, args ...interface{}) {
timestamp := time.Now().UTC().Format("2006-01-02T15:04:05.000Z")
logMsg := fmt.Sprintf(msg, args...)
fmt.Fprintf(os.Stdout, "[%s] [INFO] [%s] %s\n", timestamp, l.name, logMsg)
}
// Warn logs warning messages to stdout
func (l *ContainerLogger) Warn(msg string, args ...interface{}) {
timestamp := time.Now().UTC().Format("2006-01-02T15:04:05.000Z")
logMsg := fmt.Sprintf(msg, args...)
fmt.Fprintf(os.Stdout, "[%s] [WARN] [%s] %s\n", timestamp, l.name, logMsg)
}
// Error logs error messages to stderr
func (l *ContainerLogger) Error(msg string, args ...interface{}) {
timestamp := time.Now().UTC().Format("2006-01-02T15:04:05.000Z")
logMsg := fmt.Sprintf(msg, args...)
fmt.Fprintf(os.Stderr, "[%s] [ERROR] [%s] %s\n", timestamp, l.name, logMsg)
}
// Debug logs debug messages to stdout (only if DEBUG env var is set)
func (l *ContainerLogger) Debug(msg string, args ...interface{}) {
if os.Getenv("DEBUG") != "" {
timestamp := time.Now().UTC().Format("2006-01-02T15:04:05.000Z")
logMsg := fmt.Sprintf(msg, args...)
fmt.Fprintf(os.Stdout, "[%s] [DEBUG] [%s] %s\n", timestamp, l.name, logMsg)
}
}

View File

@@ -1,330 +0,0 @@
# SLURP: Storage, Logic, Understanding, Retrieval, Processing
SLURP is the contextual intelligence system for BZZZ, implementing a sophisticated Leader-coordinated architecture where only the elected BZZZ Leader generates role-specific encrypted context for AI agents.
## Overview
SLURP provides the "WHY" behind every UCXL address while maintaining strict need-to-know security boundaries through:
- **Leader-Only Context Generation**: Single source of truth for contextual intelligence
- **Role-Based Encryption**: Context encrypted per AI agent role with need-to-know access
- **Bounded Hierarchical Context**: CSS-like cascading context inheritance with configurable depth limits
- **Decision-Hop Temporal Analysis**: Track related decisions by decision distance, not chronological time
- **Project-Aligned Intelligence**: Context generation aligned with project goals and team dynamics
## Architecture
SLURP operates as five integrated subsystems:
```
┌─────────────────────────────────────┐
│ PROJECT ALIGNMENT │ Strategic goal integration
├─────────────────────────────────────┤
│ CONTEXTUAL INTELLIGENCE │ Advanced analysis & insights
├─────────────────────────────────────┤
│ CONTEXT INTELLIGENCE │ Core context generation
├─────────────────────────────────────┤
│ TEMPORAL ANALYSIS │ Decision-hop tracking
├─────────────────────────────────────┤
│ STORAGE ARCHITECTURE │ Distributed encrypted storage
└─────────────────────────────────────┘
```
## Core Subsystems
### 1. Context Intelligence (`context-intelligence/`)
**Purpose**: Core context generation engine implementing CSS-like cascading inheritance
**Key Features**:
- Hierarchical context metadata generation
- Bounded depth traversal (configurable limits)
- Space-efficient storage (85%+ space savings)
- Global context support for system-wide applicable metadata
- Role-aware context generation
**Main Components**:
- `cascading_metadata_generator.py` - CSS-like context inheritance system
- `context_resolver.py` - Efficient hierarchical context resolution
- `bounded_context_demo.py` - Complete demonstration system
### 2. Storage Architecture (`storage/`)
**Purpose**: Distributed encrypted storage for contextual intelligence data
**Key Features**:
- Role-based encryption with need-to-know access
- DHT integration for network-wide distribution
- Temporal versioning of context evolution
- Efficient querying and indexing
- Leader-coordinated consistency
**Architecture Layers**:
- Application Layer → Role-Based Encryption → Context Serialization → DHT → Network Transport
### 3. Temporal Analysis (`temporal/`)
**Purpose**: Decision-hop based temporal tracking (not time-based)
**Key Features**:
- Decision influence graph construction
- Context evolution through decision relationships
- Temporal queries: "context as it was at decision point X"
- Decision genealogy and rationale tracking
- Confidence evolution monitoring
**Main Component**:
- `temporal_context_system.py` - Complete temporal graph implementation
### 4. Contextual Intelligence (`intelligence/`)
**Purpose**: Advanced analysis and insight generation
**Key Features**:
- File purpose analysis with architectural pattern detection
- Cross-component relationship mapping
- Role-specific insight generation
- Architectural decision extraction
- Impact analysis and change prediction
**Analysis Capabilities**:
- Code pattern recognition across multiple languages
- Architectural decision documentation
- Component relationship analysis
- Performance and security insight generation
### 5. Project Alignment (`alignment/`)
**Purpose**: Ensure context aligns with project goals and team objectives
**Key Features**:
- Mission-context integration
- Team goal awareness
- Strategic objective mapping
- Dynamic priority adjustment
- Success metrics tracking
**Alignment Layers**:
- Success Metrics → Priority Adjustment → Objective Mapping → Team Goals → Mission Integration
## BZZZ Leader Integration
SLURP operates under BZZZ's Leader election system:
### Leader Responsibilities
- **Context Authority**: Only Leader generates contextual intelligence
- **Quality Control**: Ensures consistent, high-quality context across system
- **Role-Based Distribution**: Encrypts and distributes context per agent role
- **Priority Coordination**: Coordinates context generation priorities
### Role-Based Access Control
| Role | Context Access | Encryption Level | Scope |
|------|----------------|------------------|--------|
| Senior Architect | Architecture decisions, system design, technical debt | High | System-wide |
| Frontend Developer | UI/UX decisions, component relationships, user flows | Medium | Frontend scope |
| Backend Developer | API design, data flow, service architecture | Medium | Backend scope |
| DevOps Engineer | Deployment config, infrastructure decisions | High | Infrastructure |
| Project Manager (Leader) | All context for coordination | Highest | Global |
### Security Model
- **Multi-layer encryption**: Base context + role-specific overlays
- **Key derivation**: From role definitions and Shamir shares
- **Access logging**: Complete audit trail of context access
- **Context compartmentalization**: Prevents cross-role information leakage
## Key Innovations
### 1. CSS-Like Context Inheritance
```
/project/ # Root context: "BZZZ distributed system"
├── src/ # Inherits + adds: "Source code implementation"
│ ├── api/ # Inherits + adds: "API layer functionality"
│ │ └── handlers.rs # Inherits all above + adds: "HTTP request handling"
│ └── core/ # Inherits src + adds: "Core business logic"
└── tests/ # Inherits root + adds: "Quality assurance"
```
**Benefits**:
- 85%+ space savings vs traditional metadata
- Automatic context inheritance with overrides
- Configurable depth limits prevent excessive traversal
- Global contexts apply system-wide
### 2. Decision-Hop Temporal Analysis
Unlike traditional time-based systems, SLURP tracks context evolution through **decision relationships**:
```
Decision A (Architecture change)
↓ influences
Decision B (API refactoring) ← 1 hop from A
↓ influences
Decision C (Frontend update) ← 2 hops from A
```
**Benefits**:
- Conceptual relevance like RAG systems
- Find related decisions by influence, not time
- Track decision genealogy and rationale
- Understand how changes propagate through system
### 3. Leader-Coordinated Intelligence
- **Single Source of Truth**: Prevents conflicting context from multiple sources
- **Quality Assurance**: Centralized generation ensures consistent context quality
- **Role-Based Security**: Each agent gets exactly the context they need
- **Efficient Resource Usage**: Eliminates redundant context generation
## Usage Examples
### Basic Context Resolution
```python
from slurp.context_intelligence.context_resolver import CascadingContextResolver
# Initialize resolver with bounded depth
resolver = CascadingContextResolver(metadata_dir, max_hierarchy_depth=10)
# Resolve context for a UCXL address
context = resolver.resolve("ucxl://any:any@BZZZ:RUSTLE-testing/src/main.rs")
print(f"Purpose: {context.purpose}")
print(f"Technologies: {context.technologies}")
print(f"Insights: {context.insights}")
```
### Temporal Decision Analysis
```python
from slurp.temporal.temporal_context_system import TemporalContextGraph
# Initialize temporal graph
temporal_graph = TemporalContextGraph(metadata_base, "BZZZ")
# Find decisions within 3 hops of a component
related_decisions = temporal_graph.find_related_decisions(
"ucxl://any:any@BZZZ:RUSTLE-testing/src/main.rs",
max_hops=3
)
# Get decision timeline (by decision influence, not time)
timeline = temporal_graph.get_decision_timeline(
ucxl_address,
include_related=True
)
```
### Bounded Context System
```python
from slurp.context_intelligence.bounded_context_demo import BoundedContextSystem
# Initialize complete bounded system
bounded_system = BoundedContextSystem(
project_path="/home/tony/chorus/project-queues/active/BZZZ",
metadata_base="/tmp/slurp-metadata",
max_hierarchy_depth=8
)
# Generate hierarchical metadata with bounded traversal
bounded_system.initialize_system()
results = bounded_system.generate_hierarchical_metadata()
```
## Performance Characteristics
### Space Efficiency
- **85%+ space savings** through intelligent context inheritance
- **10:1 compression ratio** through deduplication
- **Minimal network bandwidth** through differential updates
- **Sub-second lookups** with indexing and caching
### Processing Efficiency
- **O(log n) context resolution** with bounded depth limits
- **Parallel analysis** across multiple components
- **Lazy loading** of detailed context when needed
- **Efficient caching** with >90% hit rates for frequent access
### Security Performance
- **Fast role-based decryption** using pre-derived keys
- **Minimal encryption overhead** through layered approach
- **Complete audit trails** without performance impact
- **Scalable key management** through Shamir's Secret Sharing
## Configuration
SLURP can be configured through environment variables and configuration files:
```yaml
# slurp_config.yaml
slurp:
context_intelligence:
max_hierarchy_depth: 10
global_context_enabled: true
cache_size: 1000
temporal_analysis:
max_decision_hops: 5
staleness_threshold: 0.5
confidence_decay_rate: 0.1
storage:
encryption_enabled: true
replication_factor: 3
cache_ttl: 3600
alignment:
mission_weight: 0.4
team_goals_weight: 0.3
strategic_objectives_weight: 0.3
```
## Monitoring and Observability
SLURP provides comprehensive monitoring:
- **Context Generation Metrics**: Generation time, quality scores, cache hit rates
- **Temporal Analysis Metrics**: Decision graph size, traversal performance, staleness rates
- **Storage Metrics**: Storage utilization, replication health, query performance
- **Security Metrics**: Access patterns, encryption performance, audit log health
- **Alignment Metrics**: Goal achievement rates, team satisfaction, context relevance
## Future Roadmap
### Phase 1: Foundation (Complete)
- ✅ CSS-like context inheritance system
- ✅ Bounded hierarchy traversal
- ✅ Decision-hop temporal analysis
- ✅ Role-based security architecture
### Phase 2: Intelligence Enhancement (In Progress)
- 🔄 RAG integration for enhanced context analysis
- 🔄 ML-based context quality scoring
- 🔄 Advanced architectural decision extraction
- 🔄 Cross-component relationship mapping
### Phase 3: Advanced Features (Planned)
- 📋 Real-time context updates via WebSocket
- 📋 Predictive context generation based on patterns
- 📋 Cross-project context learning
- 📋 Advanced visualization dashboards
### Phase 4: AI Integration (Future)
- 📋 Large Language Model integration
- 📋 Natural language context querying
- 📋 Automated context validation
- 📋 Self-improving context generation
## Contributing
SLURP is designed for extensibility. To contribute:
1. Each subsystem has clear interfaces and documentation
2. Follow the role-based security model in all components
3. Maintain bounded resource usage (depth limits, cache sizes)
4. Ensure Leader-coordination compatibility
5. Add comprehensive tests for new features
## Integration Points
SLURP integrates with existing BZZZ systems:
- **DHT Integration**: Uses existing BZZZ DHT for distributed storage
- **Leader Election**: Extends current election system for context coordination
- **Crypto Infrastructure**: Leverages existing Shamir's Secret Sharing
- **UCXL Protocol**: Native support for UCXL address resolution
- **MCP Server**: Provides context through BZZZ MCP server endpoints
---
*SLURP represents a revolutionary approach to contextual intelligence in distributed systems, providing the "WHY" behind every component while maintaining security, performance, and alignment with project goals.*

View File

@@ -1,455 +0,0 @@
# SLURP Project Goal Alignment System
The Project Goal Alignment System ensures that contextual intelligence generation and distribution aligns with current project objectives, team goals, and strategic priorities within the BZZZ ecosystem.
## Purpose
This module provides:
- **Mission-Context Integration**: Align context generation with project mission
- **Team Goal Awareness**: Incorporate team objectives into context generation
- **Strategic Objective Mapping**: Map context relevance to strategic objectives
- **Dynamic Priority Adjustment**: Adjust context focus based on changing priorities
- **Success Metrics Tracking**: Monitor alignment effectiveness over time
## Architecture
The Alignment System operates as a goal-aware overlay on all SLURP components:
```
┌─────────────────────────────────────┐
│ Success Metrics Tracking │
├─────────────────────────────────────┤
│ Dynamic Priority Adjustment │
├─────────────────────────────────────┤
│ Strategic Objective Mapping │
├─────────────────────────────────────┤
│ Team Goal Awareness │
├─────────────────────────────────────┤
│ Mission-Context Integration │
├─────────────────────────────────────┤
│ Goal Definition Layer │
└─────────────────────────────────────┘
```
## Core Components
### Goal Definition System
Defines and manages project goals at multiple levels:
#### Goal Hierarchy
```python
@dataclass
class ProjectGoal:
goal_id: str
title: str
description: str
level: GoalLevel # STRATEGIC, TACTICAL, OPERATIONAL
priority: Priority # CRITICAL, HIGH, MEDIUM, LOW
status: GoalStatus # ACTIVE, PAUSED, COMPLETED, CANCELLED
# Temporal aspects
created_at: datetime
target_date: Optional[datetime]
completed_at: Optional[datetime]
# Relationships
parent_goals: List[str] # Higher-level goals this supports
child_goals: List[str] # Lower-level goals that support this
related_goals: List[str] # Peer goals that interact with this
# Metrics
success_criteria: List[str]
progress_indicators: List[str]
current_progress: float # 0.0 to 1.0
# Context relevance
relevant_components: List[str] # UCXL addresses relevant to this goal
context_keywords: List[str] # Keywords that indicate relevance
technology_focus: List[str] # Technologies relevant to this goal
```
#### Goal Categories
**Strategic Goals** (3-12 month horizon)
- System architecture evolution
- Technology stack modernization
- Performance and scalability targets
- Security and compliance objectives
**Tactical Goals** (1-3 month horizon)
- Feature development milestones
- Technical debt reduction
- Infrastructure improvements
- Team capability building
**Operational Goals** (1-4 week horizon)
- Bug fixes and stability
- Code quality improvements
- Documentation updates
- Testing coverage increases
### Mission-Context Integration Engine
Integrates project mission and vision into context generation:
#### Mission Analysis
```python
@dataclass
class ProjectMission:
mission_statement: str
vision_statement: str
core_values: List[str]
success_principles: List[str]
# Technical mission aspects
architectural_principles: List[str]
quality_attributes: List[str] # Performance, security, maintainability
technology_philosophy: str # Innovation vs stability balance
# Context generation guidance
context_priorities: Dict[str, float] # What to emphasize in context
insight_focus_areas: List[str] # What insights to prioritize
role_alignment_weights: Dict[AgentRole, float] # Role importance weighting
```
#### Mission-Driven Context Weighting
```python
def apply_mission_alignment(context: ContextNode, mission: ProjectMission) -> ContextNode:
# Boost insights that align with mission
aligned_insights = []
for insight in context.insights:
relevance_score = calculate_mission_relevance(insight, mission)
if relevance_score > 0.7:
aligned_insights.append(f"[MISSION-CRITICAL] {insight}")
elif relevance_score > 0.4:
aligned_insights.append(f"[MISSION-ALIGNED] {insight}")
else:
aligned_insights.append(insight)
context.insights = aligned_insights
# Adjust technology emphasis based on mission technology philosophy
context.technologies = reweight_technologies(
context.technologies,
mission.technology_philosophy
)
return context
```
### Team Goal Awareness System
Incorporates team-specific goals and dynamics into context generation:
#### Team Structure Modeling
```python
@dataclass
class TeamStructure:
team_id: str
team_name: str
team_mission: str
# Team composition
team_members: List[TeamMember]
team_roles: List[AgentRole]
expertise_areas: List[str]
# Team goals and priorities
current_goals: List[str] # Goal IDs team is working on
priority_weights: Dict[str, float] # How much team prioritizes each goal
success_metrics: List[str] # How team measures success
# Team dynamics
collaboration_patterns: Dict[str, float] # How roles collaborate
communication_preferences: Dict[str, str] # Preferred communication styles
decision_making_style: str # Consensus, hierarchical, etc.
```
#### Goal-Aware Context Generation
```python
def generate_team_aligned_context(
context: ContextNode,
team: TeamStructure,
active_goals: List[ProjectGoal]
) -> ContextNode:
# Find goals relevant to this team
team_goals = [g for g in active_goals if g.goal_id in team.current_goals]
# Calculate context relevance to team goals
goal_relevance_scores = {}
for goal in team_goals:
relevance = calculate_context_goal_relevance(context, goal)
weight = team.priority_weights.get(goal.goal_id, 0.5)
goal_relevance_scores[goal.goal_id] = relevance * weight
# Enhance context with goal-relevant insights
if goal_relevance_scores:
max_relevance_goal = max(goal_relevance_scores, key=goal_relevance_scores.get)
goal = next(g for g in team_goals if g.goal_id == max_relevance_goal)
# Add goal-specific insights
context.insights.append(f"TEAM-GOAL: Supports {goal.title}")
context.insights.append(f"GOAL-RELEVANCE: {goal_relevance_scores[max_relevance_goal]:.2f}")
# Add goal-specific tags
context.tags.extend([f"goal-{goal.goal_id}", f"team-{team.team_id}"])
return context
```
### Strategic Objective Mapping
Maps context relevance to high-level strategic objectives:
#### Objective-Context Mapping
```python
@dataclass
class StrategicObjective:
objective_id: str
title: str
description: str
business_value: float # Expected business value (0.0-1.0)
technical_complexity: float # Technical complexity (0.0-1.0)
risk_level: float # Risk level (0.0-1.0)
# Success criteria
success_metrics: List[str]
milestone_criteria: List[str]
completion_indicators: List[str]
# Context mapping
primary_components: List[str] # UCXL addresses central to objective
supporting_components: List[str] # UCXL addresses that support objective
context_indicators: List[str] # Patterns that indicate relevance
# Resource allocation
allocated_team_capacity: float # Fraction of team time allocated
priority_ranking: int # 1 = highest priority
dependency_objectives: List[str] # Other objectives this depends on
```
#### Objective-Driven Insight Prioritization
```python
def prioritize_insights_by_objectives(
context: ContextNode,
objectives: List[StrategicObjective]
) -> ContextNode:
# Calculate context relevance to each objective
objective_scores = {}
for objective in objectives:
relevance = calculate_objective_relevance(context, objective)
business_weight = objective.business_value * (1.0 / objective.priority_ranking)
objective_scores[objective.objective_id] = relevance * business_weight
# Sort insights by strategic value
insight_priorities = []
for insight in context.insights:
max_relevance = 0.0
best_objective = None
for obj_id, score in objective_scores.items():
insight_relevance = calculate_insight_objective_relevance(insight, obj_id)
total_score = score * insight_relevance
if total_score > max_relevance:
max_relevance = total_score
best_objective = obj_id
insight_priorities.append((insight, max_relevance, best_objective))
# Reorder insights by strategic priority
insight_priorities.sort(key=lambda x: x[1], reverse=True)
# Enhance high-priority insights
enhanced_insights = []
for insight, priority, objective_id in insight_priorities:
if priority > 0.7:
enhanced_insights.append(f"[HIGH-STRATEGIC-VALUE] {insight}")
elif priority > 0.4:
enhanced_insights.append(f"[STRATEGIC] {insight}")
else:
enhanced_insights.append(insight)
context.insights = enhanced_insights
return context
```
### Dynamic Priority Adjustment
Adjusts context generation focus based on changing priorities:
#### Priority Change Detection
```python
@dataclass
class PriorityChange:
change_id: str
timestamp: datetime
change_type: PriorityChangeType # GOAL_ADDED, GOAL_REMOVED, PRIORITY_CHANGED
affected_goals: List[str]
previous_state: Dict[str, Any]
new_state: Dict[str, Any]
change_rationale: str
impact_assessment: str
```
#### Adaptive Context Generation
```python
class AdaptiveContextGenerator:
def __init__(self):
self.priority_history = []
self.context_cache = {}
self.adaptation_weights = {}
def adjust_for_priority_changes(self, changes: List[PriorityChange]):
# Analyze priority change patterns
change_impacts = self.analyze_change_impacts(changes)
# Update adaptation weights
for change in changes:
if change.change_type == PriorityChangeType.GOAL_ADDED:
self.boost_goal_context_generation(change.affected_goals)
elif change.change_type == PriorityChangeType.PRIORITY_CHANGED:
self.reweight_goal_priorities(change.affected_goals, change.new_state)
# Invalidate affected context cache
self.invalidate_affected_cache(change_impacts)
def boost_goal_context_generation(self, goal_ids: List[str]):
for goal_id in goal_ids:
self.adaptation_weights[goal_id] = self.adaptation_weights.get(goal_id, 1.0) * 1.5
def reweight_goal_priorities(self, goal_ids: List[str], new_priorities: Dict[str, float]):
for goal_id in goal_ids:
if goal_id in new_priorities:
self.adaptation_weights[goal_id] = new_priorities[goal_id]
```
### Success Metrics Tracking
Monitors the effectiveness of goal alignment over time:
#### Alignment Metrics
```python
@dataclass
class AlignmentMetrics:
measurement_timestamp: datetime
measurement_period: timedelta
# Goal achievement metrics
goals_on_track: int
goals_at_risk: int
goals_completed: int
average_goal_progress: float
# Context alignment metrics
contexts_generated: int
goal_aligned_contexts: int
alignment_score_average: float
alignment_confidence_average: float
# Team satisfaction metrics
team_alignment_satisfaction: Dict[str, float] # team_id -> satisfaction
role_context_relevance: Dict[AgentRole, float] # role -> relevance score
# System performance metrics
context_generation_time: float
alignment_calculation_time: float
cache_hit_rate: float
```
#### Alignment Effectiveness Analysis
```python
def analyze_alignment_effectiveness(
metrics_history: List[AlignmentMetrics],
goals: List[ProjectGoal]
) -> AlignmentReport:
# Trend analysis
alignment_trend = calculate_alignment_trend(metrics_history)
goal_completion_trend = calculate_completion_trend(metrics_history)
satisfaction_trend = calculate_satisfaction_trend(metrics_history)
# Correlation analysis
context_goal_correlation = analyze_context_goal_correlation(metrics_history, goals)
# Identify improvement opportunities
improvement_areas = identify_improvement_opportunities(
alignment_trend,
satisfaction_trend,
context_goal_correlation
)
return AlignmentReport(
overall_alignment_score=alignment_trend.current_score,
trending_direction=alignment_trend.direction,
goal_achievement_rate=goal_completion_trend.achievement_rate,
team_satisfaction_average=satisfaction_trend.average,
improvement_recommendations=improvement_areas,
success_indicators=extract_success_indicators(metrics_history)
)
```
## Integration with BZZZ Leader System
### Leader-Coordinated Goal Management
- **Goal Authority**: Leader maintains authoritative goal definitions
- **Priority Coordination**: Leader coordinates priority changes across team
- **Alignment Oversight**: Leader monitors and adjusts alignment strategies
- **Performance Tracking**: Leader tracks alignment effectiveness metrics
### Role-Based Goal Distribution
- **Goal Visibility**: Agents see goals relevant to their role
- **Priority Communication**: Role-specific priority information
- **Progress Updates**: Regular updates on goal progress relevant to role
- **Alignment Feedback**: Mechanisms for agents to provide alignment feedback
## Configuration and Customization
### Goal Configuration
```yaml
project_goals:
- goal_id: "performance_optimization_2024"
title: "System Performance Optimization"
level: "STRATEGIC"
priority: "HIGH"
context_keywords: ["performance", "optimization", "latency", "throughput"]
technology_focus: ["caching", "indexing", "algorithms"]
success_criteria:
- "Reduce average response time to <200ms"
- "Increase throughput by 50%"
- "Maintain 99.9% availability"
alignment_settings:
mission_weight: 0.4 # How much mission influences context
team_goals_weight: 0.3 # How much team goals influence context
strategic_objectives_weight: 0.3 # How much strategic objectives influence
adaptation_responsiveness: 0.7 # How quickly to adapt to priority changes
cache_invalidation_threshold: 0.5 # When to invalidate cached contexts
metrics_collection_interval: "1 day"
alignment_report_frequency: "1 week"
```
## Future Enhancements
### Advanced Goal Intelligence
- **Goal Prediction**: Predict likely next goals based on project progress
- **Automatic Goal Generation**: Generate sub-goals automatically from high-level objectives
- **Goal Conflict Detection**: Identify conflicting goals and suggest resolutions
- **Success Prediction**: Predict goal completion likelihood and timeline
### Machine Learning Integration
- **Alignment Optimization**: ML models to optimize context-goal alignment
- **Priority Prediction**: Predict priority changes based on project patterns
- **Team Dynamics**: ML understanding of team collaboration patterns
- **Success Pattern Recognition**: Learn patterns that lead to goal achievement
### Real-Time Alignment
- **Live Priority Tracking**: Real-time priority adjustment based on events
- **Instant Context Adaptation**: Immediate context updates when priorities change
- **Proactive Goal Suggestions**: Suggest new goals based on project evolution
- **Dynamic Team Rebalancing**: Adjust team focus based on goal progress

View File

@@ -1,84 +0,0 @@
# SLURP Context Intelligence Engine
The Context Intelligence Engine is the core component of SLURP responsible for generating, extracting, and resolving contextual information about files and systems within the BZZZ distributed architecture.
## Purpose
This module implements the "Understanding" and "Processing" aspects of SLURP by:
- **Context Generation**: Creating intelligent, hierarchical context metadata
- **Context Resolution**: Efficiently resolving context through CSS-like inheritance
- **Bounded Hierarchy**: Limiting traversal depth to prevent excessive processing
- **Role-Aware Context**: Generating context specific to AI agent roles
## Key Components
### cascading_metadata_generator.py
Implements CSS-like cascading context inheritance system:
- Context flows DOWN the directory tree (inheritance)
- More specific contexts override parent contexts
- Only unique/different metadata is stored per level
- Massive space savings by avoiding redundant metadata
### context_resolver.py
Efficient context resolution through hierarchical lookup:
- Loads cascading metadata hierarchy
- Resolves context through CSS-like inheritance
- Fast lookups with caching
- Global context support
### bounded_context_demo.py
Complete demonstration system combining all context intelligence features:
- Bounded hierarchy walking with configurable depth limits
- Global context support for system-wide applicable metadata
- Integration with temporal decision tracking
- Smart context resolution with inheritance
## Architecture
The Context Intelligence Engine follows these principles:
1. **Hierarchical Context**: Context inherits from parent directories unless overridden
2. **Bounded Traversal**: Limits hierarchy depth to prevent excessive processing
3. **CSS-like Specificity**: More specific contexts override general ones
4. **Global Contexts**: System-wide contexts that apply everywhere
5. **Role-Based Generation**: Context tailored to specific AI agent roles
## Integration with BZZZ Leader System
In the BZZZ architecture, only the elected Leader node generates context intelligence:
- **Leader-Only Generation**: Prevents conflicting context from multiple sources
- **Role-Based Encryption**: Context is encrypted per AI agent role
- **Need-to-Know Access**: Each agent receives only relevant context
- **Quality Control**: Centralized generation ensures consistent, high-quality context
## Usage
```python
from slurp.context_intelligence.context_resolver import CascadingContextResolver
# Initialize resolver with bounded depth
resolver = CascadingContextResolver(metadata_dir, max_hierarchy_depth=10)
# Resolve context for a UCXL address
context = resolver.resolve("ucxl://any:any@BZZZ:RUSTLE-testing/src/main.rs")
# Search by tags or technologies
rust_contexts = resolver.search_by_technology("rust")
source_contexts = resolver.search_by_tag("source-code")
```
## Performance Characteristics
- **Space Efficiency**: 85%+ space savings through intelligent inheritance
- **Resolution Speed**: O(log n) average case with caching
- **Bounded Depth**: Configurable maximum traversal depth
- **Memory Usage**: Minimal through lazy loading and caching strategies
## Future Enhancements
- RAG integration for enhanced context analysis
- Machine learning-based context quality scoring
- Dynamic context refresh based on file changes
- Advanced role-based context customization

View File

@@ -1,382 +0,0 @@
#!/usr/bin/env python3
"""
Bounded Hierarchical Context System Demo
This script demonstrates the complete intelligent context system:
1. **Bounded Hierarchy Walking**: Context resolution with configurable depth limits
2. **Global Context Support**: Contexts that apply everywhere regardless of hierarchy
3. **Decision-Hop Temporal Analysis**: Track related decisions by decision distance (not time)
4. **Space-Efficient Storage**: CSS-like cascading eliminates redundant metadata
5. **Smart Context Resolution**: Intelligent lookup with caching and inheritance
Key Innovations:
- Hierarchy traversal stops at configurable depth (user's "bounded" requirement)
- Global contexts provide system-wide applicable context
- Temporal analysis based on decision hops, not chronological time
- RAG-like decision finding: discover decisions x hops away conceptually
Usage:
python3 bounded_context_demo.py [--max-depth N] [--demo-global] [--decision-hops N]
"""
import argparse
import json
import logging
from pathlib import Path
from datetime import datetime, timezone
from .cascading_metadata_generator import CascadingMetadataSystem
from .context_resolver import CascadingContextResolver
from ..temporal.temporal_context_system import (
TemporalContextGraph,
ContextChangeReason,
DecisionMetadata
)
logging.basicConfig(level=logging.INFO, format='%(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
class BoundedContextSystem:
"""Complete bounded hierarchical context system with temporal decision analysis"""
def __init__(self, project_path: str, metadata_base: str, max_hierarchy_depth: int = 10):
self.project_path = Path(project_path)
self.metadata_base = Path(metadata_base)
self.max_hierarchy_depth = max_hierarchy_depth
# Initialize subsystems
self.cascading_system = None
self.context_resolver = None
self.temporal_graph = None
# Ensure metadata directory exists
self.metadata_base.mkdir(parents=True, exist_ok=True)
def initialize_system(self, project_name: str = "BZZZ") -> None:
"""Initialize all components of the bounded context system"""
logger.info("🚀 Initializing Bounded Hierarchical Context System")
logger.info(f" 📁 Project: {self.project_path}")
logger.info(f" 💾 Metadata: {self.metadata_base}")
logger.info(f" 📏 Max depth: {self.max_hierarchy_depth}")
# 1. Initialize cascading metadata generator
self.cascading_system = CascadingMetadataSystem(
bzzz_path=str(self.project_path),
rag_endpoint="http://localhost:8000/query", # Not used in this demo
metadata_base=str(self.metadata_base)
)
# 2. Initialize context resolver with bounded depth
metadata_dir = self.metadata_base / project_name
self.context_resolver = CascadingContextResolver(
str(metadata_dir),
max_hierarchy_depth=self.max_hierarchy_depth
)
# 3. Initialize temporal graph for decision analysis
self.temporal_graph = TemporalContextGraph(
str(self.metadata_base),
project_name
)
logger.info("✅ All systems initialized")
def generate_hierarchical_metadata(self) -> dict:
"""Generate the cascading hierarchical metadata system"""
logger.info("\n🏗️ Generating Hierarchical Metadata System")
logger.info("=" * 50)
# Process repository with cascading system
results = self.cascading_system.process_repository()
logger.info(f"✅ Generated hierarchical metadata:")
logger.info(f" 📊 Processed {results['paths_processed']} paths")
logger.info(f" 🏗️ Created {results['context_nodes']} context nodes")
logger.info(f" 💾 Space savings: {results['space_savings_percent']:.1f}%")
logger.info(f" 📏 Traditional size: {results['estimated_traditional_size_kb']} KB")
logger.info(f" 🎯 Actual size: {results['actual_size_kb']} KB")
return results
def setup_global_contexts(self) -> None:
"""Set up global contexts that apply to all paths"""
logger.info("\n🌐 Setting Up Global Contexts")
logger.info("=" * 30)
# Global BZZZ project context
global_bzzz_context = {
'id': 'global_bzzz',
'summary': 'BZZZ Distributed Contextual Metadata System',
'purpose': 'Unified contextual exchange protocol with 1:1 filesystem mapping',
'technologies': ['UCXL Protocol', 'Distributed Systems', 'Contextual Metadata'],
'tags': ['bzzz', 'ucxl', 'distributed', 'metadata', 'global'],
'insights': [
'Part of CHORUS ecosystem for AI development',
'Implements hierarchical context inheritance',
'Supports temporal decision tracking',
'Enables space-efficient metadata storage'
],
'is_global': True,
'context_specificity': -1
}
# Global development practices context
global_dev_context = {
'id': 'global_development',
'summary': 'Development practices and architectural principles',
'purpose': 'Ensure consistent development patterns across all components',
'technologies': ['Git', 'Testing', 'Documentation', 'Code Review'],
'tags': ['development', 'practices', 'architecture', 'global'],
'insights': [
'All components follow consistent architectural patterns',
'Code changes tracked through decision-based temporal evolution',
'Comprehensive testing and documentation required',
'Context inheritance reduces maintenance overhead'
],
'is_global': True,
'context_specificity': -1
}
# Add global contexts
self.context_resolver.add_global_context(global_bzzz_context)
self.context_resolver.add_global_context(global_dev_context)
logger.info(f"✅ Added {len(self.context_resolver.global_contexts)} global contexts")
logger.info(" 🌍 BZZZ Project Context (applies to all files)")
logger.info(" 🛠️ Development Practices Context (applies to all files)")
def demonstrate_bounded_resolution(self) -> None:
"""Demonstrate bounded hierarchy walking with context resolution"""
logger.info(f"\n📏 Bounded Context Resolution Demo (max depth: {self.max_hierarchy_depth})")
logger.info("=" * 60)
# Test addresses at different depths
test_addresses = [
"ucxl://any:any@BZZZ:RUSTLE-testing/src/main.rs",
"ucxl://any:any@BZZZ:RUSTLE-testing/src/api/handlers.rs",
"ucxl://any:any@BZZZ:RUSTLE-testing/tests/unit/core_tests.rs",
"ucxl://any:any@BZZZ:RUSTLE-testing/assets/fonts/README.md"
]
for address in test_addresses:
logger.info(f"\n🔍 Resolving: {address}")
context = self.context_resolver.resolve(address)
if context:
logger.info(f" 📝 Summary: {context.summary}")
logger.info(f" 🏷️ Tags: {', '.join(context.tags[:5])}")
logger.info(f" 📊 Inheritance chain: {len(context.inheritance_chain)} levels")
logger.info(f" 🌐 Global contexts applied: {hasattr(context, 'global_contexts_applied')}")
logger.info(f" ⚡ Resolution confidence: {context.resolution_confidence:.2f}")
# Show bounded depth information if available
if hasattr(context, 'bounded_depth'):
logger.info(f" 📏 Hierarchy depth reached: {context.bounded_depth}/{self.max_hierarchy_depth}")
else:
logger.warning(f" ❌ Could not resolve context")
# Show system statistics
stats = self.context_resolver.get_context_statistics()
logger.info(f"\n📊 System Statistics:")
logger.info(f" 🏗️ Context nodes: {stats['context_nodes']}")
logger.info(f" 🌐 Global contexts: {stats['global_contexts']}")
logger.info(f" 📏 Max hierarchy depth: {stats['max_hierarchy_depth']}")
logger.info(f" 💾 Cached resolutions: {stats['cached_resolutions']}")
def demonstrate_decision_hop_analysis(self, max_decision_hops: int = 3) -> None:
"""Demonstrate decision-hop based temporal analysis"""
logger.info(f"\n🕸️ Decision-Hop Analysis Demo (max hops: {max_decision_hops})")
logger.info("=" * 50)
# Create some temporal context with decisions
base_address = "ucxl://any:any@BZZZ:RUSTLE-testing/src/main.rs"
# Initial context creation
initial_context = {
'summary': 'Main entry point implementing BZZZ distributed system bootstrap',
'purpose': 'Application startup and P2P network initialization',
'technologies': ['Rust', 'Tokio', 'P2P'],
'tags': ['main', 'entry-point', 'rust', 'distributed'],
'insights': ['Initializes P2P network for metadata distribution']
}
initial_node = self.temporal_graph.create_initial_context(
base_address, initial_context, "system_analysis"
)
# Decision 1: Architecture enhancement
arch_decision = DecisionMetadata(
decision_maker="tony",
decision_id="arch_001",
decision_rationale="Enhanced P2P architecture with bounded context hierarchy",
impact_scope="system",
confidence_level=0.9,
external_references=["design_doc_v2.md"]
)
arch_context = initial_context.copy()
arch_context['insights'].append('Implements bounded hierarchy traversal for efficient context resolution')
arch_context['technologies'].append('Bounded Context Hierarchy')
arch_node = self.temporal_graph.evolve_context(
base_address, arch_context, ContextChangeReason.ARCHITECTURE_CHANGE, arch_decision
)
# Create related addresses and decisions
related_addresses = [
"ucxl://any:any@BZZZ:RUSTLE-testing/src/context_resolver.rs",
"ucxl://any:any@BZZZ:RUSTLE-testing/src/temporal_graph.rs"
]
for i, addr in enumerate(related_addresses):
context = {
'summary': f'Component {i+1}: {addr.split("/")[-1]}',
'purpose': f'Supporting component for bounded context system',
'technologies': ['Rust', 'Context Resolution'],
'tags': ['component', 'context-system'],
'insights': [f'Implements part of the bounded context architecture']
}
decision = DecisionMetadata(
decision_maker="system_analysis",
decision_id=f"component_{i+1}",
decision_rationale=f"Component creation for bounded context implementation",
impact_scope="module",
confidence_level=0.8,
external_references=[]
)
# Create temporal context
self.temporal_graph.evolve_context(
addr, context, ContextChangeReason.ARCHITECTURE_CHANGE, decision
)
# Create influence relationships
self.temporal_graph.add_influence_relationship(base_address, addr)
# Demonstrate decision-hop analysis
logger.info(f"🎯 Decision Timeline for {base_address.split('/')[-1]}:")
timeline = self.temporal_graph.get_decision_timeline(
base_address, include_related=True, max_hops=max_decision_hops
)
for decision in timeline['decision_sequence']:
logger.info(f" Decision #{decision['decision_hop']}: {decision['change_reason']}")
logger.info(f" 👤 Decision maker: {decision['decision_maker']}")
logger.info(f" 💪 Confidence: {decision['confidence_evolution']:.2f}")
logger.info(f" 📊 Influences: {decision['influences_count']} addresses")
# Show related decisions within hop distance
if timeline['related_decisions']:
logger.info(f"\n🔗 Related Decisions (within {max_decision_hops} hops):")
for related in timeline['related_decisions']:
logger.info(f" {related['address'].split('/')[-1]} ({related['decision_hops']} hops)")
logger.info(f" 🔄 Latest: {related['change_reason']}")
logger.info(f" 📊 Confidence: {related['confidence']:.2f}")
# Find decision paths
logger.info(f"\n🛤️ Decision Paths:")
for addr in related_addresses:
path = self.temporal_graph.find_decision_path(base_address, addr)
if path:
path_names = [p[0].split('/')[-1] for p in path]
logger.info(f" {base_address.split('/')[-1]}{''.join(path_names[1:])}")
def save_bounded_system(self) -> None:
"""Save the complete bounded context system"""
logger.info("\n💾 Saving Bounded Context System")
logger.info("=" * 35)
# Save temporal data
self.temporal_graph.save_temporal_data()
# Save global contexts
global_file = self.metadata_base / "BZZZ" / "global_contexts.json"
with open(global_file, 'w') as f:
json.dump(self.context_resolver.global_contexts, f, indent=2)
# Save system configuration
config = {
'max_hierarchy_depth': self.max_hierarchy_depth,
'project_path': str(self.project_path),
'metadata_base': str(self.metadata_base),
'system_version': '1.0.0-bounded',
'features': [
'bounded_hierarchy_walking',
'global_context_support',
'decision_hop_temporal_analysis',
'space_efficient_storage'
],
'created_at': datetime.now(timezone.utc).isoformat()
}
config_file = self.metadata_base / "BZZZ" / "system_config.json"
with open(config_file, 'w') as f:
json.dump(config, f, indent=2)
logger.info(f"✅ System saved:")
logger.info(f" 📄 Global contexts: {global_file}")
logger.info(f" ⚙️ Configuration: {config_file}")
logger.info(f" 🕸️ Temporal data: {self.temporal_graph.temporal_dir}")
def main():
parser = argparse.ArgumentParser(description="Bounded Hierarchical Context System Demo")
parser.add_argument("--project-path", default="/tmp/demo-bzzz-cascading",
help="Path to project directory")
parser.add_argument("--metadata-base", default="/tmp/bounded-context-metadata",
help="Base directory for metadata storage")
parser.add_argument("--max-depth", type=int, default=8,
help="Maximum hierarchy depth for bounded walking")
parser.add_argument("--decision-hops", type=int, default=3,
help="Maximum decision hops for temporal analysis")
parser.add_argument("--demo-global", action="store_true",
help="Demonstrate global context functionality")
args = parser.parse_args()
logger.info("🎬 Bounded Hierarchical Context System Demo")
logger.info("=" * 50)
# Initialize the bounded system
bounded_system = BoundedContextSystem(
args.project_path,
args.metadata_base,
args.max_depth
)
bounded_system.initialize_system()
# Generate hierarchical metadata
results = bounded_system.generate_hierarchical_metadata()
# Setup global contexts if requested
if args.demo_global:
bounded_system.setup_global_contexts()
# Demonstrate bounded resolution
bounded_system.demonstrate_bounded_resolution()
# Demonstrate decision-hop analysis
bounded_system.demonstrate_decision_hop_analysis(args.decision_hops)
# Save the system
bounded_system.save_bounded_system()
logger.info("\n✨ Demo Complete!")
logger.info("=" * 20)
logger.info("🎉 Bounded hierarchical context system demonstrated successfully")
logger.info("📏 Hierarchy walking bounded to prevent excessive traversal")
logger.info("🌐 Global contexts provide system-wide applicable metadata")
logger.info("🕸️ Decision-hop analysis tracks conceptually related changes")
logger.info("💾 Space-efficient storage through intelligent inheritance")
if __name__ == "__main__":
main()

View File

@@ -1,672 +0,0 @@
#!/usr/bin/env python3
"""
Cascading Hierarchical Metadata Generator
This system implements CSS-like cascading for contextual metadata:
1. Context flows DOWN the directory tree (inheritance)
2. More specific contexts override parent contexts
3. Only unique/different metadata is stored per level
4. Lookups resolve by walking UP the tree to find applicable context
5. Massive space savings by avoiding redundant metadata
Key Concepts:
- Context Inheritance: Child directories inherit parent context unless overridden
- Context Specificity: More specific paths can override parent context
- Context Consolidation: Similar contexts are merged/consolidated
- Lazy Resolution: Context is resolved at query time by walking the hierarchy
Usage:
python3 cascading_metadata_generator.py [--bzzz-path PATH] [--rag-endpoint URL]
"""
import os
import json
import argparse
import hashlib
from pathlib import Path
from typing import Dict, List, Optional, Any, Set, Tuple
from datetime import datetime, timezone
from dataclasses import dataclass, asdict
import requests
import logging
from collections import defaultdict
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
@dataclass
class ContextNode:
"""Represents a context node in the hierarchical tree"""
path: str
ucxl_address: str
summary: str
purpose: str
technologies: List[str]
tags: List[str]
insights: List[str]
overrides_parent: bool = False # Does this context override parent?
context_specificity: int = 0 # Higher = more specific
applies_to_children: bool = True # Does this context cascade down?
generated_at: str = ""
rag_confidence: float = 0.0
@dataclass
class PathMetadata:
"""Lightweight path metadata - most context comes from hierarchy"""
ucxl_address: str
filesystem_path: str
file_type: str
size_bytes: Optional[int]
extension: Optional[str]
language: Optional[str]
content_hash: Optional[str]
last_modified: Optional[str]
has_local_context: bool = False # Does this path have its own context node?
class CascadingMetadataSystem:
def __init__(self, bzzz_path: str, rag_endpoint: str, metadata_base: str):
self.bzzz_path = Path(bzzz_path)
self.rag_endpoint = rag_endpoint
self.metadata_base = Path(metadata_base)
self.project_name = "BZZZ"
self.project_metadata_dir = self.metadata_base / self.project_name
# Context hierarchy storage
self.context_tree: Dict[str, ContextNode] = {}
self.path_metadata: Dict[str, PathMetadata] = {}
# Context consolidation data
self.context_patterns = defaultdict(list) # Similar contexts grouped
self.directory_purposes = {} # Common directory purposes
# Ensure metadata directory exists
self.project_metadata_dir.mkdir(parents=True, exist_ok=True)
def analyze_directory_structure(self) -> Dict[str, Any]:
"""Analyze the entire directory structure to identify patterns and hierarchy"""
logger.info("🔍 Analyzing directory structure for context patterns...")
directory_analysis = {
'common_purposes': defaultdict(list),
'technology_clusters': defaultdict(set),
'pattern_directories': defaultdict(list),
'depth_analysis': defaultdict(int)
}
for item in self.bzzz_path.rglob('*'):
if not self.should_process_path(item):
continue
rel_path = item.relative_to(self.bzzz_path)
depth = len(rel_path.parts)
directory_analysis['depth_analysis'][depth] += 1
# Analyze directory patterns
if item.is_dir():
dir_name = item.name.lower()
# Common directory patterns
if dir_name in ['src', 'source', 'lib']:
directory_analysis['common_purposes']['source_code'].append(str(rel_path))
elif dir_name in ['test', 'tests', 'spec', 'specs']:
directory_analysis['common_purposes']['testing'].append(str(rel_path))
elif dir_name in ['doc', 'docs', 'documentation']:
directory_analysis['common_purposes']['documentation'].append(str(rel_path))
elif dir_name in ['config', 'configuration', 'settings']:
directory_analysis['common_purposes']['configuration'].append(str(rel_path))
elif dir_name in ['asset', 'assets', 'static', 'public']:
directory_analysis['common_purposes']['assets'].append(str(rel_path))
elif dir_name in ['font', 'fonts']:
directory_analysis['common_purposes']['fonts'].append(str(rel_path))
elif dir_name in ['image', 'images', 'img']:
directory_analysis['common_purposes']['images'].append(str(rel_path))
elif dir_name in ['style', 'styles', 'css']:
directory_analysis['common_purposes']['styling'].append(str(rel_path))
elif dir_name in ['script', 'scripts', 'js']:
directory_analysis['common_purposes']['scripts'].append(str(rel_path))
elif dir_name in ['build', 'dist', 'output', 'target']:
directory_analysis['common_purposes']['build_output'].append(str(rel_path))
elif dir_name in ['vendor', 'third_party', 'external']:
directory_analysis['common_purposes']['third_party'].append(str(rel_path))
elif dir_name in ['util', 'utils', 'helper', 'helpers', 'common']:
directory_analysis['common_purposes']['utilities'].append(str(rel_path))
elif dir_name in ['api', 'endpoint', 'service', 'services']:
directory_analysis['common_purposes']['api_services'].append(str(rel_path))
elif dir_name in ['model', 'models', 'entity', 'entities']:
directory_analysis['common_purposes']['data_models'].append(str(rel_path))
elif dir_name in ['component', 'components', 'widget', 'widgets']:
directory_analysis['common_purposes']['ui_components'].append(str(rel_path))
elif dir_name in ['template', 'templates', 'layout', 'layouts']:
directory_analysis['common_purposes']['templates'].append(str(rel_path))
# Analyze technology clusters by file extensions
if item.is_file():
ext = item.suffix.lower()
parent_dir = str(rel_path.parent) if rel_path.parent != Path('.') else 'root'
directory_analysis['technology_clusters'][parent_dir].add(ext)
logger.info(f"📊 Found {len(directory_analysis['common_purposes'])} common directory patterns")
logger.info(f"🔧 Identified {len(directory_analysis['technology_clusters'])} technology clusters")
return directory_analysis
def create_context_hierarchy(self) -> None:
"""Create the cascading context hierarchy based on directory analysis"""
logger.info("🏗️ Building cascading context hierarchy...")
# First, analyze the structure
structure_analysis = self.analyze_directory_structure()
# Create context nodes for significant directories
contexts_created = 0
for purpose, directories in structure_analysis['common_purposes'].items():
for dir_path in directories:
full_path = self.bzzz_path / dir_path
if full_path.exists() and full_path.is_dir():
context_node = self.create_directory_context(full_path, purpose)
if context_node:
self.context_tree[str(full_path)] = context_node
contexts_created += 1
# Create root project context
root_context = self.create_root_context()
self.context_tree[str(self.bzzz_path)] = root_context
contexts_created += 1
logger.info(f"✅ Created {contexts_created} context nodes in hierarchy")
def create_root_context(self) -> ContextNode:
"""Create the root context for the entire project"""
return ContextNode(
path=str(self.bzzz_path),
ucxl_address="ucxl://any:any@BZZZ:RUSTLE-testing",
summary="BZZZ distributed system project root",
purpose="Core distributed system implementing contextual metadata architecture with 1:1 filesystem mapping",
technologies=["Rust", "Go", "Distributed Systems", "P2P", "DHT", "UCXL Protocol"],
tags=["project-root", "distributed-system", "bzzz", "ucxl", "rust", "go"],
insights=[
"Main project implementing distributed contextual metadata system",
"Uses UCXL protocol for unified contextual exchange",
"Implements 1:1 mapping between filesystem and UCXL addresses",
"Part of larger CHORUS ecosystem for AI development"
],
overrides_parent=False,
context_specificity=0,
applies_to_children=True,
generated_at=datetime.now(timezone.utc).isoformat(),
rag_confidence=0.9
)
def create_directory_context(self, dir_path: Path, purpose_type: str) -> Optional[ContextNode]:
"""Create context for a specific directory based on its purpose"""
rel_path = dir_path.relative_to(self.bzzz_path)
ucxl_address = f"ucxl://any:any@BZZZ:RUSTLE-testing/{str(rel_path).replace(os.sep, '/')}"
# Context templates based on directory purpose
context_templates = {
'source_code': {
'summary': f"Source code directory: {dir_path.name}",
'purpose': "Implementation of core system functionality and business logic",
'technologies': ["Rust", "Go", "Source Code"],
'tags': ["source-code", "implementation", "core-logic"],
'insights': [
"Contains primary implementation files",
"Houses main business logic and algorithms",
"Critical for system functionality"
]
},
'testing': {
'summary': f"Testing directory: {dir_path.name}",
'purpose': "Quality assurance, validation, and testing infrastructure",
'technologies': ["Testing Frameworks", "Unit Tests", "Integration Tests"],
'tags': ["testing", "qa", "validation", "quality-assurance"],
'insights': [
"Ensures code quality and correctness",
"Provides regression testing capabilities",
"Critical for maintaining system reliability"
]
},
'documentation': {
'summary': f"Documentation directory: {dir_path.name}",
'purpose': "Project documentation, guides, and knowledge resources",
'technologies': ["Markdown", "Documentation"],
'tags': ["documentation", "guides", "knowledge", "reference"],
'insights': [
"Provides user and developer guidance",
"Contains architectural decisions and design docs",
"Essential for project maintainability"
]
},
'configuration': {
'summary': f"Configuration directory: {dir_path.name}",
'purpose': "System configuration, settings, and environment management",
'technologies': ["TOML", "YAML", "JSON", "Configuration"],
'tags': ["configuration", "settings", "environment", "deployment"],
'insights': [
"Manages system behavior and parameters",
"Controls deployment and runtime settings",
"Centralizes configuration management"
]
},
'assets': {
'summary': f"Assets directory: {dir_path.name}",
'purpose': "Static assets, resources, and multimedia content",
'technologies': ["Static Assets", "Resources"],
'tags': ["assets", "resources", "static", "content"],
'insights': [
"Houses non-code project resources",
"Supports user interface and experience",
"Manages static content delivery"
]
},
'fonts': {
'summary': f"Fonts directory: {dir_path.name}",
'purpose': "Typography assets implementing design system specifications",
'technologies': ["Typography", "Fonts", "Design System"],
'tags': ["fonts", "typography", "design-system", "ui"],
'insights': [
"Implements brand typography guidelines",
"Ensures consistent visual identity",
"Supports responsive design requirements"
]
},
'api_services': {
'summary': f"API services directory: {dir_path.name}",
'purpose': "API endpoints, service interfaces, and external communication",
'technologies': ["REST API", "HTTP", "Service Layer"],
'tags': ["api", "services", "endpoints", "communication"],
'insights': [
"Defines external system interfaces",
"Handles inter-service communication",
"Critical for system integration"
]
},
'utilities': {
'summary': f"Utilities directory: {dir_path.name}",
'purpose': "Shared utilities, helpers, and common functionality",
'technologies': ["Utilities", "Helper Functions", "Common Code"],
'tags': ["utilities", "helpers", "shared", "common"],
'insights': [
"Provides reusable functionality",
"Reduces code duplication",
"Supports DRY principles"
]
}
}
if purpose_type not in context_templates:
return None
template = context_templates[purpose_type]
return ContextNode(
path=str(dir_path),
ucxl_address=ucxl_address,
summary=template['summary'],
purpose=template['purpose'],
technologies=template['technologies'],
tags=template['tags'],
insights=template['insights'],
overrides_parent=False,
context_specificity=len(rel_path.parts),
applies_to_children=True,
generated_at=datetime.now(timezone.utc).isoformat(),
rag_confidence=0.8
)
def resolve_context_for_path(self, file_path: Path) -> ContextNode:
"""Resolve context for a path by walking UP the hierarchy (CSS-like cascading)"""
# Start from the file's directory and walk up to find applicable context
current_path = file_path if file_path.is_dir() else file_path.parent
contexts = []
# Walk up the directory tree collecting contexts
while current_path >= self.bzzz_path:
if str(current_path) in self.context_tree:
context = self.context_tree[str(current_path)]
if context.applies_to_children:
contexts.append(context)
if context.overrides_parent:
break
current_path = current_path.parent
# If no contexts found, use root context
if not contexts:
return self.context_tree.get(str(self.bzzz_path), self.create_root_context())
# Merge contexts (more specific overrides less specific)
return self.merge_contexts(contexts, file_path)
def merge_contexts(self, contexts: List[ContextNode], file_path: Path) -> ContextNode:
"""Merge multiple contexts using CSS-like specificity rules"""
if len(contexts) == 1:
return contexts[0]
# Sort by specificity (higher = more specific)
contexts.sort(key=lambda c: c.context_specificity, reverse=True)
# Start with most specific context
merged = contexts[0]
# Merge in less specific contexts where not overridden
for context in contexts[1:]:
# Tags are additive (union)
merged.tags = list(set(merged.tags + context.tags))
# Technologies are additive (union)
merged.technologies = list(set(merged.technologies + context.technologies))
# Insights are additive (append unique)
for insight in context.insights:
if insight not in merged.insights:
merged.insights.append(insight)
# Summary and purpose use most specific unless empty
if not merged.summary:
merged.summary = context.summary
if not merged.purpose:
merged.purpose = context.purpose
# Update path-specific information
rel_path = file_path.relative_to(self.bzzz_path)
merged.ucxl_address = f"ucxl://any:any@BZZZ:RUSTLE-testing/{str(rel_path).replace(os.sep, '/')}"
merged.path = str(file_path)
return merged
def should_process_path(self, path: Path) -> bool:
"""Determine if a path should be processed"""
if any(part.startswith('.') for part in path.parts):
return False
ignore_patterns = [
'target/', 'node_modules/', '__pycache__/', '.git/',
'vendor/', 'build/', 'dist/', '.cache/', 'tmp/'
]
path_str = str(path).lower()
return not any(pattern in path_str for pattern in ignore_patterns)
def create_path_metadata(self, file_path: Path) -> PathMetadata:
"""Create lightweight metadata for a path (context comes from hierarchy)"""
is_dir = file_path.is_dir()
rel_path = file_path.relative_to(self.bzzz_path)
ucxl_address = f"ucxl://any:any@BZZZ:RUSTLE-testing/{str(rel_path).replace(os.sep, '/')}"
# Basic file information only
size_bytes = None
content_hash = None
last_modified = None
if not is_dir:
try:
stat = file_path.stat()
size_bytes = stat.st_size
last_modified = datetime.fromtimestamp(stat.st_mtime, tz=timezone.utc).isoformat()
# Only hash small text files
if size_bytes < 50000: # 50KB limit
try:
content = file_path.read_text(encoding='utf-8')
content_hash = hashlib.sha256(content.encode('utf-8')).hexdigest()
except:
pass
except:
pass
# Determine language/type
language = None
if not is_dir:
ext = file_path.suffix.lower()
lang_map = {
'.rs': 'rust', '.go': 'go', '.py': 'python',
'.js': 'javascript', '.ts': 'typescript', '.md': 'markdown',
'.toml': 'toml', '.yaml': 'yaml', '.yml': 'yaml', '.json': 'json'
}
language = lang_map.get(ext)
return PathMetadata(
ucxl_address=ucxl_address,
filesystem_path=str(file_path),
file_type="directory" if is_dir else "file",
size_bytes=size_bytes,
extension=file_path.suffix if not is_dir else None,
language=language,
content_hash=content_hash,
last_modified=last_modified,
has_local_context=str(file_path) in self.context_tree
)
def save_cascading_metadata(self) -> Dict[str, Any]:
"""Save the cascading metadata system to files"""
# Save context hierarchy
hierarchy_file = self.project_metadata_dir / "context_hierarchy.json"
hierarchy_data = {
path: asdict(context) for path, context in self.context_tree.items()
}
with open(hierarchy_file, 'w', encoding='utf-8') as f:
json.dump(hierarchy_data, f, indent=2, ensure_ascii=False)
# Save path metadata (lightweight)
paths_file = self.project_metadata_dir / "path_metadata.json"
paths_data = {
path: asdict(metadata) for path, metadata in self.path_metadata.items()
}
with open(paths_file, 'w', encoding='utf-8') as f:
json.dump(paths_data, f, indent=2, ensure_ascii=False)
# Generate lookup index for fast context resolution
lookup_index = {}
for path, metadata in self.path_metadata.items():
file_path = Path(path)
resolved_context = self.resolve_context_for_path(file_path)
lookup_index[metadata.ucxl_address] = {
'context_path': resolved_context.path,
'specificity': resolved_context.context_specificity,
'has_local_context': metadata.has_local_context
}
index_file = self.project_metadata_dir / "context_lookup_index.json"
with open(index_file, 'w', encoding='utf-8') as f:
json.dump(lookup_index, f, indent=2, ensure_ascii=False)
return {
'context_nodes': len(self.context_tree),
'path_entries': len(self.path_metadata),
'hierarchy_file': str(hierarchy_file),
'paths_file': str(paths_file),
'index_file': str(index_file)
}
def generate_context_demo(self, demo_paths: List[str]) -> Dict[str, Any]:
"""Generate a demo showing how context cascades for specific paths"""
demo_results = {}
for path_str in demo_paths:
file_path = Path(path_str)
if not file_path.exists():
continue
resolved_context = self.resolve_context_for_path(file_path)
path_metadata = self.path_metadata.get(str(file_path), {})
demo_results[path_str] = {
'ucxl_address': resolved_context.ucxl_address,
'resolved_context': {
'summary': resolved_context.summary,
'purpose': resolved_context.purpose,
'technologies': resolved_context.technologies,
'tags': resolved_context.tags,
'context_source': resolved_context.path,
'specificity': resolved_context.context_specificity
},
'path_metadata': asdict(path_metadata) if hasattr(path_metadata, '__dict__') else path_metadata,
'inheritance_chain': self.get_inheritance_chain(file_path)
}
return demo_results
def get_inheritance_chain(self, file_path: Path) -> List[str]:
"""Get the chain of context inheritance for a path"""
chain = []
current_path = file_path if file_path.is_dir() else file_path.parent
while current_path >= self.bzzz_path:
if str(current_path) in self.context_tree:
chain.append(str(current_path))
current_path = current_path.parent
return chain
def process_repository(self) -> Dict[str, Any]:
"""Process the entire repository with cascading context system"""
logger.info("🚀 Processing repository with cascading context system...")
# Step 1: Create context hierarchy
self.create_context_hierarchy()
# Step 2: Create lightweight metadata for all paths
paths_processed = 0
for item in self.bzzz_path.rglob('*'):
if not self.should_process_path(item):
continue
metadata = self.create_path_metadata(item)
self.path_metadata[str(item)] = metadata
paths_processed += 1
logger.info(f"📊 Processed {paths_processed} paths with {len(self.context_tree)} context nodes")
# Step 3: Save the system
save_results = self.save_cascading_metadata()
# Step 4: Calculate space savings
traditional_size = paths_processed * 2000 # Estimate 2KB per traditional metadata file
actual_size = len(self.context_tree) * 2000 + paths_processed * 500 # Context + lightweight metadata
space_savings = ((traditional_size - actual_size) / traditional_size) * 100
return {
'paths_processed': paths_processed,
'context_nodes': len(self.context_tree),
'space_savings_percent': space_savings,
'estimated_traditional_size_kb': traditional_size // 1024,
'actual_size_kb': actual_size // 1024,
**save_results
}
def main():
parser = argparse.ArgumentParser(description="Generate cascading hierarchical metadata for BZZZ project")
parser.add_argument("--bzzz-path", default="/home/tony/chorus/project-queues/active/BZZZ",
help="Path to BZZZ repository")
parser.add_argument("--metadata-base", default=os.path.expanduser("~/chorus/project-metadata"),
help="Base directory for metadata storage")
parser.add_argument("--demo", action="store_true",
help="Run demonstration with sample paths")
args = parser.parse_args()
# Check if BZZZ path exists, create demo if not
bzzz_path = Path(args.bzzz_path)
if not bzzz_path.exists():
logger.warning(f"BZZZ repository not found at: {bzzz_path}")
logger.info("Creating demo structure...")
demo_path = Path("/tmp/demo-bzzz-cascading")
demo_path.mkdir(exist_ok=True)
# Create comprehensive demo structure
directories = [
"src", "src/api", "src/core", "src/utils",
"tests", "tests/unit", "tests/integration",
"docs", "docs/api", "docs/user",
"config", "config/dev", "config/prod",
"assets", "assets/fonts", "assets/images",
"scripts", "build"
]
for dir_path in directories:
(demo_path / dir_path).mkdir(parents=True, exist_ok=True)
# Create demo files
files = {
"README.md": "# BZZZ Project\n\nDistributed contextual metadata system",
"Cargo.toml": "[package]\nname = \"bzzz\"\nversion = \"0.1.0\"",
"src/main.rs": "fn main() { println!(\"BZZZ!\"); }",
"src/lib.rs": "//! BZZZ core library",
"src/api/handlers.rs": "//! HTTP request handlers",
"src/core/engine.rs": "//! Core processing engine",
"src/utils/helpers.rs": "//! Utility functions",
"tests/unit/core_tests.rs": "//! Unit tests for core",
"tests/integration/api_tests.rs": "//! API integration tests",
"docs/README.md": "# Documentation\n\nProject documentation",
"docs/api/endpoints.md": "# API Endpoints",
"config/settings.toml": "[server]\nport = 8080",
"assets/fonts/README.md": "# Fonts\n\nTypography assets for UI",
"scripts/build.sh": "#!/bin/bash\ncargo build --release"
}
for file_path, content in files.items():
full_path = demo_path / file_path
full_path.write_text(content)
bzzz_path = demo_path
logger.info(f"Demo structure created at: {demo_path}")
# Initialize the cascading system
system = CascadingMetadataSystem(
bzzz_path=str(bzzz_path),
rag_endpoint="http://localhost:8000/query", # Not used in this version
metadata_base=args.metadata_base
)
# Process the repository
results = system.process_repository()
logger.info("✅ Cascading metadata system complete!")
logger.info(f"📊 Results:")
logger.info(f" 📁 Paths processed: {results['paths_processed']}")
logger.info(f" 🏗️ Context nodes: {results['context_nodes']}")
logger.info(f" 💾 Space savings: {results['space_savings_percent']:.1f}%")
logger.info(f" 📏 Traditional size: {results['estimated_traditional_size_kb']} KB")
logger.info(f" 🎯 Actual size: {results['actual_size_kb']} KB")
logger.info(f" 📂 Files saved:")
logger.info(f" 🏗️ Hierarchy: {results['hierarchy_file']}")
logger.info(f" 📄 Paths: {results['paths_file']}")
logger.info(f" 🔍 Index: {results['index_file']}")
# Run demo if requested
if args.demo:
logger.info("\n🎬 Running context resolution demo...")
demo_paths = [
str(bzzz_path / "src" / "main.rs"),
str(bzzz_path / "src" / "api" / "handlers.rs"),
str(bzzz_path / "tests" / "unit" / "core_tests.rs"),
str(bzzz_path / "assets" / "fonts" / "README.md"),
str(bzzz_path / "config" / "settings.toml")
]
demo_results = system.generate_context_demo(demo_paths)
for path, info in demo_results.items():
logger.info(f"\n📄 {path}:")
logger.info(f" 🔗 UCXL: {info['ucxl_address']}")
logger.info(f" 📝 Summary: {info['resolved_context']['summary']}")
logger.info(f" 🎯 Purpose: {info['resolved_context']['purpose']}")
logger.info(f" 🏷️ Tags: {', '.join(info['resolved_context']['tags'][:5])}")
logger.info(f" 📊 Context from: {Path(info['resolved_context']['context_source']).name}")
logger.info(f" 🔗 Inheritance: {''.join([Path(p).name for p in info['inheritance_chain']])}")
if __name__ == "__main__":
main()

View File

@@ -1,438 +0,0 @@
#!/usr/bin/env python3
"""
Cascading Context Resolution System
This system efficiently resolves context for UCXL addresses by:
1. Loading the cascading metadata hierarchy
2. Resolving context through CSS-like inheritance
3. Providing fast lookups with caching
4. Supporting temporal context evolution (decisions change over time)
The temporal component you mentioned is crucial - we need to track how context
and decisions evolve over time. A temporal graph network could indeed be the
way to handle this, where each context node has temporal versions.
Usage:
from context_resolver import CascadingContextResolver
resolver = CascadingContextResolver("~/chorus/project-metadata/BZZZ")
context = resolver.resolve("ucxl://any:any@BZZZ:RUSTLE-testing/src/main.rs")
"""
import json
from pathlib import Path
from typing import Dict, List, Optional, Any, Tuple
from datetime import datetime, timezone
from dataclasses import dataclass
import logging
logger = logging.getLogger(__name__)
@dataclass
class ResolvedContext:
"""Complete resolved context for a UCXL address"""
ucxl_address: str
summary: str
purpose: str
technologies: List[str]
tags: List[str]
insights: List[str]
# Metadata
file_type: str
size_bytes: Optional[int]
language: Optional[str]
last_modified: Optional[str]
content_hash: Optional[str]
# Resolution metadata
context_source_path: str
inheritance_chain: List[str]
resolution_confidence: float
resolved_at: str
# Temporal metadata (for future enhancement)
temporal_version: int = 1
context_evolution: List[str] = None
class CascadingContextResolver:
"""Efficiently resolves cascading context for UCXL addresses"""
def __init__(self, metadata_dir: str, max_hierarchy_depth: int = 10):
self.metadata_dir = Path(metadata_dir)
self.context_hierarchy = {}
self.path_metadata = {}
self.lookup_index = {}
self.resolution_cache = {}
self.global_contexts = [] # Contexts that apply to all paths
self.max_hierarchy_depth = max_hierarchy_depth
self.load_metadata_system()
def load_metadata_system(self) -> None:
"""Load the cascading metadata system from files"""
try:
# Load context hierarchy
hierarchy_file = self.metadata_dir / "context_hierarchy.json"
if hierarchy_file.exists():
with open(hierarchy_file, 'r') as f:
self.context_hierarchy = json.load(f)
logger.info(f"Loaded {len(self.context_hierarchy)} context nodes")
# Load path metadata
paths_file = self.metadata_dir / "path_metadata.json"
if paths_file.exists():
with open(paths_file, 'r') as f:
self.path_metadata = json.load(f)
logger.info(f"Loaded {len(self.path_metadata)} path entries")
# Load lookup index
index_file = self.metadata_dir / "context_lookup_index.json"
if index_file.exists():
with open(index_file, 'r') as f:
self.lookup_index = json.load(f)
logger.info(f"Loaded {len(self.lookup_index)} lookup entries")
# Load global contexts
global_file = self.metadata_dir / "global_contexts.json"
if global_file.exists():
with open(global_file, 'r') as f:
self.global_contexts = json.load(f)
logger.info(f"Loaded {len(self.global_contexts)} global contexts")
except Exception as e:
logger.error(f"Error loading metadata system: {e}")
raise
def resolve(self, ucxl_address: str) -> Optional[ResolvedContext]:
"""Resolve context for a UCXL address using cascading inheritance"""
# Check cache first
if ucxl_address in self.resolution_cache:
return self.resolution_cache[ucxl_address]
# Find the filesystem path from UCXL address
filesystem_path = self.ucxl_to_filesystem_path(ucxl_address)
if not filesystem_path:
return None
# Get path metadata
path_meta = self.path_metadata.get(filesystem_path)
if not path_meta:
return None
# Resolve context through hierarchy with bounded traversal
resolved_context = self.resolve_cascading_context(filesystem_path, self.max_hierarchy_depth)
if not resolved_context:
return None
# Combine path metadata with resolved context
result = ResolvedContext(
ucxl_address=ucxl_address,
summary=resolved_context.get('summary', ''),
purpose=resolved_context.get('purpose', ''),
technologies=resolved_context.get('technologies', []),
tags=resolved_context.get('tags', []),
insights=resolved_context.get('insights', []),
file_type=path_meta.get('file_type', 'unknown'),
size_bytes=path_meta.get('size_bytes'),
language=path_meta.get('language'),
last_modified=path_meta.get('last_modified'),
content_hash=path_meta.get('content_hash'),
context_source_path=resolved_context.get('source_path', ''),
inheritance_chain=resolved_context.get('inheritance_chain', []),
resolution_confidence=resolved_context.get('confidence', 0.5),
resolved_at=datetime.now(timezone.utc).isoformat(),
context_evolution=[] # Placeholder for temporal versioning
)
# Cache the result
self.resolution_cache[ucxl_address] = result
return result
def ucxl_to_filesystem_path(self, ucxl_address: str) -> Optional[str]:
"""Convert UCXL address to filesystem path"""
# Parse UCXL address: ucxl://agent:role@project:task/path
if not ucxl_address.startswith("ucxl://"):
return None
try:
# Remove scheme
remainder = ucxl_address[7:] # Remove "ucxl://"
# Split by @ to separate authority and path
if '@' not in remainder:
return None
authority_part, path_part = remainder.split('@', 1)
# Split by first / to separate project:task from path
if '/' in path_part:
project_task, file_path = path_part.split('/', 1)
# Look for matching path in our metadata
for fs_path, meta in self.path_metadata.items():
if meta['ucxl_address'] == ucxl_address:
return fs_path
# If not found, try to construct from UCXL path
# This is project-specific logic for BZZZ
if project_task == "BZZZ:RUSTLE-testing":
base_path = "/home/tony/chorus/project-queues/active/BZZZ" # Could be configurable
constructed_path = str(Path(base_path) / file_path)
if constructed_path in self.path_metadata:
return constructed_path
return None
except Exception as e:
logger.warning(f"Error parsing UCXL address {ucxl_address}: {e}")
return None
def resolve_cascading_context(self, filesystem_path: str, max_depth: int = 10) -> Optional[Dict[str, Any]]:
"""Resolve context by walking up the directory hierarchy with bounded traversal"""
current_path = Path(filesystem_path)
if not current_path.is_dir():
current_path = current_path.parent
contexts = []
inheritance_chain = []
depth = 0
# First, check for global contexts
global_contexts = self._get_global_contexts()
# Walk up the directory tree with bounded depth
while depth < max_depth:
path_str = str(current_path)
if path_str in self.context_hierarchy:
context_node = self.context_hierarchy[path_str]
contexts.append(context_node)
inheritance_chain.append(path_str)
# Stop if this context doesn't cascade to children
if not context_node.get('applies_to_children', True):
break
# Stop if this context overrides parent
if context_node.get('overrides_parent', False):
break
# Move up one level
parent = current_path.parent
if parent == current_path: # Reached filesystem root
break
current_path = parent
depth += 1
# Add global contexts (they apply everywhere)
contexts.extend(global_contexts)
if not contexts:
return None
# Merge contexts (most specific first, global contexts last)
merged_context = self.merge_contexts(contexts)
merged_context['inheritance_chain'] = inheritance_chain
merged_context['source_path'] = inheritance_chain[0] if inheritance_chain else filesystem_path
merged_context['confidence'] = min(1.0, len(contexts) * 0.2 + 0.6) # Higher confidence with more context layers
merged_context['bounded_depth'] = depth
merged_context['global_contexts_applied'] = len(global_contexts) > 0
return merged_context
def merge_contexts(self, contexts: List[Dict[str, Any]]) -> Dict[str, Any]:
"""Merge multiple contexts using CSS-like specificity"""
if len(contexts) == 1:
return contexts[0].copy()
# Sort by specificity (higher = more specific)
contexts.sort(key=lambda c: c.get('context_specificity', 0), reverse=True)
# Start with most specific context
merged = contexts[0].copy()
# Merge in less specific contexts
for context in contexts[1:]:
# Tags are additive (union)
merged_tags = set(merged.get('tags', []))
merged_tags.update(context.get('tags', []))
merged['tags'] = list(merged_tags)
# Technologies are additive (union)
merged_tech = set(merged.get('technologies', []))
merged_tech.update(context.get('technologies', []))
merged['technologies'] = list(merged_tech)
# Insights are additive (append unique)
merged_insights = merged.get('insights', []).copy()
for insight in context.get('insights', []):
if insight not in merged_insights:
merged_insights.append(insight)
merged['insights'] = merged_insights
# Use most specific for summary/purpose unless empty
if not merged.get('summary'):
merged['summary'] = context.get('summary', '')
if not merged.get('purpose'):
merged['purpose'] = context.get('purpose', '')
return merged
def _get_global_contexts(self) -> List[Dict[str, Any]]:
"""Get contexts marked as global (apply everywhere)"""
return [context for context in self.global_contexts
if context.get('is_global', False)]
def add_global_context(self, context: Dict[str, Any]) -> None:
"""Add a context that applies globally to all paths"""
context['is_global'] = True
context['context_specificity'] = -1 # Lowest specificity
self.global_contexts.append(context)
self.clear_cache() # Clear cache since global context affects everything
def remove_global_context(self, context_id: str) -> bool:
"""Remove a global context by its ID"""
initial_count = len(self.global_contexts)
self.global_contexts = [ctx for ctx in self.global_contexts
if ctx.get('id') != context_id]
if len(self.global_contexts) < initial_count:
self.clear_cache()
return True
return False
def set_hierarchy_depth_limit(self, max_depth: int) -> None:
"""Set the maximum depth for hierarchy traversal"""
self.max_hierarchy_depth = max_depth
self.clear_cache() # Clear cache since depth limit affects resolution
def get_context_statistics(self) -> Dict[str, Any]:
"""Get statistics about the loaded context system"""
return {
'context_nodes': len(self.context_hierarchy),
'path_entries': len(self.path_metadata),
'lookup_entries': len(self.lookup_index),
'cached_resolutions': len(self.resolution_cache),
'global_contexts': len(self.global_contexts),
'max_hierarchy_depth': self.max_hierarchy_depth,
'metadata_directory': str(self.metadata_dir)
}
def clear_cache(self) -> None:
"""Clear the resolution cache"""
self.resolution_cache.clear()
def batch_resolve(self, ucxl_addresses: List[str]) -> Dict[str, ResolvedContext]:
"""Efficiently resolve multiple UCXL addresses"""
results = {}
for address in ucxl_addresses:
context = self.resolve(address)
if context:
results[address] = context
return results
def search_by_tag(self, tag: str) -> List[ResolvedContext]:
"""Find all contexts containing a specific tag"""
results = []
# Search through all path metadata
for fs_path, meta in self.path_metadata.items():
ucxl_address = meta.get('ucxl_address')
if not ucxl_address:
continue
context = self.resolve(ucxl_address)
if context and tag.lower() in [t.lower() for t in context.tags]:
results.append(context)
return results
def search_by_technology(self, technology: str) -> List[ResolvedContext]:
"""Find all contexts using a specific technology"""
results = []
for fs_path, meta in self.path_metadata.items():
ucxl_address = meta.get('ucxl_address')
if not ucxl_address:
continue
context = self.resolve(ucxl_address)
if context and technology.lower() in [t.lower() for t in context.technologies]:
results.append(context)
return results
# Example usage and demo
def demo_context_resolution():
"""Demonstrate the cascading context resolution system"""
metadata_dir = Path.home() / "chorus" / "project-metadata" / "BZZZ"
if not metadata_dir.exists():
logger.error(f"Metadata directory not found: {metadata_dir}")
logger.info("Run cascading_metadata_generator.py first to generate the hierarchy")
return
resolver = CascadingContextResolver(str(metadata_dir))
# Demo UCXL addresses
demo_addresses = [
"ucxl://any:any@BZZZ:RUSTLE-testing/src/main.rs",
"ucxl://any:any@BZZZ:RUSTLE-testing/src/api/handlers.rs",
"ucxl://any:any@BZZZ:RUSTLE-testing/tests/unit/core_tests.rs",
"ucxl://any:any@BZZZ:RUSTLE-testing/assets/fonts/README.md",
"ucxl://any:any@BZZZ:RUSTLE-testing/config/settings.toml"
]
logger.info("🔍 Context Resolution Demo")
logger.info("=" * 50)
for address in demo_addresses:
logger.info(f"\n🔗 Resolving: {address}")
context = resolver.resolve(address)
if context:
logger.info(f" 📝 Summary: {context.summary}")
logger.info(f" 🎯 Purpose: {context.purpose}")
logger.info(f" 🔧 Technologies: {', '.join(context.technologies[:3])}")
logger.info(f" 🏷️ Tags: {', '.join(context.tags[:5])}")
logger.info(f" 📊 Source: {Path(context.context_source_path).name}")
logger.info(f" 🔗 Chain: {''.join([Path(p).name for p in context.inheritance_chain])}")
logger.info(f" ✨ Confidence: {context.resolution_confidence:.2f}")
else:
logger.warning(f" ❌ Could not resolve context")
# Show statistics
stats = resolver.get_context_statistics()
logger.info(f"\n📊 System Statistics:")
logger.info(f" 🏗️ Context nodes: {stats['context_nodes']}")
logger.info(f" 📄 Path entries: {stats['path_entries']}")
logger.info(f" 🔍 Lookup entries: {stats['lookup_entries']}")
logger.info(f" 💾 Cached resolutions: {stats['cached_resolutions']}")
# Demo search functionality
logger.info(f"\n🔎 Search Demo:")
rust_contexts = resolver.search_by_technology("rust")
logger.info(f" 🦀 Found {len(rust_contexts)} Rust contexts")
source_contexts = resolver.search_by_tag("source-code")
logger.info(f" 💻 Found {len(source_contexts)} source code contexts")
if __name__ == "__main__":
logging.basicConfig(level=logging.INFO, format='%(levelname)s - %(message)s')
demo_context_resolution()

View File

@@ -1,316 +0,0 @@
# SLURP Contextual Intelligence Engine
The Contextual Intelligence Engine implements advanced analysis and processing capabilities for extracting meaningful insights from code, architecture, and project structure within the BZZZ ecosystem.
## Purpose
This module provides the "Intelligence" layer of SLURP by:
- **File Purpose Analysis**: Intelligent analysis of file roles and purposes
- **Architectural Decision Extraction**: Identifying and documenting architectural decisions
- **Cross-Component Relationship Mapping**: Understanding how components interact
- **Role-Specific Insight Generation**: Tailoring insights to specific AI agent roles
- **Project-Goal Alignment**: Ensuring context aligns with project objectives
## Architecture
The Intelligence Engine operates as a multi-layered analysis system:
```
┌─────────────────────────────────────┐
│ Role-Specific Insights │
├─────────────────────────────────────┤
│ Project Goal Alignment │
├─────────────────────────────────────┤
│ Cross-Component Relationships │
├─────────────────────────────────────┤
│ Architectural Decision Analysis │
├─────────────────────────────────────┤
│ File Purpose Analysis │
├─────────────────────────────────────┤
│ Context Data Sources │
└─────────────────────────────────────┘
```
## Key Components
### File Purpose Analysis Engine
Analyzes code files to determine their purpose and role within the system:
#### Code Pattern Recognition
- **Entry Points**: Identifies main functions, application bootstraps
- **API Endpoints**: Recognizes REST endpoints, GraphQL resolvers
- **Data Models**: Detects entity definitions, database schemas
- **Utilities**: Identifies helper functions, common utilities
- **Configuration**: Finds configuration files, environment settings
#### Language-Specific Analysis
- **Rust**: Module structure, trait implementations, macro usage
- **Go**: Package organization, interface definitions, concurrency patterns
- **JavaScript/TypeScript**: Component structure, async patterns, module exports
- **Python**: Class hierarchies, decorators, async/await patterns
#### Architectural Pattern Detection
- **MVC Patterns**: Model-View-Controller structure identification
- **Microservices**: Service boundary detection and communication patterns
- **P2P Systems**: Peer discovery, network protocols, consensus mechanisms
- **Event-Driven**: Event producers, consumers, message flows
### Architectural Decision Extraction
Identifies and documents architectural decisions embedded in code:
#### Decision Indicators
- **Design Patterns**: Observer, Factory, Strategy pattern implementations
- **Technology Choices**: Framework selections, library dependencies
- **Performance Optimizations**: Caching strategies, algorithm choices
- **Security Measures**: Authentication, encryption, access control
- **Scalability Decisions**: Load balancing, data partitioning, caching
#### Decision Documentation
```python
@dataclass
class ArchitecturalDecision:
decision_id: str
title: str
status: DecisionStatus # PROPOSED, ACCEPTED, DEPRECATED, SUPERSEDED
context: str # What is the issue?
decision: str # What is the change we're proposing/doing?
rationale: str # Why are we doing this?
consequences: List[str] # What becomes easier/harder?
alternatives: List[str] # What other options were considered?
related_decisions: List[str] # Related decision IDs
extracted_from: List[str] # UCXL addresses where evidence found
```
### Cross-Component Relationship Mapping
Maps relationships and dependencies between system components:
#### Relationship Types
- **Import Dependencies**: Direct code imports and includes
- **API Dependencies**: Service calls, endpoint usage
- **Data Dependencies**: Shared data structures, database schemas
- **Configuration Dependencies**: Shared configuration, environment variables
- **Deployment Dependencies**: Infrastructure, container dependencies
#### Relationship Analysis
```python
@dataclass
class ComponentRelationship:
from_component: str # Source UCXL address
to_component: str # Target UCXL address
relationship_type: RelationshipType
strength: float # How strong is this relationship?
confidence: float # How confident are we in this relationship?
evidence: List[str] # What code/config indicates this relationship?
impact_level: str # How much does from_component depend on to_component?
```
### Role-Specific Insight Generation
Tailors contextual insights to specific AI agent roles:
#### Role-Based Analysis Focus
**Senior Architect**
- System-wide architectural patterns
- Technology stack coherence
- Scalability and performance implications
- Technical debt identification
- Cross-system integration points
**Backend Developer**
- API design patterns
- Data flow architecture
- Service boundaries
- Performance bottlenecks
- Security implementation
**Frontend Developer**
- Component architecture
- State management patterns
- User experience flow
- Performance optimization
- Accessibility considerations
**DevOps Engineer**
- Deployment configurations
- Infrastructure dependencies
- Monitoring and logging
- Security hardening
- Scalability requirements
#### Insight Personalization
```python
def generate_role_specific_insights(context: ContextNode, role: AgentRole) -> List[str]:
insights = []
if role == AgentRole.SENIOR_ARCHITECT:
insights.extend(analyze_architectural_patterns(context))
insights.extend(identify_system_boundaries(context))
insights.extend(assess_scalability_implications(context))
elif role == AgentRole.BACKEND_DEVELOPER:
insights.extend(analyze_api_design(context))
insights.extend(identify_data_flows(context))
insights.extend(assess_performance_characteristics(context))
# ... role-specific analysis continues
return insights
```
### Project Goal Alignment
Ensures contextual intelligence aligns with current project goals and objectives:
#### Goal-Context Mapping
- **Feature Development**: Context relevance to current feature development
- **Performance Optimization**: Performance-related context for optimization goals
- **Security Hardening**: Security-relevant context for hardening initiatives
- **Technical Debt**: Technical debt context for refactoring goals
- **Documentation**: Documentation needs for knowledge sharing goals
#### Alignment Scoring
```python
@dataclass
class GoalAlignment:
goal_id: str
goal_description: str
context_relevance: float # How relevant is this context to the goal?
contribution_score: float # How much does this component contribute?
priority_weight: float # How important is this goal currently?
alignment_confidence: float # How confident are we in this alignment?
```
## Analysis Algorithms
### Contextual Similarity Analysis
Identifies similar components based on contextual features:
```python
def calculate_context_similarity(context1: ContextNode, context2: ContextNode) -> float:
# Technology stack similarity
tech_similarity = jaccard_similarity(context1.technologies, context2.technologies)
# Purpose similarity (semantic analysis)
purpose_similarity = semantic_similarity(context1.purpose, context2.purpose)
# Tag overlap
tag_similarity = jaccard_similarity(context1.tags, context2.tags)
# Architectural pattern similarity
pattern_similarity = analyze_pattern_similarity(context1, context2)
return weighted_average([
(tech_similarity, 0.3),
(purpose_similarity, 0.4),
(tag_similarity, 0.2),
(pattern_similarity, 0.1)
])
```
### Impact Analysis Engine
Predicts the impact of changes based on contextual relationships:
```python
def analyze_change_impact(target_address: str, change_type: ChangeType) -> ImpactAnalysis:
# Find all components related to target
relationships = find_all_relationships(target_address)
# Calculate impact scores based on relationship strength and change type
impacts = []
for rel in relationships:
impact_score = calculate_impact_score(rel, change_type)
impacts.append(ComponentImpact(
component=rel.to_component,
impact_type=determine_impact_type(rel, change_type),
severity=impact_score,
confidence=rel.confidence
))
return ImpactAnalysis(
target_component=target_address,
change_type=change_type,
direct_impacts=impacts,
indirect_impacts=calculate_indirect_impacts(impacts),
risk_assessment=assess_change_risk(impacts)
)
```
## Integration with BZZZ Leader System
### Leader-Coordinated Intelligence Generation
- **Centralized Analysis**: Only Leader generates contextual intelligence
- **Quality Control**: Consistent analysis quality across all contexts
- **Role-Based Distribution**: Intelligence tailored and encrypted per role
- **Priority Scheduling**: Leader schedules analysis based on project priorities
### Intelligence Update Propagation
When Leader generates new intelligence:
1. **Analysis Triggering**: Code changes, decision updates trigger re-analysis
2. **Intelligence Generation**: Deep analysis of affected components
3. **Role-Based Packaging**: Package insights for each agent role
4. **Encrypted Distribution**: Distribute encrypted intelligence via DHT
5. **Update Notifications**: Notify agents of available intelligence updates
## Performance Optimization
### Analysis Caching
- **Result Caching**: Cache analysis results for unchanged components
- **Incremental Analysis**: Only re-analyze changed components
- **Batch Processing**: Process multiple components efficiently
- **Lazy Loading**: Load detailed analysis only when requested
### Parallel Processing
- **Component-Level Parallelism**: Analyze multiple components concurrently
- **Pipeline Parallelism**: Pipeline different analysis stages
- **Resource Management**: Respect CPU/memory limits during analysis
- **Priority Queuing**: Prioritize analysis based on component importance
## Quality Assurance
### Analysis Confidence Scoring
```python
@dataclass
class AnalysisConfidence:
overall_confidence: float
code_analysis_confidence: float # How well we understand the code
pattern_recognition_confidence: float # How confident in pattern detection
relationship_confidence: float # How confident in relationships
goal_alignment_confidence: float # How confident in goal alignment
factors_considered: List[str] # What factors influenced confidence
```
### Validation Mechanisms
- **Cross-Validation**: Multiple analysis approaches for critical insights
- **Human Validation**: Mechanisms for team members to validate/correct insights
- **Consistency Checking**: Ensure insights are consistent across related components
- **Temporal Validation**: Validate insights remain accurate over time
## Future Enhancements
### Advanced AI Integration
- **Large Language Models**: Integration with LLMs for enhanced code understanding
- **Machine Learning**: ML models for pattern recognition and similarity analysis
- **Natural Language Processing**: Better extraction of intent from comments/docs
- **Computer Vision**: Analysis of architectural diagrams and documentation
### Enhanced Analysis Capabilities
- **Behavioral Analysis**: Understanding runtime behavior from code patterns
- **Performance Prediction**: Predict performance characteristics from code structure
- **Security Analysis**: Automated security vulnerability detection
- **Maintainability Scoring**: Assess code maintainability and technical debt
### Real-Time Intelligence
- **Live Analysis**: Real-time analysis of code changes
- **Predictive Insights**: Predict likely next steps in development
- **Proactive Recommendations**: Suggest improvements before problems arise
- **Continuous Learning**: System learns and improves analysis over time

View File

@@ -1,179 +0,0 @@
# SLURP Storage Architecture
The Storage Architecture component implements the "Storage" aspect of SLURP, providing efficient, encrypted, and distributed storage for contextual intelligence data within the BZZZ ecosystem.
## Purpose
This module handles:
- **Context Storage**: Persistent storage of hierarchical context metadata
- **Encrypted Storage**: Role-based encryption for secure context distribution
- **Distributed Architecture**: Integration with BZZZ DHT for network-wide access
- **Version Management**: Temporal versioning of context evolution
- **Efficient Querying**: Fast lookup and retrieval systems
## Architecture Components
### Core Storage Systems
#### Context Database Schema
- **Hierarchical Storage**: Tree-structured context inheritance
- **Version Control**: Temporal evolution tracking
- **Encryption Layers**: Per-role encryption boundaries
- **Index Structures**: Fast lookup and search capabilities
#### Distributed Hash Table Integration
- **DHT Storage**: Leverages existing BZZZ DHT infrastructure
- **Replication**: Context data replicated across cluster nodes
- **Consistency**: Leader-coordinated updates ensure consistency
- **Fault Tolerance**: Automatic failover and recovery
### Storage Layers
```
┌─────────────────────────────────────┐
│ Application Layer │
├─────────────────────────────────────┤
│ Role-Based Encryption │
├─────────────────────────────────────┤
│ Context Serialization │
├─────────────────────────────────────┤
│ Distributed Hash Table │
├─────────────────────────────────────┤
│ Network Transport Layer │
└─────────────────────────────────────┘
```
## Key Features
### Hierarchical Context Storage
- **Cascading Metadata**: CSS-like inheritance reduces storage overhead
- **Differential Storage**: Only stores unique/changed context per level
- **Compression**: Intelligent deduplication and compression
- **Space Efficiency**: 85%+ space savings vs traditional metadata
### Role-Based Encryption
- **Per-Role Keys**: Each AI agent role has unique encryption keys
- **Need-to-Know Access**: Agents only decrypt relevant context
- **Key Rotation**: Automated key management and rotation
- **Shamir's Secret Sharing**: Distributed key management
### Temporal Versioning
- **Decision-Based Versions**: Tracks context evolution through decisions
- **Branching History**: Supports parallel context evolution
- **Rollback Capability**: Can restore previous context versions
- **Change Attribution**: Links changes to specific decisions/commits
## Storage Schema
### Context Node Storage
```json
{
"ucxl_address": "ucxl://agent:role@project:task/path",
"context_data": {
"summary": "...",
"purpose": "...",
"technologies": [...],
"tags": [...],
"insights": [...]
},
"hierarchy_metadata": {
"parent_context": "...",
"child_contexts": [...],
"inheritance_depth": 3,
"specificity_score": 0.8
},
"encryption_metadata": {
"encrypted_for_roles": [...],
"encryption_version": 1,
"key_derivation": "..."
},
"temporal_metadata": {
"version": 3,
"parent_version": 2,
"created_at": "...",
"created_by": "...",
"change_reason": "architecture_change"
}
}
```
### Index Structures
- **UCXL Address Index**: Fast lookup by address
- **Tag Index**: Search by context tags
- **Technology Index**: Search by technology stack
- **Role Index**: Find contexts accessible to specific roles
- **Temporal Index**: Navigate context evolution history
## Integration Points
### BZZZ DHT Integration
- **Store Operations**: Encrypted context storage in DHT
- **Retrieve Operations**: Fast context retrieval with caching
- **Update Operations**: Leader-coordinated context updates
- **Replication**: Automatic data replication across nodes
### Leader Election System
- **Context Generation Authority**: Only Leader generates context
- **Update Coordination**: Leader coordinates all context updates
- **Failover Handling**: Context generation transfers with leadership
- **Consistency Guarantees**: Single source of truth maintenance
### Crypto Infrastructure
- **Encryption Integration**: Uses existing BZZZ crypto systems
- **Key Management**: Integrates with Shamir's Secret Sharing
- **Access Control**: Role-based decryption capabilities
- **Audit Trail**: Encrypted access logging
## Performance Characteristics
### Storage Efficiency
- **Space Savings**: 85%+ reduction vs traditional metadata
- **Compression Ratio**: Average 10:1 through intelligent deduplication
- **Network Bandwidth**: Minimal through differential updates
- **Disk I/O**: Optimized through caching and batching
### Query Performance
- **Lookup Speed**: O(log n) average case with indexing
- **Search Performance**: Sub-second tag/technology searches
- **Hierarchy Resolution**: Bounded depth prevents excessive traversal
- **Cache Hit Rate**: >90% for frequently accessed contexts
## Security Model
### Encryption Strategy
- **Multi-Layer Encryption**: Base context + role-specific overlays
- **Key Derivation**: From role definitions and Shamir shares
- **Access Logging**: Complete audit trail of context access
- **Compartmentalization**: Prevents cross-role information leakage
### Access Control Matrix
| Role | Access Level | Encryption | Scope |
|------|--------------|------------|--------|
| Senior Architect | Full System Context | High | System-wide |
| Frontend Developer | UI/UX Context | Medium | Frontend scope |
| Backend Developer | API/Service Context | Medium | Backend scope |
| DevOps Engineer | Infrastructure Context | High | Infrastructure |
| Project Manager | Coordination Context | Highest | Global |
## Monitoring and Maintenance
### Health Monitoring
- **Storage Capacity**: Track available storage across nodes
- **Replication Status**: Monitor data replication health
- **Access Patterns**: Analyze context access patterns
- **Performance Metrics**: Query latency and throughput monitoring
### Maintenance Operations
- **Garbage Collection**: Clean up orphaned context versions
- **Index Optimization**: Rebuild and optimize search indexes
- **Key Rotation**: Automated encryption key rotation
- **Backup Operations**: Regular encrypted backup creation
## Future Enhancements
- **Advanced Compression**: ML-based context compression
- **Smart Caching**: Predictive context caching based on usage patterns
- **Cross-Cluster Replication**: Context sharing across BZZZ clusters
- **Real-time Updates**: WebSocket-based context update notifications
- **Analytics Dashboard**: Context usage and health visualization

View File

@@ -1,247 +0,0 @@
# SLURP Temporal Decision Graph System
The Temporal Decision Graph System implements decision-hop based temporal analysis, tracking how context and decisions evolve through decision relationships rather than chronological time.
## Purpose
This module provides:
- **Decision-Based Temporal Analysis**: Track related decisions by decision distance, not time
- **Context Evolution**: Monitor how understanding deepens over time
- **Decision Influence Graphs**: Map how decisions affect other decisions
- **Temporal Queries**: Query context "as it was" at any decision point
- **Decision Genealogy**: Track the lineage and rationale of decisions
## Key Innovation: Decision Hops vs Time
Traditional temporal systems track changes chronologically. SLURP's temporal system tracks changes by **decision relationships**:
- **Decision Hop 0**: Original context/decision
- **Decision Hop 1**: Decisions directly influenced by the original
- **Decision Hop 2**: Decisions influenced by hop-1 decisions
- **Decision Hop N**: N-degrees of decision influence away
This provides **conceptual relevance** similar to RAG systems - finding related decisions by influence rather than time proximity.
## Architecture
### Core Components
#### TemporalContextNode
```python
@dataclass
class TemporalContextNode:
ucxl_address: str # What file/component this affects
timestamp: str # When the decision occurred
version: int # Monotonic version number
# Context at this decision point
summary: str
purpose: str
technologies: List[str]
tags: List[str]
insights: List[str]
# Decision metadata
change_reason: ContextChangeReason # Why context changed
parent_version: Optional[int] # Previous version
decision_metadata: DecisionMetadata # Who, why, how confident
# Decision influence graph
influences: List[str] # UCXL addresses this decision influences
influenced_by: List[str] # UCXL addresses that influenced this
```
#### DecisionMetadata
```python
@dataclass
class DecisionMetadata:
decision_maker: str # Who made the decision
decision_id: Optional[str] # Git commit, ticket ID, etc.
decision_rationale: str # Why the decision was made
impact_scope: str # local, module, project, system
confidence_level: float # How confident in this decision
external_references: List[str] # Links to PRs, issues, docs
```
### Decision Change Reasons
The system tracks **why** context changes:
- `INITIAL_CREATION`: First context creation
- `CODE_CHANGE`: File/code modifications
- `DESIGN_DECISION`: Architectural or design choices
- `REFACTORING`: Code structure improvements
- `ARCHITECTURE_CHANGE`: System-wide architectural changes
- `REQUIREMENTS_CHANGE`: Changed requirements/specifications
- `LEARNING_EVOLUTION`: Improved understanding over time
- `RAG_ENHANCEMENT`: RAG system provided better insights
- `TEAM_INPUT`: Human team member input/corrections
- `BUG_DISCOVERY`: Found issues that change understanding
- `PERFORMANCE_INSIGHT`: Performance analysis changed context
- `SECURITY_REVIEW`: Security analysis added context
## Key Features
### Decision-Hop Analysis
```python
# Find all decisions within 3 hops of a component
related_decisions = temporal_graph.find_related_decisions(
"ucxl://any:any@BZZZ:RUSTLE-testing/src/main.rs",
max_hops=3
)
# Get decision timeline (by decision influence, not time)
timeline = temporal_graph.get_decision_timeline(
ucxl_address,
include_related=True,
max_hops=2
)
```
### Decision Path Discovery
```python
# Find shortest decision path between two components
path = temporal_graph.find_decision_path(
from_address="ucxl://any:any@BZZZ:RUSTLE-testing/src/main.rs",
to_address="ucxl://any:any@BZZZ:RUSTLE-testing/src/api.rs"
)
```
### Context Time Travel
```python
# Get context as it was at a specific decision point
historical_context = temporal_graph.get_version_at_time(
ucxl_address,
target_time=datetime(2024, 1, 15)
)
# Get complete evolution history
evolution = temporal_graph.get_context_evolution(ucxl_address)
```
### Confidence Evolution Tracking
The system tracks how confidence in context changes over time:
```python
confidence_adjustments = {
RAG_ENHANCEMENT: +0.1, # RAG usually improves confidence
TEAM_INPUT: +0.15, # Human input is valuable
LEARNING_EVOLUTION: +0.1, # Learning improves understanding
BUG_DISCOVERY: -0.15, # Bugs indicate misunderstanding
ARCHITECTURE_CHANGE: -0.05, # Major changes create uncertainty
}
```
## Decision Influence Propagation
When a decision changes context, it propagates influence to related components:
1. **Direct Influence**: Components directly affected by the decision
2. **Indirect Influence**: Components affected by the directly influenced ones
3. **Staleness Propagation**: Increases staleness indicator for affected contexts
4. **Re-analysis Triggers**: May trigger automated re-analysis of stale contexts
### Staleness Calculation
```python
def calculate_staleness_impact(reason, impact_scope):
base_staleness = {
"local": 0.1,
"module": 0.3,
"project": 0.5,
"system": 0.8
}[impact_scope]
reason_multipliers = {
ARCHITECTURE_CHANGE: 2.0,
REQUIREMENTS_CHANGE: 1.5,
REFACTORING: 1.2,
CODE_CHANGE: 1.0
}
return base_staleness * reason_multipliers.get(reason, 1.0)
```
## Integration with BZZZ Leader System
### Leader-Coordinated Temporal Updates
- **Leader Authority**: Only Leader can create new temporal context versions
- **Decision Validation**: Leader validates and approves decision metadata
- **Consistency Guarantee**: Single source of truth for decision evolution
- **Failover Handling**: Temporal graph transfers with leadership election
### Role-Based Temporal Access
- **Decision Visibility**: Agents see temporal context relevant to their role
- **Encrypted Temporal Data**: Decision history encrypted per role
- **Need-to-Know Temporal**: Only relevant decision paths visible
- **Audit Trail**: Complete log of temporal context access
## Query Patterns
### Temporal Context Queries
```python
# What decisions led to this current state?
decision_path = get_decision_ancestry(ucxl_address)
# What would be affected if I change this component?
impact_analysis = find_decision_influence_cone(ucxl_address, max_depth=5)
# Show me all decisions made by this person/system
maker_decisions = filter_decisions_by_maker("tony")
# Find components that evolved similarly to this one
similar_evolution = find_similar_decision_patterns(ucxl_address)
```
### Decision Pattern Analysis
```python
# Analyze decision-making patterns over time
patterns = temporal_graph.analyze_decision_patterns()
# Returns:
# - decision_makers: Who makes what types of decisions
# - change_reasons: Most common reasons for context changes
# - impact_scopes: Distribution of decision impact levels
# - decision_frequency: Decision activity over time
```
## Performance Characteristics
### Decision Graph Traversal
- **Bounded Depth**: Configurable maximum decision hop depth
- **Efficient Pathfinding**: Breadth-first search with early termination
- **Cached Relationships**: In-memory caching of decision relationships
- **Lazy Loading**: Load decision details only when needed
### Storage Efficiency
- **Differential Versioning**: Only store changes between versions
- **Decision Deduplication**: Reuse common decision metadata
- **Compressed History**: Compress old temporal data
- **Index Optimization**: Efficient indexing for temporal queries
## Use Cases
### Architectural Decision Tracking
- **Why was this design chosen?**: Trace back through decision genealogy
- **What else was affected?**: Find all components influenced by a decision
- **Who should review changes?**: Identify decision makers for related components
- **Risk Assessment**: Understand decision confidence and validation history
### Impact Analysis
- **Change Impact**: What decisions would be affected by this change?
- **Regression Risk**: Has similar context been changed before? What happened?
- **Knowledge Transfer**: New team members can see decision rationale history
- **Technical Debt**: Identify areas with low-confidence or outdated decisions
### Learning and Optimization
- **Decision Quality**: Track which types of decisions lead to better outcomes
- **Team Patterns**: Understand how different team members make decisions
- **Context Staleness**: Identify areas needing decision review/update
- **Knowledge Evolution**: See how understanding improved over time
## Future Enhancements
- **ML-Based Decision Prediction**: Predict likely next decisions based on patterns
- **Automated Staleness Detection**: ML models to identify potentially stale context
- **Decision Recommendation**: Suggest reviewers based on decision history
- **Cross-Project Decision Learning**: Learn from decisions across multiple projects
- **Real-time Decision Streaming**: WebSocket-based decision update notifications

View File

@@ -1,720 +0,0 @@
#!/usr/bin/env python3
"""
Temporal Context Evolution System
Implements a temporal graph network for tracking how context and decisions evolve over time.
Key concepts:
- Temporal Nodes: Context at specific points in time
- Decision Points: When context changes due to decisions/events
- Context Evolution: How understanding deepens over time
- Temporal Queries: Query context "as it was" at any point in time
- Decision Tracking: Why context changed (git commits, design decisions, etc.)
This addresses your insight that decisions change over time and we need to track
the evolution of understanding and context within the project.
Architecture:
```
TemporalContextNode
├── timestamp: when this context was valid
├── context_data: the actual context at this time
├── change_reason: why this context changed
├── parent_version: previous version of this context
├── decision_metadata: what decision caused this change
└── confidence_evolution: how confidence changed over time
```
"""
import json
from pathlib import Path
from typing import Dict, List, Optional, Any, Tuple, Set
from datetime import datetime, timezone, timedelta
from dataclasses import dataclass, asdict
from enum import Enum
import logging
import hashlib
logger = logging.getLogger(__name__)
class ContextChangeReason(Enum):
"""Reasons why context might change over time"""
INITIAL_CREATION = "initial_creation"
CODE_CHANGE = "code_change"
DESIGN_DECISION = "design_decision"
REFACTORING = "refactoring"
ARCHITECTURE_CHANGE = "architecture_change"
REQUIREMENTS_CHANGE = "requirements_change"
LEARNING_EVOLUTION = "learning_evolution" # Understanding improved
RAG_ENHANCEMENT = "rag_enhancement" # RAG provided better insights
TEAM_INPUT = "team_input" # Team provided corrections/additions
BUG_DISCOVERY = "bug_discovery" # Found issues that change understanding
PERFORMANCE_INSIGHT = "performance_insight" # Performance analysis changed understanding
SECURITY_REVIEW = "security_review" # Security analysis added context
@dataclass
class DecisionMetadata:
"""Metadata about a decision that changed context"""
decision_maker: str # Who made the decision (person, system, process)
decision_id: Optional[str] # Git commit hash, ticket ID, etc.
decision_rationale: str # Why the decision was made
impact_scope: str # Local, module, project, system
confidence_level: float # How confident we are in this decision
external_references: List[str] # Links to PRs, issues, docs, etc.
@dataclass
class TemporalContextNode:
"""A context node at a specific point in time"""
ucxl_address: str
timestamp: str # ISO format
version: int # Monotonic version number
# Core context data
summary: str
purpose: str
technologies: List[str]
tags: List[str]
insights: List[str]
# Temporal metadata
change_reason: ContextChangeReason
parent_version: Optional[int] # Previous version
decision_metadata: Optional[DecisionMetadata]
# Evolution tracking
context_hash: str # Hash of the context content
confidence_score: float # How confident we are in this context
staleness_indicator: float # How likely this context is outdated
# Graph relationships
influences: List[str] # Other UCXL addresses this context influences
influenced_by: List[str] # Other UCXL addresses that influence this
# Validation metadata
validated_by: List[str] # Who/what validated this context
last_validation: Optional[str] # When it was last validated
class TemporalContextGraph:
"""Manages the temporal evolution of context across the project"""
def __init__(self, metadata_base: str, project_name: str):
self.metadata_base = Path(metadata_base)
self.project_name = project_name
self.temporal_dir = self.metadata_base / project_name / "temporal"
self.temporal_dir.mkdir(parents=True, exist_ok=True)
# In-memory graph
self.temporal_nodes: Dict[str, List[TemporalContextNode]] = {} # ucxl_address -> [nodes]
self.decision_log: List[DecisionMetadata] = []
self.influence_graph: Dict[str, Set[str]] = {} # ucxl_address -> influenced addresses
self.load_temporal_data()
def create_initial_context(self, ucxl_address: str, context_data: Dict[str, Any],
creator: str = "system") -> TemporalContextNode:
"""Create the first version of context for a UCXL address"""
decision = DecisionMetadata(
decision_maker=creator,
decision_id=None,
decision_rationale="Initial context creation",
impact_scope="local",
confidence_level=0.7,
external_references=[]
)
node = TemporalContextNode(
ucxl_address=ucxl_address,
timestamp=datetime.now(timezone.utc).isoformat(),
version=1,
summary=context_data.get('summary', ''),
purpose=context_data.get('purpose', ''),
technologies=context_data.get('technologies', []),
tags=context_data.get('tags', []),
insights=context_data.get('insights', []),
change_reason=ContextChangeReason.INITIAL_CREATION,
parent_version=None,
decision_metadata=decision,
context_hash=self.calculate_context_hash(context_data),
confidence_score=0.7,
staleness_indicator=0.0,
influences=[],
influenced_by=[],
validated_by=[creator],
last_validation=datetime.now(timezone.utc).isoformat()
)
# Add to graph
if ucxl_address not in self.temporal_nodes:
self.temporal_nodes[ucxl_address] = []
self.temporal_nodes[ucxl_address].append(node)
self.decision_log.append(decision)
return node
def evolve_context(self, ucxl_address: str, new_context_data: Dict[str, Any],
change_reason: ContextChangeReason, decision_metadata: DecisionMetadata) -> TemporalContextNode:
"""Evolve context for a UCXL address - create new temporal version"""
# Get current version
current_version = self.get_latest_version(ucxl_address)
if not current_version:
return self.create_initial_context(ucxl_address, new_context_data, decision_metadata.decision_maker)
# Calculate changes and impact
context_hash = self.calculate_context_hash(new_context_data)
confidence_evolution = self.calculate_confidence_evolution(current_version, new_context_data, change_reason)
# Create new version
new_node = TemporalContextNode(
ucxl_address=ucxl_address,
timestamp=datetime.now(timezone.utc).isoformat(),
version=current_version.version + 1,
summary=new_context_data.get('summary', current_version.summary),
purpose=new_context_data.get('purpose', current_version.purpose),
technologies=new_context_data.get('technologies', current_version.technologies),
tags=new_context_data.get('tags', current_version.tags),
insights=new_context_data.get('insights', current_version.insights),
change_reason=change_reason,
parent_version=current_version.version,
decision_metadata=decision_metadata,
context_hash=context_hash,
confidence_score=confidence_evolution,
staleness_indicator=0.0, # Fresh context
influences=current_version.influences.copy(),
influenced_by=current_version.influenced_by.copy(),
validated_by=[decision_metadata.decision_maker],
last_validation=datetime.now(timezone.utc).isoformat()
)
self.temporal_nodes[ucxl_address].append(new_node)
self.decision_log.append(decision_metadata)
# Propagate influence
self.propagate_context_change(ucxl_address, change_reason, decision_metadata)
return new_node
def calculate_context_hash(self, context_data: Dict[str, Any]) -> str:
"""Calculate hash of context content for change detection"""
content = json.dumps(context_data, sort_keys=True)
return hashlib.sha256(content.encode()).hexdigest()[:16]
def calculate_confidence_evolution(self, current: TemporalContextNode,
new_data: Dict[str, Any], reason: ContextChangeReason) -> float:
"""Calculate how confidence should evolve based on the type of change"""
base_confidence = current.confidence_score
# Confidence adjustments based on change reason
confidence_adjustments = {
ContextChangeReason.RAG_ENHANCEMENT: 0.1, # RAG usually improves confidence
ContextChangeReason.TEAM_INPUT: 0.15, # Human input is valuable
ContextChangeReason.LEARNING_EVOLUTION: 0.1, # Learning improves understanding
ContextChangeReason.DESIGN_DECISION: 0.05, # Decisions clarify purpose
ContextChangeReason.ARCHITECTURE_CHANGE: -0.05, # Major changes create uncertainty
ContextChangeReason.REQUIREMENTS_CHANGE: -0.1, # Requirements changes create uncertainty
ContextChangeReason.BUG_DISCOVERY: -0.15, # Bugs indicate misunderstanding
ContextChangeReason.CODE_CHANGE: 0.02, # Code changes slightly improve confidence
ContextChangeReason.REFACTORING: 0.05, # Refactoring clarifies intent
}
adjustment = confidence_adjustments.get(reason, 0.0)
# Additional confidence boost if multiple sources agree
if len(new_data.get('insights', [])) > len(current.insights):
adjustment += 0.05
return min(1.0, max(0.1, base_confidence + adjustment))
def propagate_context_change(self, changed_address: str, reason: ContextChangeReason,
decision: DecisionMetadata) -> None:
"""Propagate context changes to influenced addresses"""
# Get addresses influenced by this change
influenced = self.influence_graph.get(changed_address, set())
for influenced_address in influenced:
# Increase staleness indicator for influenced contexts
latest = self.get_latest_version(influenced_address)
if latest:
# Create a "staleness update" - not a full context change
staleness_increase = self.calculate_staleness_impact(reason, decision.impact_scope)
# This could trigger a re-analysis of the influenced context
logger.info(f"Context change in {changed_address} affects {influenced_address} (staleness +{staleness_increase:.2f})")
def calculate_staleness_impact(self, reason: ContextChangeReason, impact_scope: str) -> float:
"""Calculate how much a change affects staleness of related contexts"""
base_staleness = {
"local": 0.1,
"module": 0.3,
"project": 0.5,
"system": 0.8
}.get(impact_scope, 0.2)
reason_multipliers = {
ContextChangeReason.ARCHITECTURE_CHANGE: 2.0,
ContextChangeReason.REQUIREMENTS_CHANGE: 1.5,
ContextChangeReason.REFACTORING: 1.2,
ContextChangeReason.CODE_CHANGE: 1.0,
ContextChangeReason.DESIGN_DECISION: 1.3,
}
multiplier = reason_multipliers.get(reason, 1.0)
return base_staleness * multiplier
def get_latest_version(self, ucxl_address: str) -> Optional[TemporalContextNode]:
"""Get the most recent version of context for an address"""
versions = self.temporal_nodes.get(ucxl_address, [])
if not versions:
return None
return max(versions, key=lambda n: n.version)
def get_version_at_time(self, ucxl_address: str, target_time: datetime) -> Optional[TemporalContextNode]:
"""Get the context as it was at a specific point in time"""
versions = self.temporal_nodes.get(ucxl_address, [])
if not versions:
return None
# Find the latest version before or at the target time
target_iso = target_time.isoformat()
valid_versions = [v for v in versions if v.timestamp <= target_iso]
if not valid_versions:
return None
return max(valid_versions, key=lambda n: n.timestamp)
def get_context_evolution(self, ucxl_address: str) -> List[TemporalContextNode]:
"""Get the complete evolution history of a context"""
versions = self.temporal_nodes.get(ucxl_address, [])
return sorted(versions, key=lambda n: n.version)
def add_influence_relationship(self, influencer: str, influenced: str,
relationship_type: str = "affects") -> None:
"""Add an influence relationship between contexts"""
if influencer not in self.influence_graph:
self.influence_graph[influencer] = set()
self.influence_graph[influencer].add(influenced)
# Update the influenced_by lists in the temporal nodes
influencer_latest = self.get_latest_version(influencer)
influenced_latest = self.get_latest_version(influenced)
if influencer_latest and influenced not in influencer_latest.influences:
influencer_latest.influences.append(influenced)
if influenced_latest and influencer not in influenced_latest.influenced_by:
influenced_latest.influenced_by.append(influencer)
def analyze_decision_patterns(self) -> Dict[str, Any]:
"""Analyze patterns in decision-making over time"""
analysis = {
'total_decisions': len(self.decision_log),
'decision_makers': {},
'change_reasons': {},
'impact_scopes': {},
'confidence_trends': [],
'decision_frequency': {} # decisions per time period
}
# Analyze decision makers
for decision in self.decision_log:
maker = decision.decision_maker
analysis['decision_makers'][maker] = analysis['decision_makers'].get(maker, 0) + 1
# Analyze change reasons
for address, versions in self.temporal_nodes.items():
for version in versions:
reason = version.change_reason.value
analysis['change_reasons'][reason] = analysis['change_reasons'].get(reason, 0) + 1
# Analyze impact scopes
for decision in self.decision_log:
scope = decision.impact_scope
analysis['impact_scopes'][scope] = analysis['impact_scopes'].get(scope, 0) + 1
return analysis
def find_related_decisions(self, ucxl_address: str, max_hops: int = 3) -> List[Tuple[str, TemporalContextNode, int]]:
"""Find decisions that are x number of decision-hops away from a given address"""
visited = set()
decision_queue = [(ucxl_address, 0)] # (address, hop_distance)
related_decisions = []
while decision_queue and len(decision_queue) > 0:
current_address, hop_distance = decision_queue.pop(0)
if current_address in visited or hop_distance > max_hops:
continue
visited.add(current_address)
# Get latest version for this address
latest_version = self.get_latest_version(current_address)
if latest_version:
related_decisions.append((current_address, latest_version, hop_distance))
# Add influenced addresses to queue (1 hop further)
for influenced_addr in latest_version.influences:
if influenced_addr not in visited:
decision_queue.append((influenced_addr, hop_distance + 1))
# Add addresses that influence this one (1 hop further)
for influencer_addr in latest_version.influenced_by:
if influencer_addr not in visited:
decision_queue.append((influencer_addr, hop_distance + 1))
# Sort by hop distance, then by decision recency
return sorted(related_decisions, key=lambda x: (x[2], -x[1].version))
def find_decision_path(self, from_address: str, to_address: str) -> Optional[List[Tuple[str, TemporalContextNode]]]:
"""Find the shortest decision path between two UCXL addresses"""
if from_address == to_address:
latest = self.get_latest_version(from_address)
return [(from_address, latest)] if latest else None
visited = set()
queue = [(from_address, [])] # (current_address, path)
while queue:
current_address, path = queue.pop(0)
if current_address in visited:
continue
visited.add(current_address)
latest_version = self.get_latest_version(current_address)
if not latest_version:
continue
new_path = path + [(current_address, latest_version)]
if current_address == to_address:
return new_path
# Explore influences (decisions made here affect these addresses)
for influenced_addr in latest_version.influences:
if influenced_addr not in visited:
queue.append((influenced_addr, new_path))
# Explore influenced_by (decisions made there affect this address)
for influencer_addr in latest_version.influenced_by:
if influencer_addr not in visited:
queue.append((influencer_addr, new_path))
return None # No path found
def get_decision_timeline(self, ucxl_address: str, include_related: bool = False, max_hops: int = 2) -> Dict[str, Any]:
"""Get decision timeline showing how context evolved through decisions (not time)"""
timeline = {
'primary_address': ucxl_address,
'decision_sequence': [],
'related_decisions': [] if include_related else None
}
# Get primary decision sequence
versions = self.get_context_evolution(ucxl_address)
for i, version in enumerate(versions):
decision_info = {
'version': version.version,
'decision_hop': i, # Decision distance from initial context
'change_reason': version.change_reason.value,
'decision_maker': version.decision_metadata.decision_maker if version.decision_metadata else 'unknown',
'decision_rationale': version.decision_metadata.decision_rationale if version.decision_metadata else '',
'confidence_evolution': version.confidence_score,
'timestamp': version.timestamp,
'influences_count': len(version.influences),
'influenced_by_count': len(version.influenced_by)
}
timeline['decision_sequence'].append(decision_info)
# Get related decisions if requested
if include_related and max_hops > 0:
related = self.find_related_decisions(ucxl_address, max_hops)
for addr, version, hops in related:
if addr != ucxl_address: # Don't include the primary address
related_info = {
'address': addr,
'decision_hops': hops,
'latest_version': version.version,
'change_reason': version.change_reason.value,
'decision_maker': version.decision_metadata.decision_maker if version.decision_metadata else 'unknown',
'confidence': version.confidence_score,
'last_decision_timestamp': version.timestamp
}
timeline['related_decisions'].append(related_info)
return timeline
def find_stale_contexts(self, staleness_threshold: float = 0.5) -> List[Tuple[str, TemporalContextNode]]:
"""Find contexts that might be outdated"""
stale_contexts = []
current_time = datetime.now(timezone.utc)
for address, versions in self.temporal_nodes.items():
latest = self.get_latest_version(address)
if not latest:
continue
# Calculate time-based staleness
last_update = datetime.fromisoformat(latest.timestamp.replace('Z', '+00:00'))
days_old = (current_time - last_update).days
time_staleness = min(1.0, days_old / 30.0) # Max staleness after 30 days
total_staleness = latest.staleness_indicator + time_staleness
if total_staleness >= staleness_threshold:
stale_contexts.append((address, latest))
return sorted(stale_contexts, key=lambda x: x[1].staleness_indicator, reverse=True)
def save_temporal_data(self) -> None:
"""Save temporal graph data to files"""
# Save temporal nodes
nodes_file = self.temporal_dir / "temporal_nodes.json"
nodes_data = {}
for address, versions in self.temporal_nodes.items():
nodes_data[address] = [asdict(version) for version in versions]
with open(nodes_file, 'w') as f:
json.dump(nodes_data, f, indent=2, default=str)
# Save decision log
decisions_file = self.temporal_dir / "decision_log.json"
decisions_data = [asdict(decision) for decision in self.decision_log]
with open(decisions_file, 'w') as f:
json.dump(decisions_data, f, indent=2, default=str)
# Save influence graph
influence_file = self.temporal_dir / "influence_graph.json"
influence_data = {k: list(v) for k, v in self.influence_graph.items()}
with open(influence_file, 'w') as f:
json.dump(influence_data, f, indent=2)
logger.info(f"Temporal data saved to {self.temporal_dir}")
def load_temporal_data(self) -> None:
"""Load temporal graph data from files"""
try:
# Load temporal nodes
nodes_file = self.temporal_dir / "temporal_nodes.json"
if nodes_file.exists():
with open(nodes_file, 'r') as f:
nodes_data = json.load(f)
for address, versions_data in nodes_data.items():
versions = []
for version_data in versions_data:
# Convert change_reason back to enum
version_data['change_reason'] = ContextChangeReason(version_data['change_reason'])
# Convert decision_metadata back to dataclass
if version_data['decision_metadata']:
version_data['decision_metadata'] = DecisionMetadata(**version_data['decision_metadata'])
versions.append(TemporalContextNode(**version_data))
self.temporal_nodes[address] = versions
# Load decision log
decisions_file = self.temporal_dir / "decision_log.json"
if decisions_file.exists():
with open(decisions_file, 'r') as f:
decisions_data = json.load(f)
self.decision_log = [DecisionMetadata(**d) for d in decisions_data]
# Load influence graph
influence_file = self.temporal_dir / "influence_graph.json"
if influence_file.exists():
with open(influence_file, 'r') as f:
influence_data = json.load(f)
self.influence_graph = {k: set(v) for k, v in influence_data.items()}
logger.info(f"Loaded temporal data: {len(self.temporal_nodes)} addresses, {len(self.decision_log)} decisions")
except Exception as e:
logger.warning(f"Error loading temporal data: {e}")
# Example usage and integration
def demo_temporal_context():
"""Demonstrate the temporal context evolution system"""
temporal_graph = TemporalContextGraph(
metadata_base=str(Path.home() / "chorus" / "project-metadata"),
project_name="BZZZ"
)
# Example: Context evolution for a main.rs file
ucxl_address = "ucxl://any:any@BZZZ:RUSTLE-testing/src/main.rs"
# Initial context
initial_context = {
'summary': 'Main entry point for BZZZ application',
'purpose': 'Application startup and initialization',
'technologies': ['Rust'],
'tags': ['main', 'entry-point', 'rust'],
'insights': ['Basic application entry point']
}
# Create initial version
initial_node = temporal_graph.create_initial_context(
ucxl_address, initial_context, "system_analysis"
)
logger.info(f"Created initial context v{initial_node.version}")
# Evolution 1: RAG enhancement
rag_decision = DecisionMetadata(
decision_maker="rag_system",
decision_id="rag_001",
decision_rationale="RAG analysis provided enhanced understanding of application structure",
impact_scope="module",
confidence_level=0.8,
external_references=["rag_analysis_log"]
)
enhanced_context = {
'summary': 'Main entry point implementing BZZZ distributed system bootstrap',
'purpose': 'Application startup, configuration loading, and P2P network initialization',
'technologies': ['Rust', 'Tokio', 'P2P Networking'],
'tags': ['main', 'entry-point', 'rust', 'distributed-system', 'p2p'],
'insights': [
'Initializes distributed hash table for metadata storage',
'Sets up P2P networking for node discovery',
'Loads configuration for hybrid mock/real backend switching'
]
}
enhanced_node = temporal_graph.evolve_context(
ucxl_address, enhanced_context, ContextChangeReason.RAG_ENHANCEMENT, rag_decision
)
logger.info(f"Enhanced context to v{enhanced_node.version} (confidence: {enhanced_node.confidence_score:.2f})")
# Evolution 2: Architecture change
arch_decision = DecisionMetadata(
decision_maker="tony",
decision_id="commit_abc123",
decision_rationale="Refactored to use cascading context hierarchy instead of flat metadata",
impact_scope="system",
confidence_level=0.9,
external_references=["github.com/project/commit/abc123"]
)
arch_context = enhanced_context.copy()
arch_context['insights'].append('Now supports hierarchical context inheritance for efficient metadata storage')
arch_context['tags'].append('hierarchical-context')
arch_node = temporal_graph.evolve_context(
ucxl_address, arch_context, ContextChangeReason.ARCHITECTURE_CHANGE, arch_decision
)
logger.info(f"Architecture update to v{arch_node.version} (confidence: {arch_node.confidence_score:.2f})")
# Demonstrate decision-hop based temporal analysis
logger.info(f"\n📈 Decision Evolution for {ucxl_address}:")
# Show decision timeline (decisions, not time)
decision_timeline = temporal_graph.get_decision_timeline(ucxl_address, include_related=True, max_hops=2)
logger.info(f" 🎯 Primary Decision Sequence:")
for decision in decision_timeline['decision_sequence']:
logger.info(f" Decision #{decision['decision_hop']}: {decision['change_reason']}")
logger.info(f" 👤 By: {decision['decision_maker']}")
logger.info(f" 💪 Confidence: {decision['confidence_evolution']:.2f}")
logger.info(f" 📊 Influences {decision['influences_count']} addresses")
if decision_timeline['related_decisions']:
logger.info(f"\n 🔗 Related Decisions (within 2 hops):")
for related in decision_timeline['related_decisions'][:3]: # Show first 3
logger.info(f" {related['address']} ({related['decision_hops']} hops away)")
logger.info(f" 🔄 Latest: {related['change_reason']} by {related['decision_maker']}")
logger.info(f" 📊 Confidence: {related['confidence']:.2f}")
# Demonstrate decision path finding
sample_addresses = [
"ucxl://any:any@BZZZ:RUSTLE-testing/src/api.rs",
"ucxl://any:any@BZZZ:RUSTLE-testing/src/lib.rs"
]
# Create additional contexts for demo
for addr in sample_addresses:
if addr != ucxl_address:
context = {
'summary': f'Related component: {addr.split("/")[-1]}',
'purpose': 'Supporting component in BZZZ architecture',
'technologies': ['Rust', 'BZZZ Protocol'],
'tags': ['component', 'rust'],
'insights': [f'Related to main entry point through architectural decisions']
}
decision = DecisionMetadata(
decision_maker="system_analysis",
decision_id="related_001",
decision_rationale="Architectural relationship analysis",
impact_scope="module",
confidence_level=0.7,
external_references=[]
)
related_node = temporal_graph.evolve_context(
addr, context, ContextChangeReason.ARCHITECTURE_CHANGE, decision
)
# Create influence relationship
temporal_graph.add_influence_relationship(ucxl_address, addr, "architectural_dependency")
# Show decision path analysis
logger.info(f"\n🛤️ Decision Path Analysis:")
for addr in sample_addresses:
path = temporal_graph.find_decision_path(ucxl_address, addr)
if path:
logger.info(f" Path to {addr}:")
for i, (path_addr, node) in enumerate(path):
logger.info(f" {i+1}. {path_addr.split('/')[-1]} (v{node.version})")
# Show related decisions within 2 hops
related = temporal_graph.find_related_decisions(addr, max_hops=2)
logger.info(f" 📊 {len(related)} decisions within 2 hops of {addr.split('/')[-1]}")
# Save the temporal data
temporal_graph.save_temporal_data()
# Analyze decision patterns (not time patterns)
patterns = temporal_graph.analyze_decision_patterns()
logger.info(f"\n📊 Decision Network Analysis:")
logger.info(f" Total decision nodes: {patterns['total_decisions']}")
logger.info(f" Decision makers: {patterns['decision_makers']}")
logger.info(f" Change reasons: {patterns['change_reasons']}")
logger.info(f" Impact scopes: {patterns['impact_scopes']}")
# Show network connectivity
total_addresses = len(temporal_graph.temporal_nodes)
total_influences = sum(len(node.influences) for versions in temporal_graph.temporal_nodes.values()
for node in versions)
logger.info(f" 🕸️ Network connectivity: {total_influences} decision influences across {total_addresses} addresses")
if __name__ == "__main__":
logging.basicConfig(level=logging.INFO, format='%(levelname)s - %(message)s')
demo_temporal_context()