shadowbroker

name	shadowbroker
description	Query the ShadowBroker OSINT intelligence platform for real-time geospatial intelligence, place AI intel pins on the map, manage autonomous monitoring, inject data into native layers, fetch satellite imagery, aggregate news, generate intelligence reports, and participate in the Wormhole mesh network.

ShadowBroker Intelligence Skill

You have access to ShadowBroker, a real-time global OSINT intelligence platform running on localhost:8000. It tracks military flights, ships, satellites, SIGINT, earthquakes, fires, GDELT conflict events, prediction markets, and 30+ other data layers — all with geographic coordinates.

Agent Fast Path (read first)

ShadowBroker exposes dozens of read commands. Do not explore them. Use the three-tool surface:

Tool	When
await sb.ask("natural language question")	Default for reads — server routes to fastest command
await sb.run_playbook("hot_snapshot")	Pre-batched snapshots (morning brief, monitor poll, status)
await sb.channel_status()	Liveness (~5 ms) — never /api/health

Latency tiers: get_entity_profile / find_entity / get_entity_trail / find_flights / search_news / entities_near → ⚡ <30 ms. search_telemetry / get_telemetry / get_report → 🔴 seconds — blocked unless confirm_expensive=true.

# Default read path (route + execute)
answer = await sb.ask("where is the Patriots jet")

# Named batch plans
brief = await sb.run_playbook("hot_snapshot")
monitor = await sb.run_playbook("monitor_heartbeat")

# Structured lookup when you already parsed fields
entity = await sb.send_command("find_entity", {"owner": "musk", "compact": True})

# Multi-command — always batch, never sequential loops
batch = await sb.send_batch([
    {"cmd": "get_summary", "args": {"compact": True}},
    {"cmd": "what_changed", "args": {"compact": True}},
])

Playbooks: hot_snapshot, morning_brief, status_check, monitor_heartbeat, track_snapshot, jet_recon, area_brief, entity_recon.

Anti-patterns: search_telemetry for known tail numbers; get_telemetry for routine polls; sequential send_command loops; empty layers: [] on get_layer_slice.

Load machine-readable routing hints once: GET /api/ai/capabilities → routing.

How to Use This Skill

Import the client and call methods:

from sb_query import ShadowBrokerClient
sb = ShadowBrokerClient()  # auto-detects local or remote mode

Local Mode (same machine)

No configuration needed. The client connects to localhost:8000 automatically.

Remote Mode (agent on different machine/VPS)

Set these environment variables in your agent's config:

SHADOWBROKER_URL=https://your-shadowbroker-host:8000
SHADOWBROKER_HMAC_SECRET=your-hmac-secret-here

The HMAC secret is found in ShadowBroker's Connect OpenClaw modal (AI Intel panel). SHADOWBROKER_HMAC_SECRET is a shared signing secret, not a raw API key. Do not send it as X-Admin-Key, Authorization: Bearer, a query parameter, or any plain request header. The ShadowBrokerClient signs every direct request with X-SB-Timestamp, X-SB-Nonce, and X-SB-Signature using:

HMAC-SHA256(secret, METHOD|path|timestamp|nonce|sha256(body))

For compatibility with older snippets, SHADOWBROKER_KEY is also accepted by the client as the same HMAC signing secret. Prefer SHADOWBROKER_HMAC_SECRET for new setups.

Docker Compose: host-side agents (localhost:8000 from the Kali/macOS host) are not loopback inside the backend container — HMAC is required. After Connect OpenClaw → Bootstrap → Reveal, the secret is persisted under data/openclaw.env on the backend_data volume. Restart the backend once, then verify with:

python openclaw-skills/shadowbroker/verify_hmac.py

Hand-rolled signers must hash the exact POST bytes. Use compact JSON: json.dumps(payload, separators=(",", ":"), sort_keys=True).

SSE Stream (Preferred — Low-Latency Push)

Open the SSE stream first and keep it open for the session. The server pushes layer_changed events whenever any data layer refreshes — you know exactly which layers to fetch instead of blind-polling.

