feat(web-floor): make floor streaming incremental and expand the harness roster

.gitignore

@@ -1,7 +1,21 @@
 /implementations/assistantClient/openfloor
-implementations/web-floor/__pycache__/flask_gateway.cpython-311.pyc
 
+# Python bytecode / caches
+__pycache__/
+*.pyc
+*.pyo
+
+# Editor / workspace
 .vercel
+*.code-workspace
+
+# Local scratch: prompt drafts, transcripts, harness run output
+implementations/web-floor/convener.txt
+implementations/web-floor/convenerPrompt.txt
+implementations/web-floor/financialDemo.txt
+implementations/web-floor/mermaid-test.md
+implementations/web-floor/harness/results*.json
+implementations/web-floor/harness/load_agents_startup_strategy.json
 implementations/assistantClient/assistantClientSpeech.py
 implementations/assistantClient/assistantClient-Pegasus.py
 implementations/assistantClient/assistantClient-Pegasus.py

implementations/assistantClient/known_agents-Pegasus.py

@@ -4,6 +4,7 @@
 KNOWN_AGENTS = [
     "http://localhost:8767/",
     "http://secondAssistant.pythonanywhere.com/verity/",
+    "http://localhost:8208/",
     "openvoice-stella.vercel.app",
     "http://openvoice-stella.vercel.app",
     "https://bladeszasza-ofpbadword.hf.space/ofp",

implementations/assistantClient/known_agents.py

@@ -5,6 +5,7 @@
     "http://192.168.4.51:8080/",
     "http://192.168.4.51:8767/",
     "http://192.168.4.51:8768/",
+    "http://localhost:8208/",
     "http://secondAssistant.pythonanywhere.com/verity/",
     "https://openvoice-stella.vercel.app/",
     "https://bladeszasza-ofpbadword.hf.space/ofp",

implementations/assistantClient/ui_components.py

@@ -62,6 +62,16 @@ def _ensure_windows_ico() -> None:
         # Pillow is commonly available; use it to generate a proper multi-size ICO.
         from PIL import Image
 
+        def _make_about_half_size(src: "Image.Image") -> "Image.Image":
+            """Reduce the image to roughly half its current size while preserving aspect ratio."""
+            width, height = src.size
+            if width <= 1 and height <= 1:
+                return src
+            new_width = max(1, int(round(width * 0.5)))
+            new_height = max(1, int(round(height * 0.5)))
+            resample = getattr(Image, "Resampling", Image).LANCZOS
+            return src.resize((new_width, new_height), resample)
+
         def _translated_rgba(src: "Image.Image", dx: int, dy: int) -> "Image.Image":
             dst = Image.new("RGBA", src.size, (0, 0, 0, 0))
             if dx == 0 and dy == 0:
@@ -144,6 +154,13 @@ def _stroke_thicken(src: "Image.Image", passes: int) -> "Image.Image":
                 except Exception:
                     pass
 
+            # Make the final exported image about 50% smaller while keeping the
+            # aspect ratio intact for the PNG/ICO outputs.
+            try:
+                im = _make_about_half_size(im)
+            except Exception:
+                pass
+
             # Save derived bold PNG so iconphoto() matches the Windows ICO.
             try:
                 im.save(_APP_ICON_PATH, format="PNG")

implementations/web-floor/ARCHITECTURE.md

@@ -15,7 +15,7 @@ conversant/floor state for the conversation and decides where each event
 actually goes.
 
 ```mermaid
-flowchart LR
+graph LR
   U[User in Browser]
 
   subgraph B[Browser Client]
@@ -103,7 +103,7 @@ implementation: a stateless single-target relay with no floor/conversant
 tracking, for manually poking one agent directly from the UI.
 
 ```mermaid
-flowchart LR
+graph LR
   U[User in Browser]
 
   subgraph B[web-floor Browser Client]

implementations/web-floor/api/app.py

@@ -1,4 +1,7 @@
 #!/usr/bin/env python3
+# NOTE: flask_gateway.py is the canonical gateway used by the Vercel entrypoint
+# (index.py). Keep this file's proxy logic in sync with it, or prefer importing
+# from flask_gateway directly.
 import json
 import mimetypes
 import os
@@ -7,7 +10,6 @@
 from urllib.request import Request, urlopen
 from urllib.error import HTTPError, URLError
 from flask import Flask, request, jsonify, send_from_directory
-from flask_cors import CORS
 
 # Fix Windows registry often mapping .js to text/plain
 mimetypes.add_type("application/javascript", ".js")
@@ -18,27 +20,44 @@
 
 app = Flask(__name__, static_folder=str(PUBLIC_DIR), static_url_path="")
 
-# Read allowed origins from environment variable (comma-separated)
-allowed_origins = os.environ.get("ALLOWED_ORIGINS", "").split(",")
-
-# Remove empty strings and strip whitespace
-allowed_origins = [origin.strip() for origin in allowed_origins if origin.strip()]
-
-# Apply CORS
-CORS(app, origins=allowed_origins)
-
 def _parse_csv_env(value: str) -> list[str]:
 	return [item.strip() for item in (value or "").split(",") if item.strip()]
 
 CORS_ORIGINS = _parse_csv_env(os.environ.get("CORS_ALLOW_ORIGINS", "*")) or ["*"]
 TARGET_ALLOWLIST = _parse_csv_env(os.environ.get("GATEWAY_TARGET_ALLOWLIST", ""))
 
+MAX_PROXY_TIMEOUT_SECONDS = 240.0
+MIN_UTTERANCE_TIMEOUT_SECONDS = 180.0
+
 def _normalize_timeout_seconds(timeout_ms) -> float:
 	try:
 		timeout = float(timeout_ms) / 1000.0
 	except (TypeError, ValueError):
 		timeout = 10.0
-	return max(0.1, min(timeout, 60.0))
+	return max(0.1, min(timeout, MAX_PROXY_TIMEOUT_SECONDS))
+
+
+def _contains_utterance_event(payload: object) -> bool:
+	if not isinstance(payload, dict):
+		return False
+	envelope = payload.get("openFloor") or payload.get("openfloor") or payload.get("ovon") or payload
+	if not isinstance(envelope, dict):
+		return False
+	events = envelope.get("events")
+	if not isinstance(events, list):
+		return False
+	for event in events:
+		if isinstance(event, dict) and event.get("eventType") == "utterance":
+			return True
+	return False
+
+
+def _effective_timeout_seconds(timeout_ms, payload: object) -> float:
+	timeout = _normalize_timeout_seconds(timeout_ms)
+	# Strategy convener fan-out can exceed 2 minutes under load.
+	if _contains_utterance_event(payload):
+		return max(timeout, MIN_UTTERANCE_TIMEOUT_SECONDS)
+	return timeout
 
 def _is_allowed_target(target_url: str) -> tuple[bool, str]:
 	parsed = urlparse(target_url)
@@ -69,7 +88,9 @@ def proxy_send():
 	body = request.get_json(silent=True) or {}
 	target_url = body.get("targetUrl")
 	payload = body.get("payload") or {}
-	timeout_seconds = _normalize_timeout_seconds(body.get("timeoutMs", 10000))
+	# Let _effective_timeout_seconds decide: control events use the caller's
+	# timeout; only utterance events are raised to the long fan-out floor.
+	timeout_seconds = _effective_timeout_seconds(body.get("timeoutMs", 10000), payload)
 	if not isinstance(target_url, str) or not target_url.strip():
 		return jsonify({"error": "targetUrl is required"}), 400
 	target_url = target_url.strip()
@@ -103,8 +124,6 @@ def health():
 
 @app.route("/", methods=["GET"])
 def index():
-	index_path = (PUBLIC_DIR / "index.html").resolve()
-	print(f"[DEBUG] index.html resolved path: {index_path}")
 	response = send_from_directory(PUBLIC_DIR, "index.html")
 	response.headers["Cache-Control"] = "no-cache, no-store, must-revalidate"
 	response.headers["Pragma"] = "no-cache"
@@ -122,14 +141,3 @@ def serve_static(asset_path: str):
 	response.headers["Expires"] = "0"
 	return response
 
-@app.route("/debug-files", methods=["GET"])
-def debug_files():
-	files = []
-	try:
-		for root, dirs, filenames in os.walk(PUBLIC_DIR):
-			for filename in filenames:
-				rel_path = os.path.relpath(os.path.join(root, filename), PUBLIC_DIR)
-				files.append(rel_path)
-	except Exception as e:
-		return jsonify({"error": str(e)}), 500
-	return jsonify({"files": files})

implementations/web-floor/api/flask_gateway.py

@@ -11,7 +11,9 @@
 import json
 import mimetypes
 import os
+import queue
 import sys
+import threading
 from pathlib import Path
 from urllib.parse import urlparse
 from urllib.request import Request, urlopen
@@ -306,7 +308,15 @@ def _deliver_via_http(target_url: str, envelope: dict, timeout: float) -> list:
     to actually reach a conversant's serviceUrl. Mirrors /api/proxy-send's
     request-building and SSRF allowlist, but returns the parsed list of
     reply events directly instead of a wrapped proxy response, since this
-    is called internally by floor_router's routing loop, not by a browser."""
+    is called internally by floor_router's routing loop, not by a browser.
+
+    Connection-level failures (refused, reset, timed out) are intentionally
+    NOT swallowed here -- they propagate to floor_router.py's delivery
+    layer (deliver_and_collect/_deliver_with_retry), which is what decides
+    whether to retry and logs the failure. Swallowing them here made every
+    agent failure silent and indistinguishable from "the agent legitimately
+    said nothing at all", which is exactly what made a real agent crash
+    mid-conversation invisible in the floor manager's own logs."""
     allowed, _reason = _is_allowed_target(target_url)
     if not allowed:
         return []
@@ -321,11 +331,8 @@ def _deliver_via_http(target_url: str, envelope: dict, timeout: float) -> list:
             "User-Agent": "web-floor-flask-gateway/0.1",
         },
     )
