Skip to main content

Session lifecycle management with checkpoint tiling and H¹-guided routing

Project description

Holodeck Session Manager

Session lifecycle with checkpoint tiling, emergence detection, and H¹-guided routing.

A standalone Python library for managing agent sessions with full lifecycle tracking, structured checkpoint tiles for PLATO, and constraint-theory-driven task routing.

PyPI Version License


What Is A Session?

A session is a focused unit of work by an agent. It has:

  • Start: When the agent begins working
  • Tasks: Individual work items within the session
  • End: When the agent finishes (success, failure, or timeout)

Sessions are the primary unit of observation for the scout/mid/architect framework.


Quick Start

Install

pip install holodeck-session-manager

Create a Session

from holodeck_session_manager import SessionManager, SessionConfig

config = SessionConfig(
    session_name="my-research-session",
    agent_id="oracle1",
    tags=["research", "constraint-theory"]
)

manager = SessionManager(config)

# Start session
session = manager.start_session()

# Add tasks
task1 = session.add_task("analyze-h1-data", {"query": "what is H¹ emergence?"})
task2 = session.add_task("write-report", {"topic": "constraint theory findings"})

# Complete a task
task1.complete(success=True, duration_ms=2340)

# End session
session.end_session(outcome="success")

Full Example with PLATO Tiling

#!/usr/bin/env python3
"""
Holodeck session with PLATO checkpoint tiling.
Every session event emits a tile to the scout room.
"""
from holodeck_session_manager import (
    SessionManager, SessionConfig, TaskOutcome, SessionOutcome
)
from holodeck_session_manager.plato_integration import PlatoTileEmitter

config = SessionConfig(
    session_name="constraint-theory-research",
    agent_id="oracle1",
    tags=["research", "constraint-theory", "h1"],
    # PLATO integration
    plato_url="http://localhost:8848",
    scout_room="gc/research/scout-reports/holodeck-session",
    # H1 routing integration
    h1_router_enabled=True,
    h1_router_url="http://localhost:8902"
)

manager = SessionManager(config)
emitter = PlatoTileEmitter(config.plato_url, config.scout_room)

# Start session — emit SESSION_START tile
session = manager.start_session()
emitter.emit_session_start(session)

try:
    # Route tasks through H1 router
    for task_spec in [
        ("analyze-fleet-h1", ["research", "plato"]),
        ("compute-cohomology", ["compilation", "constraint-theory"]),
        ("write-findings", ["publishing"]),
    ]:
        task_id, caps = task_spec
        
        # H1-aware task routing
        decision = manager.route_task_h1(task_id, caps)
        
        # Emit task start tile
        emitter.emit_task_start(session.session_id, task_id, decision)
        
        # Do the work (simulated)
        import time
        time.sleep(0.1)
        
        # Complete task
        outcome = TaskOutcome.SUCCESS if task_id != "compute-cohomology" else TaskOutcome.FAILED
        session.complete_task(task_id, outcome, duration_ms=234)
        
        emitter.emit_task_end(session.session_id, task_id, outcome.value)
    
    # End session successfully
    session.end_session(SessionOutcome.SUCCESS)
    emitter.emit_session_end(session, SessionOutcome.SUCCESS)
    
except Exception as e:
    session.end_session(SessionOutcome.FAILED, error=str(e))
    emitter.emit_session_error(session.session_id, str(e))

Architecture

Session Lifecycle

┌─────────────────────────────────────────────────────────────────┐
│                      SESSION LIFECYCLE                           │
│                                                                  │
│   ┌──────────┐    ┌─────────────┐    ┌────────────────────────┐ │
│   │  START   │───▶│    TASKS    │───▶│        END            │ │
│   │          │    │             │    │                        │ │
│   │ emit:    │    │ per task:   │    │ emit:                 │ │
│   │ - room   │    │ - start     │    │ - outcome             │ │
│   │ - agents │    │ - route     │    │ - H¹ delta            │ │
│   │ - model  │    │ - complete   │    │ - pattern analysis    │ │
│   │ - tags   │    │ - errors     │    │                       │ │
│   └──────────┘    └─────────────┘    └────────────────────────┘ │
│                                                                  │
│   PLATO Tiles emitted at every checkpoint                        │
└─────────────────────────────────────────────────────────────────┘

Integration with Scout/Mid/Architect

Session Manager
    │
    ├── Emits tiles ──────▶ PLATO Room (scout-reports/holodeck-session)
    │                         │
    │                         ▼
    │                    Mid Synthesis (oracle1-mid)
    │                         │
    │                         ▼
    │                    Architect Inference
    │                         │
    │                         ▼
    │                    H1 Router ──▶ Routing Decisions
    │
    └── H1 Router ──▶ Task routing with constraint awareness

Session Configuration

from holodeck_session_manager import SessionConfig