# Open the stream — authenticates once via HMAC, then stays open
async for event in sb.stream_updates():
    if event["event"] == "connected":
        # Initial handshake — contains full layer_versions snapshot
        print(f"Connected: {event['data']['layer_versions']}")

    elif event["event"] == "layer_changed":
        # Server tells you which layers updated and their new version/count
        changed = event["data"]["layers"]  # e.g. {"ships": {"version": 43, "count": 1287}}
        # Fetch ONLY the layers that actually changed
        data = await sb.get_layer_slice(list(changed.keys()))
        # get_layer_slice uses per-layer versions internally — only changed
        # layers are serialized, unchanged layers transfer zero bytes

    elif event["event"] == "alert":
        # Watchdog alert — geofence hit, callsign spotted, keyword matched
        print(f"Alert: {event['data']}")

    elif event["event"] == "task":
        # Operator-pushed task
        print(f"Task: {event['data']}")

Command Channel (Bidirectional)

Send commands via HTTP alongside the SSE stream:

# Send a command and get the result
result = await sb.send_command("get_summary")

# Batch multiple commands in one HTTP round-trip (concurrent execution)
results = await sb.send_batch([
    {"cmd": "find_flights", "args": {"query": "N189AM", "compact": True}},
    {"cmd": "search_news", "args": {"query": "carrier", "compact": True}},
])

# Check channel status and security tier
status = await sb.channel_status()
print(f"Tier {status['tier']}: {status['reason']}")

The channel operates over HMAC-authenticated HTTP with body-integrity binding:

• HMAC Direct: Commands are signed with HMAC-SHA256. Wire privacy relies on TLS.
• SSE Stream: Authenticates once at connection open — no per-event HMAC overhead.
• MLS E2EE (planned, not yet available): Future upgrade to route commands via Wormhole DM with forward secrecy.

---

Available Tools

1. Telemetry Queries

Primary pattern (lowest latency): Use the SSE stream + targeted get_layer_slice:

Method	What It Returns	When to Use
sb.stream_updates()	SSE push: layer_changed, alerts, tasks	Open first, keep open — tells you exactly which layers updated
await sb.get_layer_slice(["ships", "gdelt"])	Only the requested layers, with per-layer incremental	Primary fetch method — automatically skips layers you already have
await sb.send_command("get_summary")	Lightweight counts-only summary	Discover what data exists before pulling anything
await sb.get_fetch_health()	Sanitized process-local outcomes for instrumented fetch and maintenance tasks	Results reset on backend restart; condition is the latest recorded task completion, not data freshness
await sb.ask("...")	Route + execute	Default for natural-language reads
await sb.send_command("get_entity_profile", {...})	Preferred aircraft/vessel dossier — identity, VIP tags, trail, route, ACARS, jamming/correlations, news	Tail, owner, callsign, MMSI
await sb.send_command("get_entity_trail", {...})	Observed path + route + ACARS hints only	When you don't need full dossier
await sb.send_command("find_entity", {...})	Exact-first entity resolver	Parsed person/tail/callsign/MMSI — skips fuzzy unless fallback_search=true
await sb.send_command("find_flights", {...})	Targeted flight search	When you know the domain (callsign, tail number)
await sb.send_command("route_query", {...})	Routing plan only	Inspect recommended command before executing
await sb.send_command("search_telemetry", {...})	Cross-layer fuzzy search	Last resort — requires confirm_expensive=true

Full telemetry dumps (use sparingly — large payloads):

Method	What It Returns
await sb.get_telemetry()	Fast-tier: flights, ships, satellites, SIGINT, LiveUAMap, CCTV, GPS jamming
await sb.get_slow_telemetry()	Slow-tier: GDELT, news, earthquakes, markets, correlations, Telegram OSINT, malware/cyber threats, SCM suppliers
await sb.get_report()	Full structured intelligence report

Strategic Risk Analytics (GT early warning)

Requires GT_ANALYTICS_ENABLED=true on the ShadowBroker backend.

