ed/manual_slop
Private
Public Access
0
0
Fork 0
Code Activity

Antigravity is dog shit.

Browse Source
This commit is contained in:
Edward R. Gonzalez
2026-05-20 07:51:58 -04:00
parent 180dc167d2
commit e2305ff49a
15 changed files with 123 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/api_hooks.py
+27 -6
View File
@@ -67,6 +67,7 @@ def _set_app_attr(app: Any, name: str, value: Any) -> None:
setattr(app, name, value)
class HookServerInstance(ThreadingHTTPServer):
allow_reuse_address = True
"""Custom HTTPServer that carries a reference to the main App instance."""
def __init__(self, server_address: tuple[str, int], RequestHandlerClass: type, app: Any) -> None:
"""
@@ -85,6 +86,9 @@ def _serialize_for_api(obj: Any) -> Any:
return [_serialize_for_api(x) for x in obj]
if isinstance(obj, dict):
return {k: _serialize_for_api(v) for k, v in obj.items()}
from pathlib import PurePath
if isinstance(obj, PurePath):
return str(obj)
return obj
class HookHandler(BaseHTTPRequestHandler):
@@ -272,6 +276,13 @@ class HookHandler(BaseHTTPRequestHandler):
files = _get_app_attr(app, "files", [])
screenshots = _get_app_attr(app, "screenshots", [])
self.wfile.write(json.dumps({"files": _serialize_for_api(files), "screenshots": _serialize_for_api(screenshots)}).encode("utf-8"))
elif self.path == "/api/v1/context":
self.send_response(200)
self.send_header("Content-Type", "application/json")
self.end_headers()
from src.app_controller import _api_get_context
ctx_data = _api_get_context(app.controller)
self.wfile.write(json.dumps(_serialize_for_api(ctx_data)).encode("utf-8"))
elif self.path == "/api/metrics/financial":
self.send_response(200)
self.send_header("Content-Type", "application/json")
@@ -765,12 +776,22 @@ class WebSocketServer:
asyncio.set_event_loop(self.loop)
self._stop_event = asyncio.Event()
async def main():
"""
[C: simulation/live_walkthrough.py:module, simulation/ping_pong.py:module, src/gui_2.py:module, tests/mock_concurrent_mma.py:module, tests/mock_gemini_cli.py:module, tests/test_cli_tool_bridge.py:TestCliToolBridge.test_allow_decision, tests/test_cli_tool_bridge.py:TestCliToolBridge.test_deny_decision, tests/test_cli_tool_bridge.py:TestCliToolBridge.test_unreachable_hook_server, tests/test_cli_tool_bridge.py:module, tests/test_cli_tool_bridge_mapping.py:TestCliToolBridgeMapping.test_mapping_from_api_format, tests/test_cli_tool_bridge_mapping.py:module, tests/test_discussion_takes.py:module, tests/test_external_editor_gui.py:module, tests/test_headless_service.py:TestHeadlessStartup.test_headless_flag_triggers_run, tests/test_headless_service.py:TestHeadlessStartup.test_normal_startup_calls_app_run, tests/test_mma_skeleton.py:module, tests/test_orchestrator_pm.py:module, tests/test_orchestrator_pm_history.py:module, tests/test_presets.py:module, tests/test_project_serialization.py:module, tests/test_run_worker_lifecycle_abort.py:module, tests/test_symbol_lookup.py:module, tests/test_system_prompt_exposure.py:module, tests/test_theme_nerv_fx.py:module]
"""
async with serve(self._handler, "127.0.0.1", self.port) as server:
self.server = server
await self._stop_event.wait()
max_retries = 10
current_port = self.port
for attempt in range(max_retries):
try:
async with serve(self._handler, "127.0.0.1", current_port) as server:
self.port = current_port
self.server = server
logging.info(f"WebSocketServer successfully bound to port {self.port}")
await self._stop_event.wait()
break
except OSError as e:
if attempt == max_retries - 1:
logging.error(f"WebSocketServer failed to bind after {max_retries} attempts: {e}")
raise
logging.warning(f"WebSocketServer port {current_port} in use, retrying on {current_port + 1}...")
current_port += 1
self.loop.run_until_complete(main())
def start(self) -> None: