diff --git a/.gitignore b/.gitignore index d060538..55062d2 100644 --- a/.gitignore +++ b/.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 diff --git a/implementations/assistantClient/__pycache__/assistantClient.cpython-311.pyc b/implementations/assistantClient/__pycache__/assistantClient.cpython-311.pyc deleted file mode 100644 index 59864df..0000000 Binary files a/implementations/assistantClient/__pycache__/assistantClient.cpython-311.pyc and /dev/null differ diff --git a/implementations/assistantClient/__pycache__/event_handlers.cpython-311.pyc b/implementations/assistantClient/__pycache__/event_handlers.cpython-311.pyc deleted file mode 100644 index 3f046f0..0000000 Binary files a/implementations/assistantClient/__pycache__/event_handlers.cpython-311.pyc and /dev/null differ diff --git a/implementations/assistantClient/__pycache__/event_handlers.cpython-39.pyc b/implementations/assistantClient/__pycache__/event_handlers.cpython-39.pyc deleted file mode 100644 index 5e8a836..0000000 Binary files a/implementations/assistantClient/__pycache__/event_handlers.cpython-39.pyc and /dev/null differ diff --git a/implementations/assistantClient/__pycache__/floor.cpython-311.pyc b/implementations/assistantClient/__pycache__/floor.cpython-311.pyc deleted file mode 100644 index ef7f7b2..0000000 Binary files a/implementations/assistantClient/__pycache__/floor.cpython-311.pyc and /dev/null differ diff --git a/implementations/assistantClient/__pycache__/floor.cpython-39.pyc b/implementations/assistantClient/__pycache__/floor.cpython-39.pyc deleted file mode 100644 index dee0ba1..0000000 Binary files a/implementations/assistantClient/__pycache__/floor.cpython-39.pyc and /dev/null differ diff --git a/implementations/assistantClient/__pycache__/known_agents.cpython-311-Pegasus.pyc b/implementations/assistantClient/__pycache__/known_agents.cpython-311-Pegasus.pyc deleted file mode 100644 index 5788b7f..0000000 Binary files a/implementations/assistantClient/__pycache__/known_agents.cpython-311-Pegasus.pyc and /dev/null differ diff --git a/implementations/assistantClient/__pycache__/known_agents.cpython-311.pyc b/implementations/assistantClient/__pycache__/known_agents.cpython-311.pyc deleted file mode 100644 index ec53fb8..0000000 Binary files a/implementations/assistantClient/__pycache__/known_agents.cpython-311.pyc and /dev/null differ diff --git a/implementations/assistantClient/__pycache__/known_agents.cpython-39.pyc b/implementations/assistantClient/__pycache__/known_agents.cpython-39.pyc deleted file mode 100644 index 7e30864..0000000 Binary files a/implementations/assistantClient/__pycache__/known_agents.cpython-39.pyc and /dev/null differ diff --git a/implementations/assistantClient/__pycache__/ui_components.cpython-311.pyc b/implementations/assistantClient/__pycache__/ui_components.cpython-311.pyc deleted file mode 100644 index 1cfb18c..0000000 Binary files a/implementations/assistantClient/__pycache__/ui_components.cpython-311.pyc and /dev/null differ diff --git a/implementations/assistantClient/__pycache__/ui_components.cpython-39.pyc b/implementations/assistantClient/__pycache__/ui_components.cpython-39.pyc deleted file mode 100644 index 720ca6c..0000000 Binary files a/implementations/assistantClient/__pycache__/ui_components.cpython-39.pyc and /dev/null differ diff --git a/implementations/assistantClient/known_agents-Pegasus.py b/implementations/assistantClient/known_agents-Pegasus.py index 7d46e42..18cfa86 100644 --- a/implementations/assistantClient/known_agents-Pegasus.py +++ b/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", diff --git a/implementations/assistantClient/known_agents.py b/implementations/assistantClient/known_agents.py index a217385..0094982 100644 --- a/implementations/assistantClient/known_agents.py +++ b/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", diff --git a/implementations/assistantClient/ui_components.py b/implementations/assistantClient/ui_components.py index 6183ca9..678d115 100644 --- a/implementations/assistantClient/ui_components.py +++ b/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") diff --git a/implementations/web-floor/ARCHITECTURE.md b/implementations/web-floor/ARCHITECTURE.md index dda5361..e3c5db0 100644 --- a/implementations/web-floor/ARCHITECTURE.md +++ b/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] diff --git a/implementations/web-floor/api/__pycache__/flask_gateway.cpython-311.pyc b/implementations/web-floor/api/__pycache__/flask_gateway.cpython-311.pyc deleted file mode 100644 index 7b54cf4..0000000 Binary files a/implementations/web-floor/api/__pycache__/flask_gateway.cpython-311.pyc and /dev/null differ diff --git a/implementations/web-floor/api/__pycache__/index.cpython-311.pyc b/implementations/web-floor/api/__pycache__/index.cpython-311.pyc deleted file mode 100644 index 46c8e4b..0000000 Binary files a/implementations/web-floor/api/__pycache__/index.cpython-311.pyc and /dev/null differ diff --git a/implementations/web-floor/api/app.py b/implementations/web-floor/api/app.py index 4c8eec7..f49d980 100644 --- a/implementations/web-floor/api/app.py +++ b/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}) diff --git a/implementations/web-floor/api/flask_gateway.py b/implementations/web-floor/api/flask_gateway.py index 3fa18b6..b03e7ca 100644 --- a/implementations/web-floor/api/flask_gateway.py +++ b/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,14 +391,16 @@ 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) @@ -399,9 +408,32 @@ def floor_stream(): 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") diff --git a/implementations/web-floor/api/floor_router.py b/implementations/web-floor/api/floor_router.py index 1cd6b99..5f23741 100644 --- a/implementations/web-floor/api/floor_router.py +++ b/implementations/web-floor/api/floor_router.py @@ -15,11 +15,14 @@ implements -- so no rewrite is needed when Phase 2 lands. """ +import logging from collections import deque from concurrent.futures import ThreadPoolExecutor from floor_state import ConversationState, normalize_id +logger = logging.getLogger(__name__) + UTTERANCE = "utterance" INVITE = "invite" UNINVITE = "uninvite" @@ -141,13 +144,24 @@ def _extract_dialog_feature_value(event: dict, feature_name: str) -> str: def resolve_pass_through_targets(conv: ConversationState, event: dict, sender_speaker_uri: str) -> list: """Conversants (ConversantState) this event should be delivered to, honoring the utterance `to.private` narrowing. Never includes the - sender itself. For a public (non-private) broadcast, also never - includes the registered convener -- the convener always gets its own - separate, richer delivery (delegation or courtesy-copy, carrying the - conversant roster and round context), never the plain broadcast copy, - so it's never called twice for the same event.""" + sender itself. + + An UTTERANCE also never includes the registered convener in this plain + broadcast -- convener gets its own separate, richer delivery instead + (delegation or courtesy-copy, carrying the conversant roster and round + context), so it's never called twice for the same utterance. + + Every OTHER Pass-Through-Always event type (acceptInvite, + declineInvite, bye, getManifests, publishManifests, yieldFloor) has no + such alternate delivery to convener, so convener is NOT excluded for + those -- it's just another conversant on the broadcast, per the spec's + own Pass-Through semantics. Excluding it there silently meant convener + could never learn a specialist it had asked to be invited actually + accepted (confirmed live: acceptInvite from a newly invited specialist + never reached the convener that requested the invite at all).""" to = _event_to(event) - if event.get("eventType") == UTTERANCE and _is_private(event): + event_type = event.get("eventType") + if event_type == UTTERANCE and _is_private(event): target_speaker = _to_speaker_uri(to) target_service = _to_service_url(to) if target_speaker: @@ -160,7 +174,8 @@ def resolve_pass_through_targets(conv: ConversationState, event: dict, sender_sp return [match] return [] sender_normalized = normalize_id(sender_speaker_uri) - convener_normalized = normalize_id(conv.convener_speaker_uri) if conv.convener_speaker_uri else None + exclude_convener = event_type == UTTERANCE and conv.convener_speaker_uri + convener_normalized = normalize_id(conv.convener_speaker_uri) if exclude_convener else None return [ c for c in conv.conversants.values() if normalize_id(c.speaker_uri) != sender_normalized and normalize_id(c.speaker_uri) != convener_normalized @@ -178,26 +193,82 @@ def _build_outbound_envelope(floor_manager_identity: dict, conv_id: str, event: } -def deliver_and_collect(conversant, event: dict, floor_manager_identity: dict, conv_id: str, deliver, timeout: float) -> list: +def deliver_and_collect(conversant, event: dict, floor_manager_identity: dict, conv_id: str, deliver, timeout: float, on_progress=None) -> list: """Send `event` to one conversant's serviceUrl via the injected `deliver` callback; returns whatever event dicts it replied with (an empty list on failure -- delivery failures must never abort the round). Each reply is stamped with _ORIGIN_KEY: this call is a direct request/response against `conversant`, so any events it hands back unambiguously originated there, even though the reply itself carries - no sender/dialogEvent of its own (e.g. a bare acceptInvite).""" + no sender/dialogEvent of its own (e.g. a bare acceptInvite). + + `on_progress(speaker_uri, service_url, status)` -- status "working" then + "idle" -- is an optional live progress hook, called around the actual + network call so a caller (e.g. flask_gateway.py's streaming endpoint) + can surface real per-conversant activity to a client. May be called from + a worker thread when reached via deliver_concurrently's thread pool, so + it must be safe to call concurrently (a thread-safe queue.put is).""" envelope = _build_outbound_envelope(floor_manager_identity, conv_id, event) + if on_progress: + on_progress(conversant.speaker_uri, conversant.service_url, "working") try: - events = deliver(conversant.service_url, envelope, timeout) or [] - except Exception: - return [] + events = _deliver_with_retry(conversant, envelope, deliver, timeout) + finally: + if on_progress: + on_progress(conversant.speaker_uri, conversant.service_url, "idle") for reply_event in events: if isinstance(reply_event, dict): reply_event[_ORIGIN_KEY] = conversant.speaker_uri return events -def deliver_concurrently(targets: list, event: dict, floor_manager_identity: dict, conv_id: str, deliver, timeout: float) -> list: +def _is_timeout_error(error: Exception) -> bool: + """True for a genuine timeout, whether raised directly (TimeoutError / + socket.timeout, which is TimeoutError as of Python 3.10) or wrapped + inside another exception's .reason (urllib.error.URLError's shape for + a connect-phase timeout) -- checked via duck typing so this stays + transport-agnostic rather than importing a specific HTTP library.""" + if isinstance(error, TimeoutError): + return True + return isinstance(getattr(error, "reason", None), TimeoutError) + + +def _deliver_with_retry(conversant, envelope: dict, deliver, timeout: float) -> list: + """A single transient, FAST-failing delivery error (a dropped + connection, an agent process mid-restart) must not silently lose that + conversant's whole turn for the round -- confirmed live: an agent + crash mid-conversation looks exactly like "that conversant's results + just never showed up," with nothing in the floor manager's own logs to + explain why (delivery failures used to be swallowed with no logging at + all). One immediate retry, without backoff, rides out that kind of + momentary hiccup. + + A genuine TIMEOUT is handled differently on purpose: the full `timeout` + has already been spent waiting once, so retrying would just spend it + again for no benefit -- give up immediately rather than doubling the + wait, and log that the agent was unavailable.""" + try: + return deliver(conversant.service_url, envelope, timeout) or [] + except Exception as first_error: + if _is_timeout_error(first_error): + logger.warning( + "Agent at %s is unavailable (timed out after %.0fs) -- giving up on this turn", + conversant.service_url, timeout, + ) + return [] + logger.warning("Delivery to %s failed (%s), retrying once", conversant.service_url, first_error) + + try: + return deliver(conversant.service_url, envelope, timeout) or [] + except Exception as second_error: + logger.warning( + "Agent at %s is unavailable (failed again after retry: %s) -- giving up on this turn", + conversant.service_url, second_error, + ) + return [] + + +def deliver_concurrently(targets: list, event: dict, floor_manager_identity: dict, conv_id: str, deliver, timeout: float, on_progress=None) -> list: """Deliver `event` to every target concurrently (they're independent recipients of the same one event -- the spec's normative sequential-processing rule governs the EVENT QUEUE, not fan-out to @@ -207,10 +278,10 @@ def deliver_concurrently(targets: list, event: dict, floor_manager_identity: dic if not targets: return [] if len(targets) == 1: - return deliver_and_collect(targets[0], event, floor_manager_identity, conv_id, deliver, timeout) + return deliver_and_collect(targets[0], event, floor_manager_identity, conv_id, deliver, timeout, on_progress) with ThreadPoolExecutor(max_workers=min(MAX_CONCURRENT_DELIVERIES, len(targets))) as pool: futures = [ - pool.submit(deliver_and_collect, target, event, floor_manager_identity, conv_id, deliver, timeout) + pool.submit(deliver_and_collect, target, event, floor_manager_identity, conv_id, deliver, timeout, on_progress) for target in targets ] results = [f.result() for f in futures] @@ -274,15 +345,31 @@ def apply_local_state(conv: ConversationState, event: dict, sender_speaker_uri: _TRUSTED_KEY = "_floorManagerTrusted" -def process_envelope(conv: ConversationState, in_envelope: dict, floor_manager_identity: dict, deliver, timeout: float = 30.0) -> list: +def process_envelope(conv: ConversationState, in_envelope: dict, floor_manager_identity: dict, deliver, timeout: float = 30.0, on_progress=None, on_event=None) -> list: """Process every event in in_envelope's "openFloor.events" list per the routing table, returning the ordered list of events the caller (browser - or convener) should be shown/rendered.""" + or convener) should be shown/rendered. + + `on_progress(speaker_uri, service_url, status)` is an optional live + progress hook (see deliver_and_collect) threaded through to every + delivery call this makes, so a streaming caller can surface real-time + per-conversant activity instead of only a final aggregated result. + + `on_event(event)` is an optional hook called the moment an event is + finalized into `executed` -- e.g. the instant one specialist's reply is + ready, not after the whole round (which can take minutes across many + conversants) finishes. A streaming caller uses this to show each + response as it arrives instead of batching everything until the end.""" openfloor = _unwrap_envelope(in_envelope) raw_events = openfloor.get("events") or [] queue = deque(raw_events) executed = [] + def finalize(event: dict) -> None: + executed.append(event) + if on_event: + on_event(event) + while queue: event = queue.popleft() # Events the convener itself returned are privileged: execute them @@ -297,7 +384,7 @@ def process_envelope(conv: ConversationState, in_envelope: dict, floor_manager_i if event_type == UTTERANCE: if not trusted and not sender_currently_holds_floor(conv, sender_speaker_uri): if conv.convener_speaker_uri: - convener_events = delegate_to_convener(conv, event, sender_speaker_uri, floor_manager_identity, deliver, timeout) + convener_events = delegate_to_convener(conv, event, sender_speaker_uri, floor_manager_identity, deliver, timeout, on_progress) queue.extendleft(reversed(convener_events)) # else: Ignore, per table. continue @@ -322,20 +409,60 @@ def process_envelope(conv: ConversationState, in_envelope: dict, floor_manager_i elif not is_convener_sender: conv.record_turn(sender_speaker_uri, sender_conversant.conversational_name, _extract_utterance_text(event)) + # Cap each specialist to one comment per turn so peer-to-peer + # cross-talk can't cascade: the moment a specialist's own reply + # is about to be broadcast to its peers (Pass-Through, below), + # close ITS OWN local floor gate first via a real, synchronously + # delivered revokeFloor -- so by the time any peer's reply comes + # back around, this conversant no longer holds the floor and + # won't be re-triggered. Skipped for exactly 2 non-convener + # conversants, so a pair of specialists can freely converse. + # + # The revokeFloor is delivered/applied here (early), but NOT + # finalized yet -- finalize it AFTER this specialist's own + # utterance below instead. A streaming caller finalizes/shows + # events in the order it receives them, and a client-side + # "don't show an utterance from an already-revoked speaker" + # guard (a real one exists in web-floor's app.js, for a + # DIFFERENT, legitimate case: a stale/late reply arriving after + # that speaker was revoked for some other reason) would + # otherwise see this specialist as already revoked by the time + # its own triggering utterance arrives, and wrongly suppress it + # -- confirmed live: this made every >2-specialist reply vanish + # from the conversation history while still showing in the raw + # event log. + pending_self_revoke = None + if not is_convener_sender and sender_conversant is not None: + other_conversants = [ + c for c in conv.conversants.values() + if not (conv.convener_speaker_uri and normalize_id(c.speaker_uri) == normalize_id(conv.convener_speaker_uri)) + ] + if len(other_conversants) > 2 and sender_conversant.floor_granted: + revoke_event = { + "eventType": REVOKE_FLOOR, + "to": {"speakerUri": sender_conversant.speaker_uri, "serviceUrl": sender_conversant.service_url}, + } + revoke_replies = deliver_and_collect(sender_conversant, revoke_event, floor_manager_identity, conv.conv_id, deliver, timeout, on_progress) + apply_local_state(conv, revoke_event, sender_speaker_uri) + pending_self_revoke = revoke_event + queue.extend(revoke_replies) + targets = resolve_pass_through_targets(conv, event, sender_speaker_uri) - reply_events = deliver_concurrently(targets, event, floor_manager_identity, conv.conv_id, deliver, timeout) - # NOT executed.extend(reply_events) here -- each reply re-enters - # the SAME table (queue.extend below) and gets appended to - # `executed` exactly once, when ITS OWN turn through the loop - # reaches the "executed.append(event)" below. Adding it here too - # would double-report every specialist reply. + reply_events = deliver_concurrently(targets, event, floor_manager_identity, conv.conv_id, deliver, timeout, on_progress) + # NOT finalize()-ing reply_events here -- each reply re-enters + # the SAME table (queue.extend below) and gets finalized exactly + # once, when ITS OWN turn through the loop reaches the + # "finalize(event)" below. Finalizing it here too would + # double-report every specialist reply. queue.extend(reply_events) # each reply re-enters the SAME table, at the tail if not trusted and conv.convener_speaker_uri: - courtesy_events = deliver_courtesy_copy_to_convener(conv, event, sender_speaker_uri, floor_manager_identity, deliver, timeout) + courtesy_events = deliver_courtesy_copy_to_convener(conv, event, sender_speaker_uri, floor_manager_identity, deliver, timeout, on_progress) queue.extendleft(reversed(courtesy_events)) - executed.append(event) + finalize(event) + if pending_self_revoke is not None: + finalize(pending_self_revoke) continue if event_type in DELEGATABLE_CONTROL: @@ -353,7 +480,7 @@ def process_envelope(conv: ConversationState, in_envelope: dict, floor_manager_i # answer anyway, so this always executes directly instead. target_is_convener = _event_targets_convener(conv, event) if not trusted and conv.convener_speaker_uri and not target_is_convener: - convener_events = delegate_to_convener(conv, event, sender_speaker_uri, floor_manager_identity, deliver, timeout) + convener_events = delegate_to_convener(conv, event, sender_speaker_uri, floor_manager_identity, deliver, timeout, on_progress) queue.extendleft(reversed(convener_events)) continue @@ -379,7 +506,7 @@ def process_envelope(conv: ConversationState, in_envelope: dict, floor_manager_i target = conv.get_conversant(_to_speaker_uri(to)) or conv.get_conversant_by_service_url(_to_service_url(to)) extra = apply_local_state(conv, resolved, sender_speaker_uri) - executed.append(resolved) + finalize(resolved) # Pass-Through: the resolved event must actually reach its target # conversant -- apply_local_state above only updates the floor @@ -387,12 +514,12 @@ def process_envelope(conv: ConversationState, in_envelope: dict, floor_manager_i # own local floor gate (e.g. base_strategy_agent.py's # _floor_granted), which is the real point of grantFloor/revokeFloor. if target is not None: - reply_events = deliver_and_collect(target, resolved, floor_manager_identity, conv.conv_id, deliver, timeout) - # NOT executed.extend(reply_events) -- same reasoning as the + reply_events = deliver_and_collect(target, resolved, floor_manager_identity, conv.conv_id, deliver, timeout, on_progress) + # NOT finalize()-ing reply_events -- same reasoning as the # utterance branch above: each reply (e.g. an acceptInvite in # response to this invite) re-enters the table via - # queue.extend and gets appended to `executed` exactly once - # when its own turn through the loop is processed. + # queue.extend and gets finalized exactly once when its own + # turn through the loop is processed. queue.extend(reply_events) queue.extend(extra) @@ -400,18 +527,19 @@ def process_envelope(conv: ConversationState, in_envelope: dict, floor_manager_i if event_type in PASS_THROUGH_ALWAYS: targets = resolve_pass_through_targets(conv, event, sender_speaker_uri) - reply_events = deliver_concurrently(targets, event, floor_manager_identity, conv.conv_id, deliver, timeout) - executed.extend(reply_events) + reply_events = deliver_concurrently(targets, event, floor_manager_identity, conv.conv_id, deliver, timeout, on_progress) + for reply_event in reply_events: + finalize(reply_event) apply_local_state(conv, event, sender_speaker_uri) if event_type == ACCEPT_INVITE and not conv.convener_speaker_uri: accepting_conversant = conv.get_conversant(sender_speaker_uri) if accepting_conversant is not None: detect_convener_role(conv, accepting_conversant, floor_manager_identity, deliver, timeout) - executed.append(event) + finalize(event) continue # Unknown event type: report it unchanged, do nothing else. - executed.append(event) + finalize(event) return executed @@ -442,15 +570,15 @@ def detect_convener_role(conv: ConversationState, conversant, floor_manager_iden return False -def delegate_to_convener(conv, event, sender_speaker_uri, floor_manager_identity, deliver, timeout): - return _call_convener(conv, event, sender_speaker_uri, floor_manager_identity, deliver, timeout) +def delegate_to_convener(conv, event, sender_speaker_uri, floor_manager_identity, deliver, timeout, on_progress=None): + return _call_convener(conv, event, sender_speaker_uri, floor_manager_identity, deliver, timeout, on_progress) -def deliver_courtesy_copy_to_convener(conv, event, sender_speaker_uri, floor_manager_identity, deliver, timeout): - return _call_convener(conv, event, sender_speaker_uri, floor_manager_identity, deliver, timeout) +def deliver_courtesy_copy_to_convener(conv, event, sender_speaker_uri, floor_manager_identity, deliver, timeout, on_progress=None): + return _call_convener(conv, event, sender_speaker_uri, floor_manager_identity, deliver, timeout, on_progress) -def _call_convener(conv, event, sender_speaker_uri, floor_manager_identity, deliver, timeout): +def _call_convener(conv, event, sender_speaker_uri, floor_manager_identity, deliver, timeout, on_progress=None): convener = conv.convener if convener is None: return [] @@ -473,7 +601,12 @@ def _call_convener(conv, event, sender_speaker_uri, floor_manager_identity, deli } for c in conv.conversants.values() ] - openfloor["conversation"]["floorGranted"] = [c.service_url for c in conv.conversants.values() if c.floor_granted] + # Spec section 1.6: floorGranted is "an array of speakerURIs", not + # serviceUrls -- speakerUri and serviceUrl happen to be identical for + # every agent in this project's own examples, which is what let this + # go unnoticed; a genuinely spec-compliant participant (or this + # project's own tag:-URI manifest fallback) would have them differ. + openfloor["conversation"]["floorGranted"] = [c.speaker_uri for c in conv.conversants.values() if c.floor_granted] openfloor["events"][0] = { **event, "parameters": { @@ -485,10 +618,13 @@ def _call_convener(conv, event, sender_speaker_uri, floor_manager_identity, deli "roundMaxWords": conv.round_max_words, }, } + if on_progress: + on_progress(convener.speaker_uri, convener.service_url, "working") try: - returned_events = deliver(convener.service_url, envelope, timeout) or [] - except Exception: - return [] + returned_events = _deliver_with_retry(convener, envelope, deliver, timeout) + finally: + if on_progress: + on_progress(convener.speaker_uri, convener.service_url, "idle") # Trust whatever convener sends back -- it's privileged, per the spec's # own wording ("the convener is then responsible for returning this # event back to the floor manager OR substituting it with different diff --git a/implementations/web-floor/harness/README.md b/implementations/web-floor/harness/README.md index 91c2b59..b0d18b2 100644 --- a/implementations/web-floor/harness/README.md +++ b/implementations/web-floor/harness/README.md @@ -10,20 +10,18 @@ The normal workflow is GUI-first: ## 1. Prerequisites -- Python environment for implementations/web-floor -- Dependencies for web-floor installed +- Python with Tkinter available - Optional chart dependencies installed if you want popup charts -From implementations/web-floor: +Optional chart packages: ```bash -python -m pip install -r requirements.txt python -m pip install seaborn pandas matplotlib ``` ## 2. Launch The Harness -From implementations/web-floor: +From the `implementations/web-floor` folder: ```bash python ofp_test.py @@ -165,7 +163,6 @@ No manifest returned: - Verify correct target URL, including any required path suffix. - Example local Verity URL is http://localhost:8768/verity/. - Confirm agent process is running and listening on the expected port. -- For gateway mode, verify Flask gateway is running and gateway URL is correct. All results are error: diff --git a/implementations/web-floor/harness/__pycache__/gui.cpython-314.pyc b/implementations/web-floor/harness/__pycache__/gui.cpython-314.pyc deleted file mode 100644 index 990e58d..0000000 Binary files a/implementations/web-floor/harness/__pycache__/gui.cpython-314.pyc and /dev/null differ diff --git a/implementations/web-floor/harness/cli.py b/implementations/web-floor/harness/cli.py index 29b4661..36c1ab1 100644 --- a/implementations/web-floor/harness/cli.py +++ b/implementations/web-floor/harness/cli.py @@ -96,6 +96,17 @@ def pick_targets(args: argparse.Namespace, scenario: dict[str, Any], known_agent return result +def build_grant_floor_payload(target_url: str, client_uri: str, client_url: str) -> dict[str, Any]: + """Build a grantFloor envelope so agents with a floor gate will accept the utterance.""" + return { + "openFloor": { + "conversation": {"id": str(uuid.uuid4()), "conversants": []}, + "sender": {"speakerUri": client_uri, "serviceUrl": client_url}, + "events": [{"eventType": "grantFloor", "to": {"serviceUrl": target_url}}], + } + } + + def build_payload(event_type: str, target_url: str, utterance: str, client_uri: str, client_url: str) -> dict[str, Any]: payload: dict[str, Any] = { "openFloor": { @@ -236,6 +247,15 @@ def run_harness(args: argparse.Namespace) -> int: print(f"Running {total} dispatches: event={event_type}, transport=direct") for agent_url, agent_name in targets: + if event_type == "utterance": + # Send grantFloor first so agents with a floor gate will respond. + # Agents without a floor gate silently ignore it. + grant_payload = build_grant_floor_payload(agent_url, client_uri, client_url) + try: + http_post_json(agent_url, grant_payload, min(timeout_ms, 5000)) + except Exception: + pass + for _ in range(repeat): payload = build_payload(event_type, agent_url, utterance, client_uri, client_url) started = time.perf_counter() diff --git a/implementations/web-floor/harness/gui.py b/implementations/web-floor/harness/gui.py index 1024eab..28041a2 100644 --- a/implementations/web-floor/harness/gui.py +++ b/implementations/web-floor/harness/gui.py @@ -14,8 +14,10 @@ from .cli import ( EVENT_CHOICES, + build_grant_floor_payload, build_payload, classify_received, + http_post_json, send_one, ) @@ -98,6 +100,11 @@ def _capture_base_named_fonts(self) -> None: def _build_menu(self) -> None: menubar = tk.Menu(self.root) + + file_menu = tk.Menu(menubar, tearoff=0) + file_menu.add_command(label="Exit", command=self.root.destroy) + menubar.add_cascade(label="File", menu=file_menu) + view_menu = tk.Menu(menubar, tearoff=0) view_menu.add_radiobutton(label="Zoom 100%", variable=self.zoom_var, value=100, command=lambda: self._apply_zoom(100)) view_menu.add_radiobutton(label="Zoom 150%", variable=self.zoom_var, value=150, command=lambda: self._apply_zoom(150)) @@ -264,8 +271,9 @@ def _build_setup_panel(self, parent: ttk.Frame) -> None: ttk.Entry(event_frame, textvariable=self.expected_var).grid(row=5, column=1, sticky="ew", padx=(8, 0), pady=(8, 0)) ttk.Label(event_frame, text="Timeout ms").grid(row=6, column=0, sticky="w", pady=(8, 0)) - self.timeout_var = tk.IntVar(value=10000) - ttk.Spinbox(event_frame, from_=100, to=120000, increment=100, textvariable=self.timeout_var, width=12).grid(row=6, column=1, sticky="w", padx=(8, 0), pady=(8, 0)) + # LLM-backed agents often exceed 10s; start at 60s by default. + self.timeout_var = tk.IntVar(value=60000) + ttk.Spinbox(event_frame, from_=100, to=600000, increment=100, textvariable=self.timeout_var, width=12).grid(row=6, column=1, sticky="w", padx=(8, 0), pady=(8, 0)) action_frame = ttk.Frame(parent) action_frame.grid(row=row, column=0, sticky="ew", pady=(10, 0)) @@ -405,13 +413,11 @@ def _build_results_panel(self, parent: ttk.Frame) -> None: detail_frame.rowconfigure(0, weight=1) detail_frame.columnconfigure(0, weight=1) - self.detail_text = tk.Text(detail_frame, wrap="none") + self.detail_text = tk.Text(detail_frame, wrap="char") self.detail_text.grid(row=0, column=0, sticky="nsew") detail_v_scroll = ttk.Scrollbar(detail_frame, orient="vertical", command=self.detail_text.yview) detail_v_scroll.grid(row=0, column=1, sticky="ns") - detail_h_scroll = ttk.Scrollbar(detail_frame, orient="horizontal", command=self.detail_text.xview) - detail_h_scroll.grid(row=1, column=0, sticky="ew") - self.detail_text.configure(yscrollcommand=detail_v_scroll.set, xscrollcommand=detail_h_scroll.set) + self.detail_text.configure(yscrollcommand=detail_v_scroll.set) status_frame = ttk.Frame(parent) status_frame.grid(row=2, column=0, sticky="ew", pady=(8, 0)) @@ -988,7 +994,7 @@ def _start_run(self) -> None: utterance = self.utterance_var.get().strip() repeat = max(1, int(self.repeat_var.get() or 1)) expected = self.expected_var.get().strip() - timeout_ms = max(100, int(self.timeout_var.get() or 10000)) + timeout_ms = max(100, int(self.timeout_var.get() or 60000)) if event_type not in EVENT_CHOICES: messagebox.showerror("Validation", "Please select a valid event type.") @@ -1050,6 +1056,17 @@ def _run_worker( for agent_url, agent_name in targets: for utterance in utterances: + if event_type == "utterance": + grant_payload = build_grant_floor_payload( + agent_url, + "openFloor://ofp-test-gui", + "gui://ofp-test-harness", + ) + try: + http_post_json(agent_url, grant_payload, 5000) + except Exception: + pass + for _ in range(repeat): if self._cancel_requested: self.root.after(0, self._finish_run, success_count, fail_count, error_count, completed, total, True) @@ -1074,7 +1091,7 @@ def _run_worker( ok = False status_code = None error = str(exc) - response = None + response = {"error": error, "error_type": type(exc).__name__} duration_ms = int((time.perf_counter() - started) * 1000) response_text = json.dumps(response, ensure_ascii=False) if not isinstance(response, str) else response diff --git a/implementations/web-floor/harness/known_agents.json b/implementations/web-floor/harness/known_agents.json index 54dd8ca..e5a40e4 100644 --- a/implementations/web-floor/harness/known_agents.json +++ b/implementations/web-floor/harness/known_agents.json @@ -11,5 +11,15 @@ {"url": "http://localhost:8084/", "conversationalName": "Prudence"}, {"url": "http://localhost:8085/", "conversationalName": "Lucky"}, {"url": "https://bladeszasza-ofpbadword.hf.space/ofp", "conversationalName": ""}, - {"url": "https://yahandhjjf.us-east-1.awsapprunner.com/", "conversationalName": ""} + {"url": "https://yahandhjjf.us-east-1.awsapprunner.com/", "conversationalName": ""}, + {"url": "http://localhost:8199/", "conversationalName": "Convener (Strategy)"}, + {"url": "http://localhost:8200/", "conversationalName": "Market Validator"}, + {"url": "http://localhost:8201/", "conversationalName": "Competitive Intelligence"}, + {"url": "http://localhost:8202/", "conversationalName": "Business Model Designer"}, + {"url": "http://localhost:8203/", "conversationalName": "Risk Identifier"}, + {"url": "http://localhost:8204/", "conversationalName": "Funding Strategist"}, + {"url": "http://localhost:8205/", "conversationalName": "Workforce Strategist"}, + {"url": "http://localhost:8208/", "conversationalName": "Technical Feasibility"}, + {"url": "http://localhost:8206/", "conversationalName": "Devil's Advocate"}, + {"url": "http://localhost:8207/", "conversationalName": "Strategy Synthesizer"} ] diff --git a/implementations/web-floor/public/app.js b/implementations/web-floor/public/app.js index 07b41fd..5aad573 100644 --- a/implementations/web-floor/public/app.js +++ b/implementations/web-floor/public/app.js @@ -63,6 +63,14 @@ const KNOWN_AGENTS = [ { url: "http://127.0.0.1:8208/", conversationalName: "Technical Feasibility" }, { url: "http://127.0.0.1:8206/", conversationalName: "Devil's Advocate" }, { url: "http://127.0.0.1:8207/", conversationalName: "Strategy Synthesizer" }, + { url: "http://127.0.0.1:8300/", conversationalName: "Cafeteria Ops Convener" }, + { url: "http://127.0.0.1:8301/", conversationalName: "Menu Designer" }, + { url: "http://127.0.0.1:8302/", conversationalName: "Nutrition Specialist" }, + { url: "http://127.0.0.1:8303/", conversationalName: "Recipe & Portion Specialist" }, + { url: "http://127.0.0.1:8304/", conversationalName: "Menu Optimization Specialist" }, + { url: "http://127.0.0.1:8305/", conversationalName: "Inventory Specialist" }, + { url: "http://127.0.0.1:8306/", conversationalName: "Procurement Specialist" }, + { url: "http://127.0.0.1:8310/", conversationalName: "Shopping List Specialist" }, { url: "https://bladeszasza-ofpbadword.hf.space/ofp", conversationalName: "" }, { url: "https://yahandhjjf.us-east-1.awsapprunner.com/", conversationalName: "" } ]; @@ -1448,7 +1456,14 @@ function buildAndQueueCombinedReport(htmlReportEntries) { const esc = s => (s || "").replace(/&/g, "&").replace(//g, ">"); const sections = htmlReportEntries.map(entry => { const nameHtml = `

${esc(entry.name)}

`; - const textHtml = entry.text + // entry.html, when present, is already a full rendering of entry.text + // (a real