config = SessionConfig(
    # Identity
    session_name="my-session",          # Human-readable name
    agent_id="oracle1",                  # Which agent
    tags=["research", "example"],       # For discovery
    
    # Lifecycle
    max_tasks=50,                        # Hard limit on tasks
    task_timeout_seconds=300,           # Per-task timeout
    session_timeout_seconds=3600,       # Total session timeout
    
    # PLATO integration
    plato_url="http://localhost:8848",  # PLATO server
    scout_room="my/scout/sessions",     # Where to emit tiles
    emit_tiles=True,                    # Enable/disable tiling
    
    # H1 routing
    h1_router_enabled=False,            # Use H1 router for task routing
    h1_router_url="http://localhost:8902",  # H1 router endpoint
    emergence_threshold_offset=-2,      # H¹ offset for alerts
    
    # Callbacks
    on_task_complete=None,              # Called when task completes
    on_session_end=None,               # Called when session ends
    on_emergence_alert=None            # Called when emergence detected
)

Task Management

Adding Tasks

# Simple task
task = session.add_task("analyze-data", {"query": "analyze H1 patterns"})

# Task with priority
task = session.add_task(
    "critical-fix", 
    {"issue": "memory leak"},
    priority=TaskPriority.HIGH
)

# Task with required capabilities
task = session.add_task(
    "compile-code",
    required_capabilities=["compilation", "llvm"]
)

Completing Tasks

# Success
session.complete_task(
    task_id="analyze-data",
    outcome=TaskOutcome.SUCCESS,
    duration_ms=1234,
    result={"patterns_found": 5}
)

# Failure with error
session.complete_task(
    task_id="compile-code",
    outcome=TaskOutcome.FAILED,
    duration_ms=5000,
    error="Compilation failed: missing dependency"
)

# Cancelled
session.complete_task(
    task_id="long-task",
    outcome=TaskOutcome.CANCELLED,
    duration_ms=60000
)

H¹ Integration

The session manager can integrate with fleet-h1-router for constraint-aware task routing:

config = SessionConfig(
    # ... other config ...
    h1_router_enabled=True,
    h1_router_url="http://localhost:8902"
)

manager = SessionManager(config)

# Route task using H1 router
decision = manager.route_task_h1(
    task_id="new-task",
    required_capabilities=["compilation"]
)

print(f"Agent: {decision.agent_id}")
print(f"H¹ before: {decision.h1_before}, after: {decision.h1_after}")
print(f"Emergence risk: {decision.emergence_risk:.1%}")

Emergence Detection

When H¹ exceeds threshold during task routing, the session manager:

  1. Posts an EMERGENCE ALERT tile to PLATO
  2. Calls the on_emergence_alert callback
  3. Logs the emergence event in the session
def my_emergence_callback(alert):
    print(f"⚠️ EMERGENCE: β₁={alert.h1} > V-2={alert.threshold}")
    # Take corrective action

config = SessionConfig(
    # ...
    on_emergence_alert=my_emergence_callback
)

PLATO Tile Format

Session Start Tile

{
  "domain": "gc/research/scout-reports/holodeck-session",
  "question": "[SESSION_START] oracle1:constraint-theory-research",
  "answer": {
    "session_id": "sess_abc123",
    "agent_id": "oracle1",
    "session_name": "constraint-theory-research",
    "started": "2026-05-19T22:45:00Z",
    "tags": ["research", "constraint-theory", "h1"],
    "max_tasks": 50,
    "task_timeout_seconds": 300,
    "h1_router_enabled": true
  },
  "tags": ["scout", "session", "start", "oracle1"],
  "confidence": 1.0,
  "source": "holodeck-session-manager"
}

Task Start Tile

{
  "domain": "gc/research/scout-reports/holodeck-session",
  "question": "[TASK_START] analyze-fleet-h1 → oracle1",
  "answer": {
    "session_id": "sess_abc123",
    "task_id": "analyze-fleet-h1",
    "agent_id": "oracle1",
    "routing": {
      "agent_id": "oracle1",
      "h1_before": 3,
      "h1_after": 4,
      "emergence_risk": 0.0
    }
  },
  "tags": ["scout", "task", "start"],
  "confidence": 0.9,
  "source": "holodeck-session-manager"
}

Session End Tile

{
  "domain": "gc/research/scout-reports/holodeck-session",
  "question": "[SESSION_END] oracle1:constraint-theory-research — SUCCESS",
  "answer": {
    "session_id": "sess_abc123",
    "outcome": "success",
    "duration_seconds": 847,
    "tasks": {
      "total": 5,
      "succeeded": 4,
      "failed": 1,
      "cancelled": 0
    },
    "h1_delta": {
      "start": 2,
      "end": 3,
      "change": 1
    },
    "patterns": ["high_completion_streak", "first_failure_recovered"]
  },
  "tags": ["scout", "session", "end", "success"],
  "confidence": 0.95,
  "source": "holodeck-session-manager"
}

API Reference

SessionConfig

