Remote deploy console: stream agent compose up to the browser (0.23.0)
Extends the live deploy console to remote/agent stacks. New agent WS endpoint
`/agent/ws/deploy/{stack_id}` runs `compose up -d` and streams its output; the
central app proxies it through `/ws/agent-deploy/{agent_id}/{stack_id}` (same
pattern + token URL-encoding as the agent-logs proxy) and records an
`agent.stack.start` audit entry. The editor's remote Deploy path now opens the
DeployConsole (agentId) instead of the blocking `agentsApi.action(start)`.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 4.8
parent
7592085ce9
commit
d46a6c3576
+33
-1
@@ -62,7 +62,7 @@ def _map_docker(exc: DockerError):
|
||||
raise HTTPException(status_code=code, detail=exc.detail or exc.error)
|
||||
raise exc # falls through to the global 502 DockerError handler
|
||||
|
||||
AGENT_VERSION = "0.22.0"
|
||||
AGENT_VERSION = "0.23.0"
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
@@ -640,6 +640,38 @@ async def ws_logs(websocket: WebSocket, stack_id: str, token: str | None = Query
|
||||
pass
|
||||
|
||||
|
||||
@app.websocket("/agent/ws/deploy/{stack_id}")
|
||||
async def ws_deploy(websocket: WebSocket, stack_id: str, token: str | None = Query(default=None)):
|
||||
"""Run `docker compose up -d` and stream its output to the central app so the
|
||||
browser sees deploy progress live (token via query param)."""
|
||||
await websocket.accept()
|
||||
expected = settings.AGENT_TOKEN
|
||||
if not expected or token != expected:
|
||||
await websocket.close(code=4401)
|
||||
return
|
||||
if not os.path.isdir(compose_service.stack_dir(stack_id)):
|
||||
await websocket.send_text(json.dumps({"type": "error", "detail": "stack not found"}))
|
||||
await websocket.close()
|
||||
return
|
||||
compose_service.mark_busy(stack_id)
|
||||
try:
|
||||
async for kind, payload in compose_service.stream_up(stack_id):
|
||||
if kind == "log":
|
||||
await websocket.send_text(json.dumps({"type": "log", "line": payload}))
|
||||
else:
|
||||
await websocket.send_text(json.dumps({"type": "done", "returncode": payload}))
|
||||
except WebSocketDisconnect:
|
||||
# Browser navigated away; the compose subprocess keeps running.
|
||||
pass
|
||||
except Exception as exc: # noqa: BLE001
|
||||
try:
|
||||
await websocket.send_text(json.dumps({"type": "error", "detail": str(exc)}))
|
||||
except Exception: # noqa: BLE001
|
||||
pass
|
||||
finally:
|
||||
compose_service.clear_busy(stack_id)
|
||||
|
||||
|
||||
@app.get("/agent/health")
|
||||
def health() -> dict:
|
||||
return {"status": "ok"}
|
||||
|
||||
+1
-1
@@ -55,7 +55,7 @@ async def lifespan(app: FastAPI):
|
||||
schedule_task.cancel()
|
||||
|
||||
|
||||
app = FastAPI(title="StackPilot", version="0.22.0", lifespan=lifespan)
|
||||
app = FastAPI(title="StackPilot", version="0.23.0", lifespan=lifespan)
|
||||
|
||||
app.add_middleware(
|
||||
CORSMiddleware,
|
||||
|
||||
@@ -223,6 +223,85 @@ async def ws_agent_logs(
|
||||
await websocket.close()
|
||||
|
||||
|
||||
@router.websocket("/ws/agent-deploy/{agent_id}/{stack_id}")
|
||||
async def ws_agent_deploy(
|
||||
websocket: WebSocket,
|
||||
agent_id: int,
|
||||
stack_id: str,
|
||||
token: str | None = Query(default=None),
|
||||
):
|
||||
"""Proxy a remote agent's `compose up` deploy stream through to the browser,
|
||||
then record the same audit entry as the REST agent lifecycle endpoint."""
|
||||
await websocket.accept()
|
||||
if not await _authorize(websocket, token):
|
||||
return
|
||||
username = decode_token(token, "access").get("sub", "unknown") if token else "unknown"
|
||||
|
||||
with Session(engine) as session:
|
||||
agent = session.get(Agent, agent_id)
|
||||
if not agent:
|
||||
await websocket.send_text(json.dumps({"type": "error", "detail": "agent not found"}))
|
||||
await websocket.close()
|
||||
return
|
||||
|
||||
base = agent.url.rstrip("/")
|
||||
ws_url = ("wss://" + base[8:] if base.startswith("https://")
|
||||
else "ws://" + base[7:] if base.startswith("http://")
|
||||
else "ws://" + base)
|
||||
ws_url += f"/agent/ws/deploy/{stack_id}?token={urllib.parse.quote(agent.token, safe='')}"
|
||||
|
||||
async def _err(detail: str) -> None:
|
||||
with contextlib.suppress(Exception):
|
||||
await websocket.send_text(json.dumps({"type": "error", "detail": detail}))
|
||||
|
||||
try:
|
||||
upstream = await websockets.connect(ws_url, open_timeout=10, ping_interval=20)
|
||||
except websockets.InvalidStatus as exc:
|
||||
code = getattr(getattr(exc, "response", None), "status_code", None)
|
||||
hint = " — the agent may be running an old version without deploy-console support; update it." if code == 404 else ""
|
||||
logger.warning("Agent deploy proxy: handshake to %s failed (%s)", agent.name, code)
|
||||
await _err(f"Agent '{agent.name}' rejected the deploy stream (HTTP {code}){hint}")
|
||||
with contextlib.suppress(Exception):
|
||||
await websocket.close()
|
||||
return
|
||||
except Exception as exc: # noqa: BLE001
|
||||
logger.warning("Agent deploy proxy: cannot reach %s at %s: %s", agent.name, agent.url, exc)
|
||||
await _err(f"Could not connect to agent '{agent.name}' at {agent.url}: {exc}")
|
||||
with contextlib.suppress(Exception):
|
||||
await websocket.close()
|
||||
return
|
||||
|
||||
rc: int | None = None
|
||||
try:
|
||||
async for message in upstream:
|
||||
text = message if isinstance(message, str) else message.decode("utf-8", "replace")
|
||||
with contextlib.suppress(Exception):
|
||||
msg = json.loads(text)
|
||||
if msg.get("type") == "done":
|
||||
rc = msg.get("returncode")
|
||||
await websocket.send_text(text)
|
||||
except WebSocketDisconnect:
|
||||
pass
|
||||
except websockets.ConnectionClosed as exc:
|
||||
if exc.code not in (1000, 1001):
|
||||
await _err(f"Agent deploy stream closed unexpectedly (code {exc.code}).")
|
||||
except Exception as exc: # noqa: BLE001
|
||||
logger.warning("Agent deploy proxy: stream error from %s: %s", agent.name, exc)
|
||||
await _err(str(exc))
|
||||
finally:
|
||||
with contextlib.suppress(Exception):
|
||||
await upstream.close()
|
||||
with contextlib.suppress(Exception):
|
||||
await websocket.close()
|
||||
|
||||
with contextlib.suppress(Exception):
|
||||
with Session(engine) as session:
|
||||
audit_service.record(
|
||||
session, user=username, action="agent.stack.start",
|
||||
target=f"{agent.name}/{stack_id}", detail=f"rc={rc} (deploy console)", ip="ws",
|
||||
)
|
||||
|
||||
|
||||
@router.websocket("/ws/events")
|
||||
async def ws_events(
|
||||
websocket: WebSocket,
|
||||
|
||||
Reference in New Issue
Block a user