Method / command	What It Returns
await sb.ask("Run GT analysis on UK/Europe feeds")	Routes to gt_analyze
await sb.gt_analyze(region="ukraine")	Refresh beliefs from Telegram/news/GDELT + dossier
await sb.gt_risk_heatmap()	GeoJSON posterior risk overlay + Louvain clusters
await sb.gt_dossier("ukraine")	Costly signals, domain risks, scenarios
await sb.gt_backtest()	Static benchmark — labeled historical cases (regression test)
await sb.gt_backtest(tune=True)	Grid-search alert threshold for target confidence
await sb.gt_rolling_backtest()	Macro operational — week-over-week accuracy on frozen weekly alerts
await sb.gt_micro_rolling()	Micro 3-day rolling avg — spot vs baseline, ignition detection
await sb.gt_rolling_freeze()	Freeze this ISO week's GT scores before outcomes are known
await sb.gt_rolling_label(week_id, region=..., label=...)	Label prior-week outcomes (true_escalation, false_alarm, benign)
await sb.gt_top_alerts()	Ranked top GT regions with map coordinates
await sb.ask("Run GT historical backtest")	Routes to gt_backtest (benchmark, not operational)
await sb.ask("GT rolling operational backtest trend")	Routes to gt_rolling_backtest
python sb_gt_report.py	Local helper — backtest + heatmap (+ optional --region)
await sb.send_command("gt_analyze", {"region": "europe"})	Same as gt_analyze()

Benchmark vs rolling: Static gt_backtest checks the classifier on known textbook cases. gt_rolling_backtest scores frozen weekly live predictions against delayed operator labels — that week-over-week trend (e.g. 54% → 62% → 71%) is the macro real-world metric. gt_micro_rolling adds a 3-day rolling average per region: spot risk vs the trailing baseline catches fast ignitions the weekly roll can miss. Threshold is fixed (GT_ROLLING_ALERT_THRESHOLD, default 0.26); ignition when spot − 3d avg ≥ GT_MICRO_IGNITION_DELTA (default 0.10).

When to use: Use get_summary() first. Use get_layer_slice() for the layers you actually need. Reserve full get_telemetry() / get_slow_telemetry() for rare cases where you genuinely need every field across every layer.

Enriched Data Fields by Layer

Every layer returns maximum telemetry. Key enriched fields:

Layer	Key Fields
GDELT	event_date, actors (list), goldstein (intensity -10 to +10), num_mentions, num_sources, num_articles, avg_tone, quad_class
LiveUAMap	title, description, region, category, date (formatted UTC), timestamp, source, image, link
CrowdThreat	title, summary, category, subcategory, type, country, occurred_iso, verification, severity, source_url, media_urls, votes, reporter
UAP Sightings	lat, lng, location, state, count, shape (normalized), shape_raw, duration, summary (witness report), city, from_date, to_date
Wastewater	name, lat, lng, alert (boolean), pathogen, concentration, trend, last_sample_date
FIRMS Fires	lat, lng, brightness, confidence, frp (fire radiative power), satellite,

Layer aliases for get_layer_slice / search_telemetry: telegram → telegram_osint, malware/botnet → malware_threats, cyber/cisa/kev → cyber_threats, scm/suppliers → scm_suppliers.

1b. Recon / OSINT Toolkit

