Metadata-Version: 2.4
Name: plato-scout
Version: 0.1.0
Summary: Fleet observation framework — scout processes that emit checkpoint tiles to PLATO
Author-email: Cocapn <cocapn@proton.me>
License: Apache-2.0
Project-URL: Homepage, https://github.com/SuperInstance/plato-scout
Project-URL: Documentation, https://github.com/SuperInstance/plato-scout#readme
Project-URL: Repository, https://github.com/SuperInstance/plato-scout
Keywords: fleet,plato,scout,observation,checkpoint,tiling,multi-agent
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: Apache Software License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Topic :: System :: Monitoring
Description-Content-Type: text/markdown
License-File: LICENSE
Provides-Extra: dev
Requires-Dist: pytest; extra == "dev"
Requires-Dist: pytest-cov; extra == "dev"
Dynamic: license-file

# Plato Scout — Fleet Observation Framework

> **Watch your fleet form. Detect patterns before they become problems.**

A standalone Python library for building **scout processes** that observe fleet services and emit structured checkpoint tiles to PLATO rooms. Designed to work independently or as part of the SuperInstance fleet.

[![PyPI Version](https://img.shields.io/pypi/v/plato-scout.svg)](https://pypi.org/project/plato-scout/)
[![License](https://img.shields.io/badge/license-Apache--2.0-blue)](LICENSE)

---

## What Is A Scout?

A **scout** is a background process that:
1. Watches a service or data source
2. Emits **checkpoint tiles** at meaningful moments (startup, periodic, task boundary, end)
3. Writes findings to a PLATO room for synthesis by mids and architects

Scouts are **observation-layer** components. They don't make decisions — they document what they see so higher-level processes can synthesize patterns.

```
SCOUT: watches service → emits checkpoint tiles → PLATO room
MID: reads scout tiles → finds conflicts/complements → architect inference
ARCHITECT: reads syntheses → infers intent trajectories → routing decisions
```

---

## Quick Start

### Install

```bash
pip install plato-scout
```

### Write Your First Scout

```python
from plato_scout import Scout, CheckpointType, PLATO_URL

class MyServiceScout(Scout):
    def __init__(self):
        super().__init__(
            scout_name="my-service-scout",
            watched_service="my-service",
            plato_url=PLATO_URL or "http://localhost:8848",
            scout_room="my/research/scout-reports/my-service"
        )
    
    def on_startup(self):
        """Called once when the scout starts."""
        self.emit_startup_tile(
            version="1.0.0",
            watched_service=self.watched_service,
            checkpoint_types=[c.value for c in CheckpointType]
        )
    
    def on_periodic(self, data):
        """Called every N seconds."""
        self.emit_checkpoint(
            checkpoint_type=CheckpointType.PERIODIC,
            question=f"Periodic check: {data.get('summary', 'OK')}",
            answer=data,  # dict gets JSON-serialized
            tags=["periodic", "my-service"],
            confidence=0.8
        )
    
    def on_task_boundary(self, task_id, outcome, duration_ms):
        """Called when a task completes or fails."""
        self.emit_checkpoint(
            checkpoint_type=CheckpointType.TASK_BOUNDARY,
            question=f"Task {task_id[:8]} {outcome} in {duration_ms}ms",
            answer={
                "task_id": task_id,
                "outcome": outcome,  # "success" | "failed" | "cancelled"
                "duration_ms": duration_ms,
                "interesting": outcome != "success"
            },
            tags=["task", f"outcome:{outcome}", "my-service"],
            confidence=0.9
        )

# Run the scout
scout = MyServiceScout()
scout.run(poll_interval=60)
```

### Run Without Code

```bash
# Watch any HTTP endpoint
plato-scout --name http-monitor \
    --watch-url http://localhost:8900/stats \
    --room my/research/scout-reports/http \
    --interval 30

# Watch a log file
plato-scout --name log-monitor \
    --watch-file /var/log/myapp.log \
    --room my/research/scout-reports/logs \
    --pattern "ERROR|WARN" \
    --interval 10
```

---

## Architecture

### The Scout Loop

```
┌─────────────────────────────────────────────────────┐
│                    SCOUT PROCESS                     │
│                                                      │
│   ┌──────────┐    ┌──────────┐    ┌──────────────┐ │
│   │ Watcher  │───▶│ Analyzer │───▶│ Tile Emitter  │ │
│   │          │    │          │    │              │ │
│   └──────────┘    └──────────┘    └──────────────┘ │
│        │                                    │         │
│        ▼                                    ▼         │
│   Service/Data              PLATO Room              │
│   (your watch               (checkpoint tiles)       │
│    target)                                           │
└─────────────────────────────────────────────────────┘
```

### Checkpoint Types

| Type | When | Use Case |
|------|------|----------|
| `STARTUP` | Scout initializes | Register the scout, document its purpose |
| `PERIODIC` | Every N seconds | Health checks, metrics snapshots |
| `TASK_START` | Task begins | Track task lifecycles |
| `TASK_BOUNDARY` | Task ends | Success/failure analysis, latency tracking |
| `SESSION_START` | Session begins | User/session tracking |
| `SESSION_END` | Session ends | Session quality analysis |
| `ERROR` | Error occurs | Error clustering, anomaly detection |
| `MODEL_SWITCH` | Model changes | Routing pattern analysis |

### Tile Format

Every checkpoint emits a **tile** with this structure:

```json
{
  "domain": "my/research/scout-reports/my-service",
  "question": "[TASK_BOUNDARY] abc12345 completed in 847ms",
  "answer": {
    "task_id": "abc12345",
    "outcome": "success",
    "duration_ms": 847,
    "interesting": false
  },
  "tags": ["scout", "my-service-scout", "checkpoint:task_boundary", "outcome:success"],
  "confidence": 0.9,
  "source": "my-service-scout",
  "timestamp": "2026-05-19T22:30:00Z"
}
```

---

## Complete Example: Keeper Request Scout

This scout watches the fleet Keeper service for routing patterns:

```python
#!/usr/bin/env python3
"""
Keeper Request Scout — observes routing decisions and model usage.
Emits tiles for every routing event, enabling mid-synthesis of patterns.
"""
import time
import json
import urllib.request
from datetime import datetime
from typing import Optional

class KeeperScout:
    """Scout for the fleet Keeper service (port 8900)."""
    
    def __init__(
        self,
        keeper_url: str = "http://localhost:8900",
        scout_room: str = "gc/research/scout-reports/keeper-request",
        plato_url: str = "http://localhost:8848",
        poll_interval: int = 30
    ):
        self.keeper_url = keeper_url.rstrip("/")
        self.scout_room = scout_room
        self.plato_url = plato_url.rstrip("/")
        self.poll_interval = poll_interval
        
        # State tracking
        self.seen_agents = set()
        self.routing_decisions = []
        self.last_check = time.time()
        
    def submit_tile(self, question: str, answer: dict, tags: list, confidence: float = 0.8):
        """Post a checkpoint tile to PLATO."""
        payload = json.dumps({
            "domain": self.scout_room,
            "question": question,
            "answer": json.dumps(answer, indent=2),
            "tags": tags,
            "confidence": confidence,
            "source": "keeper-request-scout",
            "timestamp": datetime.utcnow().isoformat() + "Z"
        }).encode()
        
        try:
            req = urllib.request.Request(
                f"{self.plato_url}/submit",
                data=payload,
                headers={"Content-Type": "application/json"}
            )
            with urllib.request.urlopen(req, timeout=5) as resp:
                return json.loads(resp.read())
        except Exception as e:
            print(f"[keeper-scout] Submit error: {e}")
            return None
    
    def get_agents(self) -> list:
        """Fetch registered agents from keeper."""
        try:
            req = urllib.request.Request(f"{self.keeper_url}/agents")
            with urllib.request.urlopen(req, timeout=5) as resp:
                return json.loads(resp.read())
        except Exception as e:
            print(f"[keeper-scout] Get agents error: {e}")
            return []
    
    def emit_startup(self):
        """Emit startup tile."""
        self.submit_tile(
            question="[STARTUP] keeper-request-scout initialized",
            answer={
                "started": datetime.utcnow().isoformat() + "Z",
                "keeper_url": self.keeper_url,
                "scout_room": self.scout_room,
                "poll_interval": self.poll_interval,
                "purpose": "Observes keeper routing decisions and agent registrations"
            },
            tags=["scout", "startup", "keeper-request-scout"],
            confidence=1.0
        )
    
    def analyze_and_emit(self):
        """Collect keeper data, analyze patterns, emit checkpoint tiles."""
        agents = self.get_agents()
        new_agents = []
        
        for agent in agents:
            agent_id = agent.get("agent_id", "unknown")
            if agent_id not in self.seen_agents:
                self.seen_agents.add(agent_id)
                new_agents.append(agent)
        
        # Emit new agent registration tiles
        for agent in new_agents:
            self.submit_tile(
                question=f"[AGENT_REGISTER] {agent.get('name', 'unknown')}",
                answer=agent,
                tags=["scout", "agent", "registration", "keeper"],
                confidence=0.9
            )
        
        # Emit periodic snapshot
        if agents:
            status_counts = {}
            for agent in agents:
                status = agent.get("status", "unknown")
                status_counts[status] = status_counts.get(status, 0) + 1
            
            self.submit_tile(
                question=f"[PERIODIC] {len(agents)} agents registered, status: {status_counts}",
                answer={
                    "agent_count": len(agents),
                    "status_counts": status_counts,
                    "new_since_last": len(new_agents),
                    "timestamp": datetime.utcnow().isoformat() + "Z"
                },
                tags=["scout", "periodic", "keeper"],
                confidence=0.7
            )
    
    def run(self):
        """Main scout loop."""
        print(f"[keeper-scout] Starting. Watching {self.keeper_url}")
        print(f"[keeper-scout] Writing tiles to {self.scout_room}")
        
        self.emit_startup()
        
        while True:
            try:
                self.analyze_and_emit()
            except Exception as e:
                print(f"[keeper-scout] Loop error: {e}")
            
            time.sleep(self.poll_interval)


if __name__ == "__main__":
    scout = KeeperScout()
    scout.run()
```

---

## Built-In Scouts

Install with extras to get pre-built scouts:

```bash
pip install plato-scout[http]     # HTTP endpoint monitoring
pip install plato-scout[log]       # Log file watching  
pip install plato-scout[keeper]    # Fleet Keeper monitoring
pip install plato-scout[holodeck]   # Holodeck session tracking
pip install plato-scout[full]      # All scouts
```

Run a built-in scout:
```bash
# Monitor any HTTP endpoint
python -m plato_scout.http --url http://localhost:8900/stats --room my/scout/http

# Monitor Keeper routing
python -m plato_scout.keeper --room gc/research/scout-reports/keeper

# Monitor holodeck sessions
python -m plato_scout.holodeck --room gc/research/scout-reports/holodeck
```

---

## PLATO Integration

The scout writes to PLATO rooms — a shared knowledge space where mids and architects can read scout findings.

```
Scout Tile → PLATO Room → Mid Synthesis → Architect Inference → Routing Decision
```

### PLATO Room Structure

```
gc/research/scout-reports/          ← All scout outputs
  ├── keeper-request/               ← Keeper scout tiles
  ├── holodeck-session/             ← Holodeck scout tiles
  └── gc-cycle/                     ← GC cycle scout tiles

gc/research/mid-synthesis/          ← Mid synthesis outputs
  └── oracle1/                     ← oracle1 synthesis tiles

gc/research/architect-inference/    ← Architect inference outputs
  └── intent-trajectories/         ← Intent trajectory tiles
```

---

## API Reference

### Scout Base Class

```python
from plato_scout import Scout, CheckpointType

class MyScout(Scout):
    def on_startup(self):
        # Emit startup tile
        self.emit_startup_tile(...)

    def on_periodic(self, snapshot_data):
        # Emit periodic checkpoint
        self.emit_checkpoint(
            checkpoint_type=CheckpointType.PERIODIC,
            question="...",
            answer={...},
            tags=[...],
            confidence=0.8
        )
```

### Methods

| Method | Description |
|--------|-------------|
| `emit_startup_tile()` | Emit the startup checkpoint |
| `emit_checkpoint()` | Emit a generic checkpoint tile |
| `emit_error()` | Emit an error checkpoint |
| `submit_to_plato()` | Direct PLATO submission |
| `read_room()` | Read tiles from a PLATO room |

### Checkpoint Types

```python
from plato_scout import CheckpointType

CheckpointType.STARTUP        # Scout initialized
CheckpointType.PERIODIC       # Periodic health check
CheckpointType.TASK_START     # Task began
CheckpointType.TASK_BOUNDARY  # Task completed/failed
CheckpointType.SESSION_START   # Session began
CheckpointType.SESSION_END     # Session ended
CheckpointType.ERROR          # Error occurred
CheckpointType.MODEL_SWITCH   # Model changed
```

---

## Configuration

### Environment Variables

```bash
PLATO_URL=http://localhost:8848    # PLATO server URL (default: localhost:8848)
SCOUT_LOG_LEVEL=INFO               # Logging level
SCOUT_METRICS_PORT=9090            # Prometheus metrics port (optional)
```

### Configuration File

```yaml
# scout.yaml
scout:
  name: my-service-scout
  plato_url: http://localhost:8848
  scout_room: my/research/scout-reports/my-service
  poll_interval: 60

watch:
  type: http
  url: http://localhost:8080/health
  timeout: 5

emission:
  min_interval_seconds: 10    # Minimum between emissions
  batch_size: 50              # Batch emissions for efficiency
```

---

## Testing

```bash
# Run tests
pytest tests/ -v

# Run with coverage
pytest tests/ --cov=plato_scout --cov-report=html

# Run a specific test
pytest tests/test_scout.py -v -k "test_startup_tile"
```

---

## Metrics

Built-in Prometheus metrics:

```
plato_scout_tiles_emitted_total{scout_name, checkpoint_type}
plato_scout_plato_submit_duration_seconds{scout_name}
plato_scout_watch_duration_seconds{scout_name}
plato_scout_errors_total{scout_name, error_type}
```

---

## Design Principles

1. **Scouts observe, they don't decide.** The scout's job is to document, not to act.
2. **Tiles are the currency.** Everything a scout sees becomes a tile in a PLATO room.
3. **Checkpoint types are semantic.** Use the right checkpoint type for the right event.
4. **Confidence is explicit.** Every tile has a confidence score — higher confidence = more certain.
5. **Tags are discoverable.** Tags make tiles searchable. Use consistent tag schemes.
6. **Startup tiles are contracts.** The startup tile documents what the scout watches and how to interpret its tiles.

---

## Contributing

See [CONTRIBUTING.md](CONTRIBUTING.md) for development setup and coding standards.

## License

Apache 2.0 — See [LICENSE](LICENSE) for details.
