# Hub-Macro: Override Call Pattern

This document describes the `override_call` feature in hub-macro, which allows activations to bypass automatic stream wrapping for methods that return `PlexusStream` directly.

## Problem

Most activation methods follow the caller-wraps pattern:

```rust
async fn echo(&self, msg: String) -> impl Stream<Item = EchoEvent> {
    stream! { yield EchoEvent::Echo { message: msg }; }
}
```

The macro wraps this with `wrap_stream()` to produce `PlexusStream`.

But some methods need to return `PlexusStream` directly—for example, **routing methods** that delegate to other activations:

```rust
// Plexus.call routes to other activations
// The routed stream is ALREADY a PlexusStream
async fn call(&self, method: String, params: Value) -> Result<PlexusStream, PlexusError> {
    self.route(&method, params).await  // Returns PlexusStream from another activation
}
```

If the macro wrapped this, it would double-wrap and corrupt the stream.

## Solution: Automatic Detection + Explicit Override

The macro uses a two-part solution:

### 1. Automatic Conflict Detection

If a method returns `Result<PlexusStream, _>` without the `override_call` attribute, the macro emits a helpful error:

```
error: Method `call` returns Result<PlexusStream, _> which conflicts with
       generated wrapping. Add #[hub_macro::hub_method(override_call)] if you
       want to handle dispatch manually, or change return type to
       `impl Stream<Item = T>` for automatic wrapping.
```

### 2. Explicit Override Attribute

Mark methods that return `PlexusStream` directly:

```rust
#[hub_macro::hub_methods(namespace = "plexus", ...)]
impl Plexus {
    // Normal methods - macro wraps with wrap_stream()
    async fn hash(&self) -> impl Stream<Item = HashEvent> { ... }
    async fn list_activations(&self) -> impl Stream<Item = ListActivationsEvent> { ... }

    // Override method - returns PlexusStream directly, no wrapping
    #[hub_macro::hub_method(override_call)]
    async fn call(&self, method: String, params: Value) -> Result<PlexusStream, PlexusError> {
        self.route(&method, params).await
    }
}
```

## Generated Code

### Normal Method Dispatch

```rust
"hash" => {
    let stream = self.hash().await;
    Ok(wrap_stream(stream, "plexus.hash", vec!["plexus".into()]))
}
```

### Override Method Dispatch

```rust
"call" => {
    let method = /* extract from params */;
    let params = /* extract from params */;
    self.call(method, params).await  // Returns Result<PlexusStream, _> directly
}
```

### RPC Implementation

For override methods, the RPC handler forwards the `PlexusStream` items directly:

```rust
async fn call(&self, pending: PendingSubscriptionSink, method: String, params: Value) -> SubscriptionResult {
    let sink = pending.accept().await?;
    let stream_result = Plexus::call(self, method, params).await;

    tokio::spawn(async move {
        match stream_result {
            Ok(mut stream) => {
                // Forward PlexusStream items directly - no additional wrapping
                while let Some(item) = stream.next().await {
                    if let Ok(raw) = serde_json::value::to_raw_value(&item) {
                        if sink.send(raw).await.is_err() { break; }
                    }
                }
            }
            Err(e) => {
                // Convert error to PlexusStreamItem::Error
                let error = PlexusStreamItem::Error { ... };
                let _ = sink.send(to_raw_value(&error)).await;
            }
        }
        // Append Done event
        let done = PlexusStreamItem::Done { ... };
        let _ = sink.send(to_raw_value(&done)).await;
    });
    Ok(())
}
```

## Implementation Details

### Parsing (`parse.rs`)

```rust
pub struct HubMethodAttrs {
    pub name: Option<String>,
    pub param_docs: HashMap<String, String>,
    pub override_call: bool,  // New field
}
```

Detected via `#[hub_method(override_call)]` attribute.

### Conflict Detection (`parse.rs`)

```rust
fn is_result_plexus_stream(ty: &Type) -> bool {
    // Returns true if type is Result<PlexusStream, _>
}

// In MethodInfo::from_fn():
if is_result_plexus_stream(&return_type) && !is_override {
    return Err(syn::Error::new_spanned(...));
}
```

### Code Generation (`activation.rs`)

```rust
fn generate_dispatch_arms(...) {
    if m.is_override {
        // Call directly, return as-is
        quote! { self.#fn_name(#params).await }
    } else {
        // Wrap with wrap_stream
        quote! { Ok(wrap_stream(self.#fn_name(#params).await, ...)) }
    }
}
```

## Use Cases

### 1. Routing/Proxy Methods

Methods that delegate to other activations:

```rust
#[hub_method(override_call)]
async fn call(&self, method: String, params: Value) -> Result<PlexusStream, PlexusError> {
    self.route(&method, params).await
}
```

### 2. Handle Resolution

Methods that resolve handles by delegating:

```rust
#[hub_method(override_call)]
async fn resolve(&self, handle: Handle) -> Result<PlexusStream, PlexusError> {
    self.resolve_handle(&handle).await
}
```

### 3. Aggregation Methods

Methods that combine streams from multiple sources:

```rust
#[hub_method(override_call)]
async fn multi_call(&self, methods: Vec<String>) -> Result<PlexusStream, PlexusError> {
    // Merge multiple PlexusStreams into one
    let streams = futures::future::join_all(
        methods.iter().map(|m| self.route(m, Value::Null))
    ).await;
    Ok(merge_streams(streams))
}
```

## Design Principles

1. **Explicit over implicit**: Override must be explicitly marked
2. **Helpful errors**: Conflict detection provides clear guidance
3. **Type-based detection**: Uses return type to detect potential conflicts
4. **Consistent streaming**: All methods still produce `PlexusStream` in the end

## Status

- [x] Attribute parsing implemented
- [x] Conflict detection with error message
- [x] RPC generation for override methods
- [x] Dispatch generation for override methods
- [ ] Apply to Plexus activation (in progress)