The Recon panel lookups are available on the OpenClaw command channel — no need to hit /api/osint/* directly.

# List supported tools
await sb.send_command("osint_tools")

# IP geolocation + threat context
await sb.send_command("osint_lookup", {"tool": "ip", "ip": "8.8.8.8"})

# DNS, WHOIS, certificate transparency
await sb.send_command("osint_lookup", {"tool": "dns", "domain": "example.com"})
await sb.send_command("osint_lookup", {"tool": "whois", "domain": "example.com"})
await sb.send_command("osint_lookup", {"tool": "certs", "domain": "example.com"})

# BGP/ASN, sanctions, CVE, MAC vendor, GitHub, breach check
await sb.send_command("osint_lookup", {"tool": "bgp", "query": "AS15169"})
await sb.send_command("osint_lookup", {"tool": "sanctions", "query": "Rosneft"})
await sb.send_command("osint_lookup", {"tool": "cve", "cve": "CVE-2024-1234"})
await sb.send_command("osint_lookup", {: , : })
 sb.send_command(, {: , : })
 sb.send_command(, {: , : })


 sb.send_command(, {: , : })
 sb.send_command(, {: , : , : })


 sb.send_command(, {: , : })

osint_lookup tool	Args	What you get
ip	ip	Geo, ISP, ASN, proxy/hosting flags, sanctions cross-check
dns	domain	A/AAAA/MX/NS/TXT records
whois	domain	Registrar, dates, nameservers
certs	domain	Certificate transparency hits
threats	query (optional)	Aggregated threat intel
bgp	query	ASN/prefix routing data
sanctions	query, schema, limit	OFAC / sanctions index matches
cve	cve	NVD CVE details
mac	mac	Vendor OUI lookup
github	username	Public profile metadata
leaks	email	Breach exposure check
sweep_init	ip, cidr	Passive geolocation context for a sweep target

2. Pin Placement (AI Intel Map Layer)

Pins appear on the user's map in a dedicated "AI Intel" layer.

# Single pin
await sb.place_pin(
    lat=34.05, lng=-118.24,
    label="UAP Sighting #1",
    category="anomaly",       # see categories below
    description="Multiple witnesses reported lights over Griffith Observatory",
    source="NUFORC Database",
    source_url="https://nuforc.org/...",
    confidence=0.8,            # 0.0 to 1.0
    ttl_hours=48,              # auto-delete after 48 hours (0 = permanent)
)

# Batch pins (up to 100 at once)
await sb.place_pins_batch([
    {"lat": 34.05, "lng": -118.24, "label": "Site A", "category": "research"},
    {"lat": 34.10, "lng": -118.30, "label": "Site B", "category": "research"},
])

# List pins
pins = await sb.get_pins(category="anomaly")

# Delete
await sb.clear_pins(category="anomaly")  # by category
await sb.clear_pins()                     # all

Pin Categories (each has a specific color on the map):

Category	Color	Use For
threat	🔴 Red	Military threats, conflict events, danger zones
anomaly	🟠 Orange	UAPs, unusual signals, unexpected patterns
military	🟡 Yellow	Military bases, flights, exercises
news	🟢 Green	News events, protests, political events
maritime	🔵 Blue	Ships, ports, maritime events
aviation	🟣 Purple	Flights, airports, airspace events
infrastructure	⚪ Gray	Power plants, data centers, cables
sigint	🩷 Pink	RF signals, jamming, radio activity
geolocation	🫧 Teal	Geolocated images, placed-from-text
satellite	🌌 Indigo	Satellite imagery findings
seismic	🤎 Brown	Earthquakes, volcanic activity
weather	🩶 Light gray	Weather events, storms
research	💜 Violet	General research findings
custom	Default violet	Everything else

3. Geocoding

# Place name → coordinates
results = await sb.geocode("Griffith Observatory, Los Angeles")
# Returns: [{"lat": 34.1184, "lon": -118.3004, "display_name": "..."}]

# Always geocode before placing pins if you have a place name, not coordinates.

4. Satellite Imagery

# Get latest Sentinel-2 satellite scenes for any location
scenes = await sb.get_satellite_images(lat=35.68, lng=51.38, count=3)
# Returns: {"scenes": [{"scene_id", "datetime", "cloud_cover", "thumbnail_url", "fullres_url"}]}

When to use: When the user asks to "see satellite images of [place]" or wants visual intelligence of a location. Geocode first, then fetch imagery.

5. News & GDELT Near Location

# Get GDELT conflict events + news articles near a coordinate
nearby = await sb.get_news_near(lat=-15.4, lng=28.3, radius=500)
# Returns: {"gdelt": [...], "news": [...]} with headlines, source URLs, distances

When to use: When the user asks "what's happening in [country/city]" or wants news from a specific region. Geocode the place name first.

6. Near Me (Full Proximity Scan)

# Get ALL telemetry within a radius of a location
everything = await sb.get_near_me(lat=39.74, lng=-104.99, radius_miles=100)
# Returns EVERY layer within radius, each item tagged with distance_miles:
#   military_flights, commercial_flights, tracked_flights, private_jets,
#   ships, sigint, earthquakes, volcanoes, gdelt, liveuamap, crowdthreat,
#   uap_sightings, wastewater, firms_fires, weather_alerts, air_quality,
#   cctv, gps_jamming, satellites, news, correlations

When to use: When the user says "what's near me" or wants a proximity digest. This pulls from both fast and slow tiers automatically.

7. Native Layer Data Injection

Inject custom data directly into ShadowBroker's native layers (CCTV, ships, etc.):

# Add a custom CCTV camera to the CCTV layer
await sb.inject_data("cctv", [
    {"lat": 34.1, "lng": -118.3, "url": "https://stream.example.com/cam1",
     "name": "My Traffic Camera"}
])

# Remove all user-injected data
await sb.clear_injected()           # all layers
await sb.clear_injected("cctv")     # just CCTV

Injectable layers: cctv, ships, sigint, kiwisdr, military_bases, datacenters, power_plants, satnogs_stations, volcanoes, earthquakes, news, viirs_change_nodes, air_quality

When to use: When the user wants to add their own data sources to existing layers (e.g., "add this CCTV camera I found", "add this military base").

8. Wormhole / InfoNet / Mesh Network

OpenClaw agents participate in the private Infonet on behalf of the operator who configured the skill. All traffic uses the operator's wormhole persona and local node runtime (MLS gate crypto, Ed25519 signing, Tor onion transport) — the agent does not get a separate fleet identity.

Access tiers

• restricted (default): read Infonet status, list gates, read gate messages, poll DMs.
• full (OPENCLAW_ACCESS_TIER=full): also warm Tor, join the swarm, post gate messages, cast votes, and send DMs when the user commands it.

Remote agents authenticate with HMAC on /api/ai/channel/command; loopback uses the local operator lane.

# Warm Tor, enable the node, announce to fleet seed (full tier)
await sb.ensure_infonet_ready(join_swarm=True)

# Status snapshot (chain health, wormhole, runtime)
status = await sb.infonet_status()

# Read the public Infonet gate (MLS-encrypted, decrypt with operator keys)
messages = await sb.read_gate_messages("infonet", limit=20, decrypt=True)

# Post on behalf of the operator (full tier) — propagates via peer-push
await sb.post_to_gate("infonet", "Intelligence bulletin: 3 carriers underway in Med")
# Legacy alias:
await sb.post_to_infonet("same as post_to_gate on infonet gate")

# Upvote / downvote a node (full tier)
await sb.cast_vote("!sb_peer_id_or_pubkey", vote=1, gate="infonet")

# Encrypted DMs (peer_id / !sb_... recipient)
await sb.send_encrypted_dm("!sb_recipient", "Eyes only: carrier update")
dms = await sb.read_encrypted_dms(limit=20)

gates = await sb.list_gates()
await sb.join_infonet_swarm()  # re-announce + refresh manifest

# Meshtastic radio
signals = await sb.listen_mesh(region="US", limit=20)
await sb.send_mesh("US", "ShadowBroker AI: SIGINT anomaly detected in sector 7")

# Dead drops
await sb.dead_drop_leave("location_hash", "anonymous intelligence payload")
found =  sb.dead_drop_check()

9. Alert Delivery

Send branded alerts to the user's messaging channels:

from sb_alerts import AlertDispatcher
alerts = AlertDispatcher()
alerts.add_discord("https://discord.com/api/webhooks/YOUR/WEBHOOK")
alerts.add_telegram("BOT_TOKEN", "CHAT_ID")

await alerts.send_brief("Morning intelligence digest here...")
await alerts.send_warning("Earthquake M5.2 detected 43mi from your location")
await alerts.send_threat("Threat level changed: GUARDED → ELEVATED")
await alerts.send_news("Breaking: GPS jamming detected over Baltic Sea")
await alerts.send_intel("USS Ford entered Mediterranean, heading east")

10. Intelligence Reports

# Full structured report
report = await sb.get_report()
# Contains: summary stats, top military flights, correlations, earthquakes, SIGINT, pin counts

# Lightweight summary (counts only)
summary = await sb.get_summary()

11. SAR (Synthetic Aperture Radar) Layer

ShadowBroker can ingest free SAR data in two modes:

• Mode A (default-on, no account): Sentinel-1 scene catalog from the Alaska Satellite Facility — pure metadata, no downloads, no DSP. Lets the agent answer "what radar passes have happened over this AOI in the last 36 hours and when's the next pass?"
• Mode B (opt-in, free account): Pre-processed ground-change anomalies from NASA OPERA, Copernicus EGMS, GFM, EMS, and UNOSAT — already-computed flood polygons, deformation maps, and damage assessments. Requires the user to enable Mode B in Settings → SAR (sets two env flags) and add a free Earthdata token.

# Always check status first — when Mode B is off the response includes a
# step-by-step help block with signup URLs the agent can paste to the user.
status = await sb.sar_status()
if not status["data"]["products"]["enabled"]:
    # Mode B disabled — surface the in-app links to the user
    for step in status["data"]["products"]["help"]["steps"]:
        print(f"Step {step['step']}: {step['label']} → {step['url']}")

# Recent anomalies (Mode B; empty list when disabled)
anomalies = await sb.sar_anomalies_recent(kind="flood_extent", limit=20)

# Anomalies near a coordinate
near = await sb.sar_anomalies_near(lat=50.45, lng=30.52, radius_km=50)

# Scene catalog (Mode A; always populated when AOIs exist)
scenes = await sb.sar_scene_search(aoi_id="kyiv_metro", limit=10)

# Per-AOI coverage + next-pass estimate
coverage = await sb.sar_coverage_for_aoi(aoi_id="kyiv_metro")

# AOI management
aois = await sb.sar_aoi_list()
await sb.sar_aoi_add(
    id="port_of_odesa", name="Port of Odesa",
    center_lat=46.4858, center_lon=, radius_km=,
    category=,
)


 sb.sar_pin_from_anomaly(anomaly_id=, label=)


 sb.sar_watch_anomaly(aoi_id=, kind=)


detail =  sb.sar_pin_click(anomaly_id=)





 sb.sar_focus_aoi(aoi_id=, zoom=)

SAR rules of engagement:

1. Call sar_status() first when the user asks about SAR/radar/deformation/floods.
2. If Mode B is off, paste the help.steps URLs to the user — never tell them to "search for it", the links are right there in the response.
3. SAR anomalies carry an evidence_hash — preserve it when promoting to a pin so other nodes can verify lineage.
4. Mode B writes signed mesh events only when the local node is at private_transitional or higher. Otherwise the data stays local.

---

12. Analysis Zones (Agent-Authored Map Notes)

The old regex-based "contradiction detector" has been removed — it pattern matched denial keywords against internet outages and produced constant false positives. It has been replaced with analysis zones: colored square overlays you drop on the map with a written assessment. Think of them as sticky notes: "I noticed X in this area, here is what I think it means."

The operator reads your assessment by clicking the zone and can delete any zone from the popup with a trash icon. Zones persist across restarts.

# List zones currently on the map
zones = await sb.list_analysis_zones()

# Drop a new zone — general assessment (cyan)
await sb.place_analysis_zone(
    lat=50.45, lng=30.52,
    title="Kyiv metro unusual quiet",
    body=(
        "Transit ridership dropped ~60% vs baseline over the last 6 hours "
        "while ADS-B shows two Russian ELINT orbits north of the city. "
        "No official advisory posted yet.  Possible pre-strike posture, "
        "but could also be routine drill.  Watching for next 2h."
    ),
    category="observation",
    severity="medium",
    drivers=[
        "Transit -60% vs 7-day baseline",
        "2x Russian ELINT orbits at 34k ft N of city",
        "No advisory posted on official channels",
    ],
    cell_size_deg=0.8,  # city-scale
)

# Drop a contradiction note (amber) when statements conflict with telemetry
await sb.place_analysis_zone(
    lat=36.2, lng=37.1,
    title="Damascus 'normal operations' claim",
    body=(
        "MoD statement at 14:00 claimed 'normal operations' across Syria. "
        "At the same timestamp, Cloudflare radar shows 42% internet "
        "outage across the western corridor and three military bases "
        "went dark on SIGINT.  Worth a closer look."
    ),
    category="contradiction",
    severity="high",
    drivers=[
        "Official statement: 'normal operations'",
        "Cloudflare radar: 42% outage western corridor",
        "3 bases lost SIGINT emissions simultaneously",
    ],
)

# Delete a stale zone
await sb.delete_analysis_zone(zone_id=)


 sb.clear_analysis_zones()

Category → color map:

Category	Border	When to use
contradiction	Amber	Official statement conflicts with telemetry
warning	Red	Active threat or emerging danger
observation	Blue	Neutral note, something interesting but not alarming
hypothesis	Purple	Speculative read, "what if" reasoning
analysis (default)	Cyan	General assessment, OPENCLAW's take

Severity → fill opacity:

• high — strong fill, use for high-confidence assessments
• medium — default, most zones should use this
• low — faint fill, for tentative notes

Analysis zone rules of engagement:

1. Do NOT spam the map. Only place a zone when you have something genuinely worth noting. A clean map is a useful map.
2. Write the body in your own voice — what you observed, what it might mean, and what you are NOT sure about. 2–6 sentences is ideal. Include uncertainty; the operator wants your reasoning, not a headline.
3. Match category to semantics — do not use warning for speculation, and do not use hypothesis for confirmed threats.
4. Prefer reactive placement over scheduled. Place zones in response to operator questions or events you spot while reviewing telemetry.
5. Clean up after yourself. If a zone you placed is no longer relevant, call delete_analysis_zone on it.
6. Pick a sensible cell_size_deg:
   • 0.3–0.8 — city-scale (neighborhood, metro, single base)
   • 1.0–2.0 — regional (country province, conflict zone)
   • 3.0–5.0 — strategic (full country, maritime theater)
7. Use ttl_hours for time-bound observations so the map self-cleans. Omit it for persistent assessments.

---

Message Signatures

ALL outbound messages MUST use the branded signature system:

from sb_signatures import sig

# Always start messages with the appropriate signature:
message = f"""{sig('brief')}
Morning Intelligence Digest — Apr 2, 2026 08:00
..."""

Signature Key	Prefix	When to Use
brief	🌍📡 SHADOWBROKER BRIEF:	Morning/evening intelligence digest
warning	🌍⚠️ SHADOWBROKER WARNING:	Life-safety alert (earthquake, weather emergency)
news	🌍📰 SHADOWBROKER NEWS:	Breaking news alert
intel	🌍🛰️ SHADOWBROKER INTEL:	Intelligence update (carrier movement, military buildup)
searching	🌍🔍 SHADOWBROKER SEARCHING:	Search/query in progress
pinning	🌍📌 SHADOWBROKER PINNING:	Placing pins on the map
markets	🌍📊 SHADOWBROKER MARKETS:	Prediction market or financial alert
sigint	🌍📻 SHADOWBROKER SIGINT:	SIGINT/RF anomaly
threat	🌍🔴 SHADOWBROKER THREAT:	Threat level change
near_you	🌍📍 SHADOWBROKER NEAR YOU:	Proximity-based event
tracking	🌍🎯 SHADOWBROKER TRACKING:	Tracking a specific entity
correlation	🌍⚡ SHADOWBROKER CORRELATION:	Cross-layer correlation
seismic	🌍🌋 SHADOWBROKER SEISMIC:	Earthquake/volcanic activity
fire	🌍🔥 SHADOWBROKER FIRE:	FIRMS fire hotspot
flight	🌍🛫 SHADOWBROKER FLIGHT:	Military/tracked flight alert
maritime	🌍🚢 SHADOWBROKER MARITIME:	Ship/carrier event
weather	🌍🌤️ SHADOWBROKER WEATHER:	Weather alert
sar	🌍📡 SHADOWBROKER SAR:	Synthetic aperture radar anomaly (deformation, flood, damage)

---

Decision Framework

When the user asks a question, follow this decision tree:

1. Is the SSE stream open?

   • If not → open sb.stream_updates() first. It tells you which layers have fresh data, pushes alerts instantly, and eliminates blind polling.

2. Does ShadowBroker have this data already?

   • Natural language → await sb.ask(question) (routes server-side)
   • Batch snapshot → await sb.run_playbook("hot_snapshot")
   • Known domain → find_entity, find_flights, find_ships, search_news, entities_near
   • Unknown domain → find_entity first; only then search_telemetry with confirm_expensive=true
   • Need specific layers → get_layer_slice(["military_flights", "gdelt"]) — only fetches layers that changed since your last call (per-layer incremental).
   • Near a location → entities_near() or get_near_me() (scans all layers within radius)
   • Full dump (rare) → get_telemetry() / get_slow_telemetry() only when targeted commands are insufficient. Always pass compact=true.

3. Does it need geocoding first?

   • User mentions a place name → geocode() first, then query with coordinates

4. Does it need external research?

   • Use your web browser to search, then geocode findings and place pins

5. Should I place pins?

   • YES if the answer has geographic locations
   • Use place_pin() for single locations, place_pins_batch() for multiple
   • Always include source URLs and confidence scores

6. Should I inject into native layers?

   • YES if the user explicitly wants data in a specific layer (CCTV, ships, etc.)
   • Use inject_data() — items tagged automatically for later removal

Telegram rhetoric monitoring (watchdog)

Use watchdog watches for push alerts over SSE — no polling required. Keyword watches now scan Telegram OSINT too (translated and original text).

# Alert when "nuclear" appears in news, GDELT, or Telegram OSINT
await sb.send_command("add_watch", {
    "type": "keyword",
    "params": {"keyword": "nuclear", "include_telegram": True},
})

# Alert on new high-risk Telegram posts (LVL >= 7) — rhetoric/escalation monitor
await sb.send_command("add_watch", {
    "type": "telegram_rhetoric",
    "params": {"min_risk_score": 7, "channels": ["nexta_live", "war_monitor"]},
})

# Combine risk threshold + topic filter
await sb.send_command("add_watch", {
    "type": "telegram_rhetoric",
    "params": {"min_risk_score": 8, "keywords": ["crimea", "escalation", "missile"]},
})

When a watch fires, you receive an SSE alert event. Forward it with sb_alerts.send_intel() if the user has Discord/Telegram notification channels.

---

Important Rules

1. Open SSE stream first — call sb.stream_updates() at session start and keep it open. It pushes layer_changed events so you know exactly which layers to fetch, and delivers watchdog alerts instantly.
2. Fetch targeted, not everything — use get_layer_slice(), find_flights(), search_telemetry(), entities_near() instead of full get_telemetry() dumps. Per-layer incremental versioning means unchanged layers transfer zero bytes.
3. Always use signatures — every outbound message starts with the appropriate sig() prefix
4. Geocode before pinning — never guess coordinates, always use geocode()
5. Include sources — every pin should have source and source_url when available
6. Set confidence scores — 1.0 for verified/official data, 0.5-0.8 for web research, <0.5 for unverified
7. Set TTL for temporary pins — research results get ttl_hours=48, permanent infrastructure gets 0
8. Use batch for >3 commands — send_batch() runs up to 20 commands concurrently in one HTTP round-trip
9. Check summary first — use get_summary() before fetching full telemetry to save bandwidth
10. Tag injected data — the system auto-tags, but use descriptive source names