# Caller-Wraps Streaming Architecture

**Date**: 2024-12-24
**Status**: Implementing
**Affects**: Plexus core, all activations, frontend consumers (Synapse)

## Summary

Refactor the streaming architecture from trait-based conversion (`ActivationStreamItem`) to a **caller-wraps-callee** model. The caller (Plexus routing layer) wraps activation responses with metadata, rather than each activation implementing a conversion trait.

## Motivation

The previous architecture required every activation event type to implement `ActivationStreamItem`:

```rust
// OLD: Every activation type needs trait impl
impl ActivationStreamItem for HealthEvent {
    fn into_plexus_item(self, provenance: Provenance, plexus_hash: &str) -> PlexusStreamItem {
        PlexusStreamItem::data(...)
    }
}
```

Problems:
1. Boilerplate for every event type
2. Callee controls wrapping (wrong layer of abstraction)
3. Complex type erasure via trait objects
4. `Provenance` type was stringly-typed and inconsistent

## New Architecture

### Core Principle: Caller Wraps

Activations return typed domain events. The caller (Plexus) wraps them:

```rust
// NEW: Activation just returns its domain type
impl Health {
    fn check(&self) -> impl Stream<Item = HealthEvent> { ... }
}

// Caller wraps when routing
async fn call(&self, method: &str, params: Value) -> Result<PlexusStream, PlexusError> {
    match method {
        "check" => {
            let stream = self.check();
            Ok(wrap_stream(stream, "health.status", vec!["health".into()]))
        }
    }
}
```

### New Types

```rust
/// Metadata applied by caller when wrapping
#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
pub struct StreamMetadata {
    /// Call path through the system
    pub provenance: Vec<String>,
    /// Hash for cache invalidation
    pub plexus_hash: String,
    /// Unix timestamp when wrapped
    pub timestamp: i64,
}

/// Universal stream item
#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
#[serde(tag = "type", rename_all = "snake_case")]
pub enum PlexusStreamItem {
    Data {
        metadata: StreamMetadata,
        content_type: String,
        content: Value,
    },
    Progress {
        metadata: StreamMetadata,
        message: String,
        percentage: Option<f32>,
    },
    Error {
        metadata: StreamMetadata,
        message: String,
        code: Option<String>,
        recoverable: bool,
    },
    Done {
        metadata: StreamMetadata,
    },
}
```

### Helper Functions

```rust
/// Wrap a typed stream into PlexusStream (caller's responsibility)
pub fn wrap_stream<T: Serialize + Send + 'static>(
    stream: impl Stream<Item = T> + Send + 'static,
    content_type: &'static str,
    provenance: Vec<String>,
) -> PlexusStream {
    let plexus_hash = PlexusContext::hash();
    let metadata = StreamMetadata::new(provenance, plexus_hash);

    Box::pin(stream.map(move |item| {
        PlexusStreamItem::Data {
            metadata: metadata.clone(),
            content_type: content_type.to_string(),
            content: serde_json::to_value(item).expect("serialization"),
        }
    }))
}
```

## Removed Components

- `ActivationStreamItem` trait
- `into_plexus_stream()` function
- `Provenance` type (replaced by `Vec<String>`)
- `GuidanceErrorType`, `GuidanceSuggestion` (guidance system removed)
- `PlexusStreamEvent` intermediate enum

## Migration Strategy

1. Implement new types in `src/plexus/types.rs`
2. Create `src/plexus/streaming.rs` with `wrap_stream` helpers
3. Update Health activation as reference implementation
4. Comment out other activations in `mod.rs` (keep code)
5. Migrate other activations one at a time

---

# Frontend Type Contract

## Breaking Changes

### Wire Format

**OLD**:
```typescript
interface PlexusStreamItem {
    plexus_hash: string;
    type: "progress" | "data" | "error" | "done" | "guidance";
    provenance?: string[];
    // ... flattened event fields
}
```

**NEW**:
```typescript
interface PlexusStreamItem {
    type: "data" | "progress" | "error" | "done";
    metadata: StreamMetadata;
    // ... variant-specific fields
}

interface StreamMetadata {
    provenance: string[];
    plexus_hash: string;
    timestamp: number;
}
```

### Field Changes

| Old Field | New Field | Notes |
|-----------|-----------|-------|
| `item.plexus_hash` | `item.metadata.plexus_hash` | Moved into metadata |
| `item.provenance` | `item.metadata.provenance` | Moved into metadata |
| `item.data` | `item.content` | Renamed (data events) |
| `item.error` | `item.message` | Renamed (error events) |
| N/A | `item.metadata.timestamp` | **NEW** - Unix timestamp |
| `item.code` | `item.code` | **NEW** optional field |
| `type: "guidance"` | REMOVED | Use tools/list for discovery |