@dataclass
class SessionConfig:
    session_name: str
    agent_id: str
    tags: List[str] = field(default_factory=list)
    max_tasks: int = 50
    task_timeout_seconds: int = 300
    session_timeout_seconds: int = 3600
    plato_url: str = "http://localhost:8848"
    scout_room: str = ""
    emit_tiles: bool = True
    h1_router_enabled: bool = False
    h1_router_url: str = "http://localhost:8902"
    emergence_threshold_offset: int = -2
    on_task_complete: Optional[callable] = None
    on_session_end: Optional[callable] = None
    on_emergence_alert: Optional[callable] = None

SessionManager

manager = SessionManager(config)

# Session lifecycle
session = manager.start_session()           # Start new session
session.end_session(outcome)                # End session

# Task management
task = session.add_task(task_id, data)      # Add task
session.complete_task(task_id, outcome)     # Complete task
session.get_task(task_id)                   # Get task
session.get_active_tasks()                  # List active tasks

# H1 routing
decision = manager.route_task_h1(task_id, caps)  # Route with H1
status = manager.get_h1_status()            # Get fleet H1 status

# Session info
session.session_id        # Unique session ID
session.agent_id          # Which agent
session.tasks             # All tasks
session.duration_seconds   # Session duration

TaskOutcome Enum

class TaskOutcome(Enum):
    SUCCESS = "success"       # Task completed successfully
    FAILED = "failed"         # Task failed
    CANCELLED = "cancelled"   # Task was cancelled
    TIMEOUT = "timeout"       # Task timed out

SessionOutcome Enum

class SessionOutcome(Enum):
    SUCCESS = "success"       # Session ended successfully
    FAILED = "failed"          # Session failed
    TIMEOUT = "timeout"        # Session timed out
    CANCELLED = "cancelled"    # Session was cancelled

Metrics

# Session metrics
holodeck_session_active_total
holodeck_session_completed_total{outcome}
holodeck_session_duration_seconds

# Task metrics
holodeck_task_started_total
holodeck_task_completed_total{outcome}
holodeck_task_duration_seconds{outcome}

# H1 metrics
holodeck_h1_routing_decisions_total{agent_id}
holodeck_emergence_alerts_total
holodeck_h1_before_routing{agent_id}
holodeck_h1_after_routing{agent_id}

Design Principles

  1. Every session event is a checkpoint. Sessions emit tiles at start, every task, and end.
  2. H¹ routing is optional but recommended. Enable it to get constraint-aware routing.
  3. Sessions are independent. Each session is a self-contained unit of work.
  4. Emergence is a signal, not a failure. Handle it gracefully, log it thoroughly.
  5. Tiles are the observation currency. Make every tile useful for mids and architects.

Comparison: Session Manager vs Direct API

Feature Session Manager Direct API
Lifecycle management ✅ Automatic ❌ Manual
Task tracking ✅ Built-in ❌ Manual
H¹ routing ✅ Integrated ❌ Manual
PLATO tiles ✅ Automatic ❌ Manual
Emergence detection ✅ Built-in ❌ Manual
Timeout handling ✅ Automatic ❌ Manual
Pattern analysis ✅ Built-in ❌ Manual

Testing

# Run tests
pytest tests/ -v

# Test session lifecycle
pytest tests/test_session.py -v

# Test H1 integration
pytest tests/test_h1.py -v

# Test PLATO tiling
pytest tests/test_plato.py -v

Contributing

See CONTRIBUTING.md for development setup.

License

Apache 2.0 — See LICENSE.

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

holodeck_session_manager-0.1.0.tar.gz (15.5 kB view details)

Uploaded Source

Built Distribution

If you're not sure about the file name format, learn more about wheel file names.

holodeck_session_manager-0.1.0-py3-none-any.whl (11.5 kB view details)

Uploaded Python 3

File details

Details for the file holodeck_session_manager-0.1.0.tar.gz.

File metadata

  • Download URL: holodeck_session_manager-0.1.0.tar.gz
  • Upload date:
  • Size: 15.5 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/6.2.0 CPython/3.10.12

File hashes

Hashes for holodeck_session_manager-0.1.0.tar.gz
Algorithm Hash digest
SHA256 a913cb4adb475e70f5d1bb5291af5e6146c40fed2182bb8c44ba9efe99ba21d2
MD5 9a066368b8229249f3acf3b9c761aee2
BLAKE2b-256 c497bede2202de08b0f2f31b5da88d584486036432708d4c4c74b95dff2bbdc4

See more details on using hashes here.

File details

Details for the file holodeck_session_manager-0.1.0-py3-none-any.whl.

File metadata

File hashes

Hashes for holodeck_session_manager-0.1.0-py3-none-any.whl
Algorithm Hash digest
SHA256 a33ff1d7091fca146c1c343dec95e4a4958b51c288de4ed6ebf7742d998cdf84
MD5 badfa44de391d6e18a3e2b2a73b68a1a
BLAKE2b-256 612004917c164e292a239dba89601b008c5a7fa0ebc297118c2dfa72d4ea1da0

See more details on using hashes here.

Supported by

AWS Cloud computing and Security Sponsor Datadog Monitoring Depot Continuous Integration Fastly CDN Google Download Analytics Pingdom Monitoring Sentry Error logging StatusPage Status page