# Arbor Stream Map: Persistent Event Streams with Handle Resolution

> Design for replacing in-memory chat buffers with Arbor-backed persistent streams

## Problem Statement

Currently, ClaudeCode chat events are stored in an in-memory buffer:

```rust
chat_buffers: RwLock<HashMap<ClaudeCodeId, ActiveChatBuffer>>

struct ActiveChatBuffer {
    info: ChatBufferInfo,
    events: Vec<BufferedEvent>,  // Lost on restart
}
```

**Problems:**
1. Events are lost on Plexus restart
2. Single consumer (one read_position)
3. No integration with Arbor despite ClaudeCode already using Arbor for conversation history
4. Can't resolve event handles through Plexus

## Proposed Solution: Stream Map

Replace the in-memory buffer with an **Arbor-backed stream** where each event is a node in a tree.

### Architecture

```
┌─────────────────────────────────────────────────────────────────┐
│                        Stream Map                                │
│                                                                  │
│  stream_map: HashMap<ClaudeCodeId, StreamHead>                  │
│                                                                  │
│  ┌──────────────────────────────────────────────────────────┐   │
│  │  Session: 17ee2aba-...                                    │   │
│  │                                                           │   │
│  │  Tree: c23a8db2-...                                       │   │
│  │       │                                                   │   │
│  │       └── User Msg Node (existing)                        │   │
│  │               │                                           │   │
│  │               └── Stream Root ─┬─ Start Event             │   │
│  │                                ├─ Content Event           │   │
│  │                                ├─ Content Event           │   │
│  │                                ├─ ToolUse Event           │   │
│  │                                ├─ AwaitingApproval Event  │   │
│  │                                ├─ ToolResult Event        │   │
│  │                                └─ Complete Event ◄── HEAD │   │
│  └──────────────────────────────────────────────────────────┘   │
│                                                                  │
│  Polling: Walk from cursor to HEAD, resolve handles via Plexus  │
└─────────────────────────────────────────────────────────────────┘
```

### Key Concepts

**Stream Head**: Points to the latest event node in the stream. Updated atomically as events are pushed.

**Stream Root**: The first event node, child of the user message node. Marks the start of this chat's event stream.

**Event Nodes**: Each `ChatEvent` becomes an external node with a handle:
```
Handle {
    plugin_id: <claudecode_plugin_id>,
    version: "1.0.0",
    method: "chat_event",
    meta: [event_id, event_type]
}
```

**Cursor**: Each consumer maintains their own cursor (node_id) for independent traversal.

## Handle Resolution Through Hub Hierarchy

### Handle Origin Tracking

Every handle contains its origin via the `plugin_id` field:

```rust
// From hub-core/src/types.rs
pub struct Handle {
    pub plugin_id: Uuid,      // Stable plugin instance identifier - THE ORIGIN
    pub version: String,      // Semantic version for schema lookup
    pub method: String,       // Creation method (e.g., "chat_event")
    pub meta: Vec<String>,    // Metadata parts
}

impl Handle {
    pub fn origin(&self) -> Origin {
        Origin { plugin_id: self.plugin_id, method: self.method.clone() }
    }
}
```

The `plugin_id` uniquely identifies which plugin created the handle and is responsible for resolving it.

### Type Hierarchy

Handle is a **hub-core level construct**, not substrate-specific:

```
hub-core/src/types.rs        ← Handle struct defined here (with plugin_id origin)
hub-core/src/plexus/         ← Plexus, Activation trait, PluginRegistry
    ↓
substrate/src/types.rs       ← re-exports: pub use hub_core::types::*
substrate/src/plexus.rs      ← re-exports hub_core::plexus
    ↓
activations/arbor/           ← stores nodes with handles
activations/claudecode/      ← creates handles for events
activations/cone/            ← resolves its own handles
```

### Recursive Resolution Pattern

Handle resolution is a **plugin-level method** (like `call` and `schema`). The key insight:

> **All hubs should be able to resolve handles within their nested structure.**

This means:
1. **Every hub** implements `resolve_handle` as a standard method
2. **Resolution is recursive**: a hub routes to its children based on `handle.plugin_id`
3. **Plexus** (being a hub) implements this same pattern at the top level

```
┌─────────────────────────────────────────────────────────────────┐
│                    Recursive Resolution Flow                     │
│                                                                  │
│  Handle { plugin_id: "abc-123", method: "chat_event", ... }     │
│                              │                                   │
│                              ▼                                   │
│  ┌──────────────────────────────────────────────────────┐       │
│  │ Plexus.resolve_handle(handle)                        │       │
│  │   → Is "abc-123" one of my direct children?          │       │
│  │   → If yes: route to that child's resolve_handle     │       │
│  │   → If no: check if any child hub owns it (recurse)  │       │
│  └──────────────────────────────────────────────────────┘       │
│                              │                                   │
│                              ▼                                   │
│  ┌──────────────────────────────────────────────────────┐       │
│  │ ClaudeCode.resolve_handle(handle)                    │       │
│  │   → "abc-123" is my plugin_id!                       │       │
│  │   → Look up event by meta[0] (event_id)              │       │
│  │   → Return resolved event data                       │       │
│  └──────────────────────────────────────────────────────┘       │
└─────────────────────────────────────────────────────────────────┘
```