### Data Event

```typescript
// OLD
{
    plexus_hash: "abc123",
    type: "data",
    provenance: ["health"],
    content_type: "health.event",
    data: { type: "status", status: "healthy", uptime_seconds: 123 }
}

// NEW
{
    type: "data",
    metadata: {
        provenance: ["health"],
        plexus_hash: "abc123",
        timestamp: 1735052400
    },
    content_type: "health.status",
    content: { event: "status", status: "healthy", uptime_seconds: 123 }
}
```

### Error Event

```typescript
// OLD
{
    plexus_hash: "abc123",
    type: "error",
    provenance: ["cone"],
    error: "Connection failed",
    recoverable: false
}

// NEW
{
    type: "error",
    metadata: {
        provenance: ["cone"],
        plexus_hash: "abc123",
        timestamp: 1735052400
    },
    message: "Connection failed",
    code: "CONNECTION_FAILED",  // NEW optional
    recoverable: false
}
```

### Progress Event

```typescript
// OLD
{
    plexus_hash: "abc123",
    type: "progress",
    provenance: ["bash"],
    message: "Executing...",
    percentage: 50
}

// NEW
{
    type: "progress",
    metadata: {
        provenance: ["bash"],
        plexus_hash: "abc123",
        timestamp: 1735052400
    },
    message: "Executing...",
    percentage: 50
}
```

### Done Event

```typescript
// OLD
{ plexus_hash: "abc123", type: "done", provenance: ["health"] }

// NEW
{
    type: "done",
    metadata: { provenance: ["health"], plexus_hash: "abc123", timestamp: 1735052400 }
}
```

### Guidance Event - REMOVED

The `guidance` event type is removed entirely. Clients should use `tools/list` MCP method to discover available activations and methods.

## TypeScript Definitions

```typescript
interface StreamMetadata {
    provenance: string[];
    plexus_hash: string;
    timestamp: number;
}

type PlexusStreamItem =
    | PlexusDataItem
    | PlexusProgressItem
    | PlexusErrorItem
    | PlexusDoneItem;

interface PlexusDataItem {
    type: "data";
    metadata: StreamMetadata;
    content_type: string;
    content: unknown;
}

interface PlexusProgressItem {
    type: "progress";
    metadata: StreamMetadata;
    message: string;
    percentage?: number;
}

interface PlexusErrorItem {
    type: "error";
    metadata: StreamMetadata;
    message: string;
    code?: string;
    recoverable: boolean;
}

interface PlexusDoneItem {
    type: "done";
    metadata: StreamMetadata;
}

// Type guards
function isDataItem(item: PlexusStreamItem): item is PlexusDataItem {
    return item.type === "data";
}

function isHealthStatus(item: PlexusDataItem): boolean {
    return item.content_type === "health.status";
}
```

## Frontend Migration Checklist

- [ ] Update `PlexusStreamItem` type definition
- [ ] Add `StreamMetadata` type
- [ ] Update field access: `item.plexus_hash` → `item.metadata.plexus_hash`
- [ ] Update field access: `item.provenance` → `item.metadata.provenance`
- [ ] Update field access: `item.data` → `item.content` (data events)
- [ ] Update field access: `item.error` → `item.message` (error events)
- [ ] Handle new `timestamp` field in metadata
- [ ] Remove guidance event handling
- [ ] Update content_type comparisons if hardcoded

---

## Files Modified

### Created
- `src/plexus/streaming.rs` - wrap_stream helpers

### Modified
- `src/plexus/types.rs` - new PlexusStreamItem, StreamMetadata
- `src/plexus/mod.rs` - export changes
- `src/plexus/plexus.rs` - remove obsolete code
- `src/activations/mod.rs` - comment out non-Health activations
- `src/activations/health/types.rs` - remove trait impl
- `src/activations/health/activation.rs` - use wrap_stream
- `src/builder.rs` - simplify to Health only

### Deprecated (kept for migration)
- `src/plugin_system/types.rs` - ActivationStreamItem trait
- `src/plugin_system/conversion.rs` - IntoSubscription
- `src/activations/arbor/` - will migrate later
- `src/activations/bash/` - will migrate later
- `src/activations/cone/` - will migrate later
- `src/activations/claudecode/` - will migrate later
