plato-scout 0.1.0

Fleet observation framework — scout processes that emit checkpoint tiles to PLATO

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.

---

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

pip install plato-scout

Write Your First Scout

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

# 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:

{
  "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:

#!/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:

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:

# 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

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

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

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

# 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

# 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 for development setup and coding standards.

License

Apache 2.0 — See LICENSE for details.