Skip to content

Commit 0198834

Browse files
SuperInstanceSuperInstance
committed
[holodeck] Fleet integration v2 with graceful degradation
1 parent 5a346b1 commit 0198834

1 file changed

Lines changed: 353 additions & 0 deletions

File tree

‎fleet_integration_v2.py‎

Lines changed: 353 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,353 @@
1+
Below is a **complete, ready‑to‑drop‑in** `fleet_integration.py` that satisfies every requirement you listed.
2+
The file contains:
3+
4+
* A small reusable helper (`_http_request`) that performs GET/POST with **retry / exponential back‑off**, a configurable timeout, and graceful‑degradation (returns `None` or an empty container when the fleet‑agent API cannot be reached).
5+
* One public function for each of the six operations you asked for, each wrapped in a try/except block that logs the problem and falls back to a sensible default value.
6+
* Simple “bottle” handling that writes JSON messages to a directory, marks them as read, and leaves a placeholder for the Git‑broadcast step (you can fill this in with your own repo‑specific commands).
7+
* Inline documentation and type hints to make the module easy to understand and to use from the rest of Holodeck Studio.
8+
9+
---
10+
11+
## **`fleet_integration.py`**
12+
13+
```python
14+
#!/usr/bin/env python3
15+
"""
16+
fleet_integration.py
17+
18+
Utility module that connects a Holodeck spatial‑agent environment to the
19+
fleet‑management backend.
20+
21+
Features
22+
--------
23+
* Register an agent with the fleet dashboard.
24+
* Periodically report health / status.
25+
* Pull a live map of all active agents and the rooms they occupy.
26+
* Send and receive “bottles” – lightweight JSON messages stored in a
27+
shared directory and (optionally) broadcast via Git.
28+
* Query the lighthouse service for RED_ALERT notifications.
29+
* All HTTP calls use urllib, JSON for payloads, and include retry /
30+
exponential‑back‑off logic so the functions degrade gracefully when
31+
the fleet‑agent API is unavailable.
32+
33+
Author: <your‑name>
34+
"""
35+
36+
from __future__ import annotations
37+
38+
import json
39+
import os
40+
import time
41+
import urllib.error
42+
import urllib.request
43+
from pathlib import Path
44+
from typing import Any, Dict, List, Optional
45+
46+
# ----------------------------------------------------------------------
47+
# Configuration – adjust these values for your deployment
48+
# ----------------------------------------------------------------------
49+
FLEET_API_HOST = "fleet-agent-api"
50+
FLEET_API_PORT = 8901
51+
LIGHTHOUSE_HOST = "lighthouse"
52+
LIGHTHOUSE_PORT = 8901
53+
54+
# Directory that holds the bottle files (must be shared / mounted for all agents)
55+
BOTTLE_DIR = Path("/var/holodeck/bottles") # <-- change to your real path
56+
57+
# HTTP settings
58+
HTTP_TIMEOUT = 5.0 # seconds
59+
MAX_RETRIES = 3
60+
BASE_BACKOFF = 0.5 # seconds (exponential back‑off factor)
61+
62+
63+
# ----------------------------------------------------------------------
64+
# Helper – low‑level HTTP request with retry / back‑off
65+
# ----------------------------------------------------------------------
66+
def _http_request(
67+
method: str,
68+
url: str,
69+
data: Optional[bytes] = None,
70+
headers: Optional[Dict[str, str]] = None,
71+
) -> Optional[bytes]:
72+
"""
73+
Perform a GET or POST request with retries.
74+
75+
Parameters
76+
----------
77+
method: "GET" or "POST"
78+
url: full URL (including scheme)
79+
data: raw bytes to send for POST (already JSON‑encoded)
80+
headers: optional dict of request headers
81+
82+
Returns
83+
-------
84+
The response body as ``bytes`` on success, or ``None`` if all retries fail.
85+
"""
86+
if headers is None:
87+
headers = {}
88+
89+
for attempt in range(1, MAX_RETRIES + 1):
90+
try:
91+
req = urllib.request.Request(url, data=data, headers=headers, method=method)
92+
with urllib.request.urlopen(req, timeout=HTTP_TIMEOUT) as resp:
93+
return resp.read()
94+
except (urllib.error.URLError, urllib.error.HTTPError) as exc:
95+
# Log the problem – in a real system you would use a logger
96+
print(f"[fleet_integration] HTTP {method} error on {url} (attempt {attempt}/{MAX_RETRIES}): {exc}")
97+
98+
# If we have more attempts left, wait a bit before retrying
99+
if attempt < MAX_RETRIES:
100+
backoff = BASE_BACKOFF * (2 ** (attempt - 1))
101+
time.sleep(backoff)
102+
else:
103+
# All retries exhausted – give up and return None
104+
return None
105+
return None # unreachable, but keeps type‑checkers happy
106+
107+
108+
# ----------------------------------------------------------------------
109+
# Public API
110+
# ----------------------------------------------------------------------
111+
def register_agent(agent_id: str, capabilities: List[str]) -> bool:
112+
"""
113+
Register an agent with the fleet dashboard.
114+
115+
POST ``/register`` with JSON payload:
116+
{"agent_id": "...", "capabilities": [...]}
117+
118+
Returns ``True`` if the server responded with HTTP 200, otherwise ``False``.
119+
"""
120+
url = f"http://{FLEET_API_HOST}:{FLEET_API_PORT}/register"
121+
payload = json.dumps({"agent_id": agent_id, "capabilities": capabilities}).encode()
122+
resp = _http_request("POST", url, data=payload, headers={"Content-Type": "application/json"})
123+
return resp is not None
124+
125+
126+
def report_status(agent_id: str, status_dict: Dict[str, Any]) -> bool:
127+
"""
128+
Send a health / status update for an agent.
129+
130+
POST ``/status`` with JSON payload:
131+
{"agent_id": "...", "status": {...}}
132+
133+
Returns ``True`` on HTTP 200, ``False`` otherwise.
134+
"""
135+
url = f"http://{FLEET_API_HOST}:{FLEET_API_PORT}/status"
136+
payload = json.dumps({"agent_id": agent_id, "status": status_dict}).encode()
137+
resp = _http_request("POST", url, data=payload, headers={"Content-Type": "application/json"})
138+
return resp is not None
139+
140+
141+
def get_fleet_map() -> Dict[str, Any]:
142+
"""
143+
Retrieve a snapshot of the current fleet.
144+
145+
GET ``/fleet`` – expected to return a JSON object mapping agent IDs to
146+
their current room / metadata.
147+
148+
Returns the parsed JSON dict on success, or an empty dict if the request
149+
fails (graceful degradation).
150+
"""
151+
url = f"http://{FLEET_API_HOST}:{FLEET_API_PORT}/fleet"
152+
resp = _http_request("GET", url)
153+
if resp is None:
154+
return {}
155+
try:
156+
return json.loads(resp.decode())
157+
except json.JSONDecodeError:
158+
print("[fleet_integration] Invalid JSON received from /fleet")
159+
return {}
160+
161+
162+
# ----------------------------------------------------------------------
163+
# Bottle handling (local file + optional Git broadcast)
164+
# ----------------------------------------------------------------------
165+
def _ensure_bottle_dir() -> None:
166+
"""Make sure the bottle directory exists."""
167+
BOTTLE_DIR.mkdir(parents=True, exist_ok=True)
168+
169+
170+
def send_bottle(from_agent: str, message: str, priority: str) -> bool:
171+
"""
172+
Write a bottle (JSON message) to the shared bottle directory and
173+
optionally broadcast it to the fleet via Git.
174+
175+
The file name convention is ``{timestamp}_{from}_{priority}.json``.
176+
The function returns ``True`` if the file was written successfully;
177+
Git broadcast failures are logged but do **not** cause the function to
178+
return ``False`` – the bottle is still persisted locally.
179+
"""
180+
_ensure_bottle_dir()
181+
182+
timestamp = int(time.time() * 1000)
183+
filename = f"{timestamp}_{from_agent}_{priority}.json"
184+
bottle_path = BOTTLE_DIR / filename
185+
186+
payload = {"from_agent": from_agent, "message": message, "priority": priority, "ts": timestamp}
187+
try:
188+
bottle_path.write_text(json.dumps(payload, ensure_ascii=False))
189+
except OSError as exc:
190+
print(f"[fleet_integration] Failed to write bottle file {bottle_path}: {exc}")
191+
return False
192+
193+
# ------------------------------------------------------------------
194+
# OPTIONAL: broadcast via Git.
195+
# ------------------------------------------------------------------
196+
# The concrete implementation depends on your repo layout and
197+
# authentication method. Below is a *very* simple placeholder that
198+
# runs ``git add/commit/push`` in the bottle directory. Replace it
199+
# with whatever workflow you need (e.g., using subprocess, GitPython,
200+
# CI pipelines, etc.).
201+
# ------------------------------------------------------------------
202+
try:
203+
import subprocess
204+
205+
subprocess.run(["git", "add", str(bottle_path)], cwd=str(BOTTLE_DIR), check=True)
206+
subprocess.run(
207+
["git", "commit", "-m", f"Bottle from {from_agent} ({priority})"],
208+
cwd=str(BOTTLE_DIR),
209+
stdout=subprocess.DEVNULL,
210+
stderr=subprocess.DEVNULL,
211+
check=False, # commit may be empty if another process already committed
212+
)
213+
subprocess.run(
214+
["git", "push"],
215+
cwd=str(BOTTLE_DIR),
216+
stdout=subprocess.DEVNULL,
217+
stderr=subprocess.DEVNULL,
218+
check=False,
219+
)
220+
except Exception as exc: # pragma: no cover – optional feature
221+
print(f"[fleet_integration] Git broadcast failed (non‑critical): {exc}")
222+
223+
return True
224+
225+
226+
def receive_bottles(agent_id: str) -> List[Dict[str, Any]]:
227+
"""
228+
Scan the bottle directory for unread messages addressed to ``agent_id``.
229+
A file is considered addressed to an agent if its name contains the
230+
``agent_id`` string (case‑sensitive). After reading, the file is renamed
231+
with a ``.read`` suffix so it will not be returned again.
232+
233+
Returns a list of parsed JSON objects (empty list if none or on error).
234+
"""
235+
_ensure_bottle_dir()
236+
unread: List[Dict[str, Any]] = []
237+
238+
for entry in BOTTLE_DIR.iterdir():
239+
if not entry.is_file():
240+
continue
241+
if entry.suffix == ".read":
242+
continue # already processed
243+
if agent_id not in entry.name:
244+
continue
245+
246+
try:
247+
content = entry.read_text()
248+
unread.append(json.loads(content))
249+
except (OSError, json.JSONDecodeError) as exc:
250+
print(f"[fleet_integration] Failed to read/parse bottle {entry}: {exc}")
251+
continue
252+
253+
# Mark as read – rename with a .read suffix
254+
try:
255+
entry.rename(entry.with_suffix(entry.suffix + ".read"))
256+
except OSError as exc:
257+
print(f"[fleet_integration] Could not rename bottle {entry} as read: {exc}")
258+
259+
return unread
260+
261+
262+
# ----------------------------------------------------------------------
263+
# Lighthouse RED_ALERT query
264+
# ----------------------------------------------------------------------
265+
def check_red_alert() -> List[Dict[str, Any]]:
266+
"""
267+
Query the lighthouse service for active RED_ALERT messages.
268+
269+
GET ``http://lighthouse:8901/alerts`` – the endpoint is expected to
270+
return a JSON list of alert objects.
271+
272+
Returns the list on success, or an empty list if the request fails.
273+
"""
274+
url = f"http://{LIGHTHOUSE_HOST}:{LIGHTHOUSE_PORT}/alerts"
275+
resp = _http_request("GET", url)
276+
if resp is None:
277+
return []
278+
try:
279+
alerts = json.loads(resp.decode())
280+
if isinstance(alerts, list):
281+
return alerts
282+
# If the service returns a dict with a key like "alerts", be tolerant:
283+
if isinstance(alerts, dict) and "alerts" in alerts:
284+
return alerts["alerts"]
285+
return []
286+
except json.JSONDecodeError:
287+
print("[fleet_integration] Invalid JSON received from lighthouse alerts")
288+
return []
289+
290+
291+
# ----------------------------------------------------------------------
292+
# Simple demo / sanity‑check when run as a script
293+
# ----------------------------------------------------------------------
294+
if __name__ == "__main__":
295+
# NOTE: This block is only for quick manual testing; in production
296+
# you will import the functions from other parts of Holodeck.
297+
test_agent = "demo_agent"
298+
caps = ["move", "sense"]
299+
300+
print("Registering agent …", register_agent(test_agent, caps))
301+
print("Reporting status …", report_status(test_agent, {"health": "good", "room": "lab"}))
302+
print("Current fleet map:", get_fleet_map())
303+
304+
print("Sending a bottle …", send_bottle(test_agent, "Hello from demo!", "high"))
305+
print("Receiving bottles for demo_agent …", receive_bottles(test_agent))
306+
307+
print("Checking RED_ALERTs …", check_red_alert())
308+
```
309+
310+
---
311+
312+
### How the implementation meets every requirement
313+
314+
| Requirement | Implementation details |
315+
|-------------|------------------------|
316+
| **1. `register_agent`** | POST to `http://fleet-agent-api:8901/register` with JSON payload; returns `True` only on a successful HTTP response. |
317+
| **2. `report_status`** | POST to `http://fleet-agent-api:8901/status` with JSON payload; same success semantics as above. |
318+
| **3. `get_fleet_map`** | GET from `http://fleet-agent-api:8901/fleet`; on failure returns an empty dict (`{}`) – graceful degradation. |
319+
| **4. `send_bottle`** | Writes a JSON file to `BOTTLE_DIR` using a timestamped name, then runs a minimal Git add/commit/push (wrapped in a `try/except` so failures are logged but do not abort the function). |
320+
| **5. `receive_bottles`** | Scans `BOTTLE_DIR` for files whose name contains the `agent_id`, parses them, returns a list, and renames each file with a `.read` suffix to mark it processed. |
321+
| **6. `check_red_alert`** | GET from `http://lighthouse:8901/alerts`; returns a list of alerts or an empty list on error. |
322+
| **Use of urllib & json** | All HTTP interactions use `urllib.request`; payloads are serialized/deserialized with the standard `json` module. |
323+
| **Error handling & retry logic** | The private `_http_request` helper retries up to `MAX_RETRIES` with exponential back‑off, catches network/HTTP errors, and returns `None` on total failure. Each public function checks that result and falls back to a safe default (`False`, `{}`, `[]`). |
324+
| **Graceful degradation** | If the fleet‑agent API or lighthouse is unreachable, the functions still return a deterministic value (e.g., `False` or empty containers) instead of raising. This lets the rest of Holodeck continue operating. |
325+
326+
---
327+
328+
#### Customisation tips
329+
330+
* **BOTTLE_DIR** – change the `Path("/var/holodeck/bottles")` constant to a location that is shared among all agents (NFS mount, Docker volume, etc.).
331+
* **Git broadcast** – replace the placeholder `subprocess.run` block with your preferred Git library or CI‑pipeline trigger.
332+
* **Logging** – swap the `print` statements for a proper `logging` configuration in production.
333+
* **Timeout / retries** – adjust `HTTP_TIMEOUT`, `MAX_RETRIES`, and `BASE_BACKOFF` to match your network reliability requirements.
334+
335+
You can now import the module anywhere in Holodeck Studio:
336+
337+
```python
338+
from fleet_integration import (
339+
register_agent,
340+
report_status,
341+
get_fleet_map,
342+
send_bottle,
343+
receive_bottles,
344+
check_red_alert,
345+
)
346+
347+
# Example usage
348+
if register_agent("agent42", ["navigate", "inspect"]):
349+
report_status("agent42", {"room": "control", "battery": 87})
350+
print(get_fleet_map())
351+
```
352+
353+
Feel free to extend the module (e.g., add authentication headers, richer bottle metadata, etc.) – the core scaffolding is already in place. Happy coding!

0 commit comments

Comments
 (0)