-    try:
-        with urlopen(outbound, timeout=timeout) as response:
-            raw_text = response.read().decode("utf-8", errors="replace")
-    except (HTTPError, URLError, Exception):
-        return []
+    with urlopen(outbound, timeout=timeout) as response:
+        raw_text = response.read().decode("utf-8", errors="replace")
     try:
         parsed = json.loads(raw_text)
     except json.JSONDecodeError:
@@ -384,24 +391,49 @@ def floor_stream_options():
 
 @app.route("/api/floor/stream", methods=["POST"])
 def floor_stream():
-    # NDJSON counterpart to /api/floor/send. Phase 1 note: this runs the
-    # full exchange synchronously and yields one aggregated envelope --
-    # true per-turn incremental flushing (matching the old
-    # /round-robin-stream's one-line-per-turn behavior) needs a
-    # generator-based floor_router.process_envelope, which lands with the
-    # round-robin decision loop in Phase 3. Kept as a real NDJSON response
-    # now (not a stub) so app.js's cutover in Phase 4 doesn't need to
-    # distinguish "streaming not implemented yet" as a separate case.
+    # NDJSON counterpart to /api/floor/send, now genuinely incremental:
+    # process_envelope runs on a background thread while this generator
+    # drains a live queue, so both a conversant's real working/idle
+    # transition (on_progress) AND each finalized event -- most
+    # importantly, an utterance the moment that reply is ready -- (on_event)
+    # reach the client as they happen, not batched into one aggregated
+    # envelope only once the whole exchange (which can run minutes for a
+    # big full-sweep round) has finished. Line shapes: zero or more
+    # {"progress": {...}} and {"event": {...}} lines, interleaved in the
+    # order they actually occurred, followed by one {"done": true} line.
     body = request.get_json(silent=True) or {}
     payload = body.get("payload") or {}
     timeout_seconds = _effective_timeout_seconds(body.get("timeoutMs", 10000), payload)
     conv_id = _conversation_id(payload)
 
     def generate():
         conv = floor_registry.get_or_create(conv_id)
-        with conv.lock:
-            executed = floor_router.process_envelope(conv, payload, FLOOR_MANAGER_IDENTITY, _deliver_via_http, timeout_seconds)
-        yield (json.dumps(_build_response_envelope(conv_id, executed)) + "\n").encode("utf-8")
+        live_queue = queue.Queue()
+
+        def on_progress(speaker_uri, service_url, status):
+            live_queue.put({"progress": {"speakerUri": speaker_uri, "serviceUrl": service_url, "status": status}})
+
+        def on_event(event):
+            live_queue.put({"event": event})
+
+        def run():
+            try:
+                with conv.lock:
+                    floor_router.process_envelope(
+                        conv, payload, FLOOR_MANAGER_IDENTITY, _deliver_via_http, timeout_seconds, on_progress, on_event
+                    )
+            finally:
+                live_queue.put(None)  # sentinel: processing finished, no more lines
+
+        threading.Thread(target=run, daemon=True).start()
+
+        while True:
+            item = live_queue.get()
+            if item is None:
+                break
+            yield (json.dumps(item) + "\n").encode("utf-8")
+
+        yield (json.dumps({"done": True}) + "\n").encode("utf-8")
 
     return Response(stream_with_context(generate()), mimetype="application/x-ndjson")