### Current State

From the codebase research:

| Component | Location | Status |
|-----------|----------|--------|
| Handle type with `plugin_id` origin | `hub-core/src/types.rs` | ✅ Complete |
| Plugin ID system & PluginRegistry | `hub-core/src/plexus/` | ✅ Complete |
| Activation trait with `resolve_handle` | `hub-core/src/plexus/` | ✅ Complete |
| Arbor storage (text + external nodes) | `substrate/activations/arbor/` | ✅ Complete |
| Individual plugin resolution (Cone) | `substrate/activations/cone/` | ✅ Complete |
| **Plexus-level recursive `resolve_handle()`** | `hub-core/src/plexus/` | ❌ Missing |

### Required: Hub.resolve_handle() (Recursive Pattern)

Every hub implements the same recursive resolution pattern:

```rust
// This is the standard pattern for any hub (including Plexus)
impl Hub {
    /// Resolve a handle by routing to the owning plugin
    pub async fn resolve_handle(&self, handle: &Handle)
        -> Result<PlexusStream, PlexusError>
    {
        let origin_id = handle.plugin_id;

        // 1. Am I the owner?
        if origin_id == self.plugin_id() {
            return self.resolve_own_handle(handle).await;
        }

        // 2. Is it one of my direct children?
        if let Some(child) = self.registry.lookup_by_id(origin_id) {
            let activation = self.get_activation(&child)?;
            return activation.resolve_handle(handle).await;
        }

        // 3. Recurse into child hubs (they might own it or have children that do)
        for child_hub in self.child_hubs() {
            if let Ok(result) = child_hub.resolve_handle(handle).await {
                return Ok(result);
            }
        }

        Err(PlexusError::HandleNotFound(format!(
            "No plugin found for handle origin: {}", origin_id
        )))
    }
}
```

For Plexus specifically (the root hub):

```rust
impl Plexus {
    /// Resolve a handle through the hub hierarchy
    pub async fn resolve_handle(&self, handle: &Handle)
        -> Result<PlexusStream, PlexusError>
    {
        // Route to owning plugin based on handle.plugin_id
        let plugin_path = self.registry.lookup_by_id(handle.plugin_id)
            .ok_or(PlexusError::ActivationNotFound(
                format!("Plugin not found for handle: {}", handle.plugin_id)
            ))?;

        let activation = self.get_activation(&plugin_path)?;
        activation.resolve_handle(handle).await
    }
}
```

### Required: ClaudeCode.resolve_handle()

ClaudeCode needs to implement handle resolution for its event handles:

```rust
#[hub_macro::hub_method]
async fn resolve_handle(&self, handle: Handle) -> impl Stream<Item = ResolveResult> {
    stream! {
        // Parse meta: [event_id, event_type]
        let event_id = handle.meta.get(0)
            .ok_or("Missing event_id in handle")?;

        // Look up event in storage
        match self.storage.get_event(event_id).await {
            Ok(event) => yield ResolveResult::Event(event),
            Err(e) => yield ResolveResult::Err { message: e.to_string() },
        }
    }
}
```

## Implementation Design

### Storage Changes

```rust
// New: Stream metadata stored in SQLite
CREATE TABLE chat_streams (
    session_id TEXT PRIMARY KEY,
    tree_id TEXT NOT NULL,
    stream_root_node TEXT,      -- First event node
    stream_head_node TEXT,      -- Latest event node
    status TEXT NOT NULL,       -- running, complete, failed
    started_at INTEGER NOT NULL,
    ended_at INTEGER,
    error TEXT,
    FOREIGN KEY (session_id) REFERENCES claudecode_sessions(id)
);

// New: Event data stored for handle resolution
CREATE TABLE chat_events (
    id TEXT PRIMARY KEY,
    session_id TEXT NOT NULL,
    seq INTEGER NOT NULL,
    event_type TEXT NOT NULL,
    event_data TEXT NOT NULL,   -- JSON
    created_at INTEGER NOT NULL,
    FOREIGN KEY (session_id) REFERENCES claudecode_sessions(id)
);
CREATE INDEX idx_chat_events_session ON chat_events(session_id, seq);
```

### Pushing Events

```rust
impl ClaudeCodeStorage {
    pub async fn stream_push_event(
        &self,
        session_id: &ClaudeCodeId,
        event: ChatEvent,
    ) -> Result<NodeId, ClaudeCodeError> {
        // 1. Get current stream head
        let stream = self.get_stream(session_id).await?;
        let parent_node = stream.head_node.unwrap_or(stream.root_node);

        // 2. Store event data for resolution
        let event_id = Uuid::new_v4();
        let seq = self.next_event_seq(session_id).await?;
        self.store_event(session_id, &event_id, seq, &event).await?;

        // 3. Create handle
        let handle = Handle::new(self.plugin_id(), "1.0.0", "chat_event")
            .with_meta(vec![event_id.to_string(), event.event_type()]);

        // 4. Create Arbor node
        let node_id = self.arbor.node_create_external(
            &stream.tree_id,
            Some(parent_node),
            handle,
            None,  // metadata
        ).await?;

        // 5. Update stream head
        self.update_stream_head(session_id, node_id).await?;

        Ok(node_id)
    }
}
```

### Polling via Arbor

```rust
impl ClaudeCodeStorage {
    pub async fn stream_poll(
        &self,
        session_id: &ClaudeCodeId,
        cursor: Option<NodeId>,  // Consumer's last-read position
        limit: Option<usize>,
    ) -> Result<(StreamInfo, Vec<ResolvedEvent>, Option<NodeId>), ClaudeCodeError> {
        let stream = self.get_stream(session_id).await?;

        // Start from cursor or stream root
        let start_node = cursor.unwrap_or(stream.root_node);

        // Walk the tree from start to head
        let path = self.arbor.node_get_path_between(
            &stream.tree_id,
            &start_node,
            &stream.head_node,
        ).await?;

        // Resolve each node's handle through Plexus
        let mut events = Vec::new();
        let plexus = self.plexus.upgrade()
            .ok_or("Plexus not available")?;

        for node in path.iter().take(limit.unwrap_or(100)) {
            if let NodeType::External { handle } = &node.data {
                // Resolve through Plexus (routes to ClaudeCode.resolve_handle)
                let resolved = plexus.resolve_handle(handle).await?;
                events.push(resolved);
            }
        }

        // Return new cursor (last node we read)
        let new_cursor = path.last().map(|n| n.id);

        Ok((stream.info(), events, new_cursor))
    }
}
```

## Consumer Cursors

Each consumer maintains their own cursor for independent traversal:

```rust
// Consumer A reads events 0-5
let (_, events_a, cursor_a) = storage.stream_poll(session_id, None, Some(5)).await?;
// cursor_a points to event 5's node

// Consumer B reads all events independently
let (_, events_b, cursor_b) = storage.stream_poll(session_id, None, None).await?;
// cursor_b points to head

// Consumer A continues from their cursor
let (_, more_events_a, cursor_a) = storage.stream_poll(session_id, cursor_a, None).await?;
// Gets events 6 onwards
```

## Benefits

1. **Persistence**: Events survive Plexus restarts
2. **Multiple Consumers**: Each has independent cursor
3. **Unified Model**: Same Arbor tree for conversation history and event streams
4. **Handle Resolution**: Any plugin's events can be resolved through Plexus
5. **Replay**: Can replay entire stream by walking from root
6. **Branching**: Could support speculative execution branches

## Integration with Loopback

The loopback approval events would also be stored as nodes:

```rust
enum ChatEvent {
    // ... existing
    AwaitingApproval {
        approval_id: Uuid,
        tool_name: String,
        tool_use_id: String,
        input: Value,
    },
    ApprovalResolved {
        approval_id: Uuid,
        approved: bool,
        message: Option<String>,
    },
}
```

Parent polling sees the full picture:
```
poll() → [..., ToolUse, AwaitingApproval, ...]
respond(approval_id)
poll() → [..., ApprovalResolved, ToolResult, ...]
```

## Migration Path

### Phase 1: Add Plexus.resolve_handle()
- Implement in `hub-core/src/plexus/`
- Route through PluginRegistry
- Add tests for cross-plugin resolution

### Phase 2: Add ClaudeCode Handle Resolution
- Implement `resolve_handle` method
- Store events in SQLite for resolution
- Create handles with proper meta

### Phase 3: Replace Buffer with Stream Map
- Add `chat_streams` and `chat_events` tables
- Modify `buffer_push_event` to create Arbor nodes
- Modify `buffer_poll` to walk Arbor and resolve handles
- Remove in-memory HashMap

### Phase 4: Consumer Cursors
- Track cursors per consumer
- Support multiple independent readers
- Add cursor management API

## Related Documents

- [Handle Architecture](16679869168467720703_handle-architecture.md) - Algebraic model of handles
- [Handle Rendering System](16679775901269181695_handle-rendering-system.md) - Mustache templates for display
- [Arbor Handles](16680556555233258495_arbor-handles.md) - Handle structure and resolution
- [ClaudeCode Loopback Integration](16677965632570341631_claudecode-loopback-integration.md) - Current polling-based flow
- [Generic Hub Context](16678638480642409727_generic-hub-context.md) - Parent context injection for Plexus access
