diff --git a/backend/app/routes/helpers.py b/backend/app/routes/helpers.py index 3c66cfa..76f861d 100644 --- a/backend/app/routes/helpers.py +++ b/backend/app/routes/helpers.py @@ -998,6 +998,10 @@ def _event_matches_session(event_lines: list[str], session_id: str) -> bool: except json.JSONDecodeError: return True + # Unwrap GlobalEvent envelope: { directory, project?, workspace?, payload: { type, properties: { sessionID, ... } } } + inner = parsed.get("payload") if isinstance(parsed, dict) else None + target = inner if (isinstance(inner, dict) and isinstance(inner.get("type"), str)) else parsed + keys = {"sessionID", "sessionId", "session_id", "session"} def _walk(value) -> bool: @@ -1013,7 +1017,7 @@ def _walk(value) -> bool: return True return False - return _walk(parsed) + return _walk(target) def _provider_headers(api_key: str) -> dict[str, str]: headers: dict[str, str] = {} diff --git a/backend/app/routes/messages.py b/backend/app/routes/messages.py index 8190086..bc2b1ea 100644 --- a/backend/app/routes/messages.py +++ b/backend/app/routes/messages.py @@ -354,6 +354,8 @@ def _stream(): continue if normalized_line.startswith(":"): + # Forward SSE comment heartbeats to keep connection alive + yield f"{normalized_line}\n" continue event_lines.append(normalized_line) except Exception: @@ -370,7 +372,6 @@ def _stream(): mimetype="text/event-stream", headers={ "Cache-Control": "no-cache", - "Connection": "keep-alive", "X-Accel-Buffering": "no", }, ) @@ -417,7 +418,6 @@ def _stream(): mimetype="text/event-stream", headers={ "Cache-Control": "no-cache", - "Connection": "keep-alive", "X-Accel-Buffering": "no", }, ) diff --git a/backend/requirements.txt b/backend/requirements.txt index 9600b38..15c2a5a 100644 --- a/backend/requirements.txt +++ b/backend/requirements.txt @@ -9,3 +9,4 @@ faster-whisper==1.1.1 TTS==0.22.0 soundfile==0.12.1 GitPython==3.1.41 +waitress==3.0.2 diff --git a/backend/run.py b/backend/run.py index 7666086..f843411 100644 --- a/backend/run.py +++ b/backend/run.py @@ -1,11 +1,53 @@ import os +import sys + +from waitress import serve from app import create_app +def _env_truthy(name: str) -> bool: + return os.environ.get(name, "").lower() in ("1", "true", "yes") + + +def _read_positive_int_env(name: str, default: int) -> int: + raw_value = os.environ.get(name) + if raw_value is None: + return default + + try: + value = int(raw_value) + except ValueError: + return default + + return value if value > 0 else default + + app = create_app() BACKEND_PORT = int(os.environ.get("BACKEND_PORT", "38473")) +WAITRESS_THREADS = _read_positive_int_env("WAITRESS_THREADS", 16) +WAITRESS_CHANNEL_TIMEOUT = _read_positive_int_env("WAITRESS_CHANNEL_TIMEOUT", 300) if __name__ == "__main__": - app.run(host="0.0.0.0", port=BACKEND_PORT, debug=True) + if _env_truthy("FLASK_DEBUG"): + debug_host = "0.0.0.0" if _env_truthy("FLASK_DEBUG_BIND_ALL") else "127.0.0.1" + app.run( + host=debug_host, + port=BACKEND_PORT, + threaded=True, + use_reloader=_env_truthy("FLASK_DEBUG_RELOAD"), + ) + else: + try: + serve( + app, + host="0.0.0.0", + port=BACKEND_PORT, + threads=WAITRESS_THREADS, + channel_timeout=WAITRESS_CHANNEL_TIMEOUT, + ) + except OSError as exc: + print(f"Failed to bind to port {BACKEND_PORT}: {exc}", file=sys.stderr) + print("Is another instance already running? Check: ss -tlnp | grep {BACKEND_PORT}".format(BACKEND_PORT=BACKEND_PORT), file=sys.stderr) + sys.exit(1) diff --git a/frontend/src/App.tsx b/frontend/src/App.tsx index d340a09..50c5dae 100644 --- a/frontend/src/App.tsx +++ b/frontend/src/App.tsx @@ -83,7 +83,7 @@ import { formatCompactSessionId, formatElapsedShort, formatSessionOptionLabel, f import { buildGroupedTimelineEntries, buildMessageStableKey, getNonTextParts, timelineEventToTaskRun } from "./utils/messageUtils"; import { buildProjectPathFromRoot, extractRootFromProjectPath, getSuggestedProjectRoot, normalizeProjectRootPath, projectInitials } from "./utils/projectUtils"; import { toDateInputPartsInTimezone, toDateTimeInputValueInTimezone, toIsoInTimezone } from "./utils/taskUtils"; -import { parseApprovalFromStreamData, parseQuestionFromStreamData, classifyStreamEvent, extractMessagePartText } from "./utils/streamUtils"; +import { parseApprovalFromStreamData, parseQuestionFromStreamData, classifyStreamEvent, extractMessageFromEvent, extractMessagePartText, extractPartFromEvent, extractPartRemovalFromEvent, extractDiffFromEvent, extractSessionStatusFromEvent } from "./utils/streamUtils"; import { buildInitialQuestionDraft, buildQuestionReplyAnswers, getManualInstallMessage, inferTelemetryCategory, markerTimeWindowMs, nextReconnectDelayMs, parseSlashCommand, removeQuestionFromList, resolveDevFixtureMode, schedulerHeartbeatState, scrollEntryIntoView } from "./utils/miscUtils"; import { LoginView } from "./components/auth/LoginView"; import { AgentActivityCard } from "./components/chat/AgentActivityCard"; @@ -2339,8 +2339,18 @@ async function loadDiff(projectId: string) { return; } + // Parse event lines from the SSE wrapper + const eventLines = ((): string[] => { + try { + const wrapper = JSON.parse(event.data) as { event?: string[] }; + return Array.isArray(wrapper.event) ? wrapper.event : []; + } catch { + return []; + } + })(); + + // --- Approvals --- const parsed = parseApprovalFromStreamData(event.data); - const parsedQuestion = parseQuestionFromStreamData(event.data); if (parsed.request) { setPendingApprovals((current) => { if (current.some((item) => item.permissionId === parsed.request?.permissionId)) { @@ -2354,6 +2364,9 @@ async function loadDiff(projectId: string) { current.filter((item) => item.permissionId !== parsed.resolvedPermissionId) ); } + + // --- Questions --- + const parsedQuestion = parseQuestionFromStreamData(event.data); if (parsedQuestion.request && parsedQuestion.request.sessionID === activeSessionId) { setPendingQuestions((current) => { if (current.some((item) => item.id === parsedQuestion.request?.id)) { @@ -2379,50 +2392,160 @@ async function loadDiff(projectId: string) { return next; }); } - let needsFullRefresh = classification.hasMessageUpdate || classification.hasCompactionUpdate; - if (classification.hasPartUpdate) { - const eventLines = ((): string[] => { - try { - const wrapper = JSON.parse(event.data) as { event?: string[] }; - return Array.isArray(wrapper.event) ? wrapper.event : []; - } catch { - return []; - } - })(); + + // --- Text delta (message.part.delta) — inline incremental update --- + if (classification.hasPartDelta) { const partData = extractMessagePartText(eventLines); - if (partData && partData.text != null && partData.messageID && partData.partType === "text") { - const trackedId = lastAssistantMessageIdBySessionRef.current[activeSessionId ?? ""]; + if (partData && partData.text != null && partData.partID && (partData.messageID || activeSessionId)) { + const targetMsgId = partData.messageID || lastAssistantMessageIdBySessionRef.current[activeSessionId ?? ""]; + if (!targetMsgId) return; setMessages((current) => { - const targetId = trackedId || partData.messageID; - const msgIndex = current.findIndex((m) => m.id === targetId); - if (msgIndex < 0) { - return current; - } + const msgIndex = current.findIndex((m) => m.id === targetMsgId); + if (msgIndex < 0) return current; + const existing = current[msgIndex]; const updatedParts = [...existing.parts]; const textPartIndex = updatedParts.findIndex( (p) => typeof p === "object" && p !== null && p.type === "text" && p.id === partData.partID ); if (textPartIndex >= 0) { - updatedParts[textPartIndex] = { ...updatedParts[textPartIndex], text: partData.text }; - } else if (partData.partID) { - updatedParts.push({ type: "text", id: partData.partID, text: partData.text }); + const existingText = (updatedParts[textPartIndex] as Record).text as string || ""; + updatedParts[textPartIndex] = { ...updatedParts[textPartIndex], text: existingText + (partData.text || "") }; } else { - updatedParts.push({ type: "text", text: partData.text }); + updatedParts.push({ type: "text", id: partData.partID, text: partData.text }); } - const allTextParts = updatedParts - .filter((p) => typeof p === "object" && p !== null && p.type === "text" && typeof p.text === "string") - .map((p) => (p as Record).text); - const newText = allTextParts.join("\n").trim(); + const allText = updatedParts + .filter((p) => typeof p === "object" && p !== null && p.type === "text") + .map((p) => (p as Record).text || "") + .join("\n").trim(); const next = [...current]; - next[msgIndex] = { ...existing, text: newText, parts: updatedParts }; + next[msgIndex] = { ...existing, text: allText, parts: updatedParts }; textDeltaScrollRef.current += 1; return next; }); - } else { - needsFullRefresh = true; } } + + // --- Part update (message.part.updated) — inline full part --- + if (classification.hasPartUpdate) { + const extracted = extractPartFromEvent(eventLines); + if (extracted && extracted.part) { + const part = extracted.part; + const msgID = typeof part.messageID === "string" ? part.messageID : null; + const partID = typeof part.id === "string" ? part.id : null; + if (msgID && partID) { + setMessages((current) => { + const msgIndex = current.findIndex((m) => m.id === msgID); + if (msgIndex < 0) return current; + + const existing = current[msgIndex]; + const updatedParts = [...existing.parts]; + const existingIdx = updatedParts.findIndex( + (p) => typeof p === "object" && p !== null && (p as Record).id === partID + ); + if (existingIdx >= 0) { + updatedParts[existingIdx] = part as unknown as ChatMessage["parts"][number]; + } else { + updatedParts.push(part as unknown as ChatMessage["parts"][number]); + } + const next = [...current]; + next[msgIndex] = { ...existing, parts: updatedParts }; + return next; + }); + } + } + } + + if (classification.hasPartRemove) { + const extracted = extractPartRemovalFromEvent(eventLines); + if (extracted?.messageID && extracted.partID) { + setMessages((current) => { + const msgIndex = current.findIndex((m) => m.id === extracted.messageID); + if (msgIndex < 0) return current; + + const existing = current[msgIndex]; + const isPartToRemove = (part: ChatMessage["parts"][number]) => + typeof part === "object" && + part !== null && + (part as Record).id === extracted.partID; + const updatedParts = existing.parts.filter((part) => !isPartToRemove(part)); + if (updatedParts.length === existing.parts.length) return current; + + const next = [...current]; + next[msgIndex] = { + ...existing, + parts: updatedParts, + text: updatedParts + .filter( + (part) => + typeof part === "object" && + part !== null && + (part as Record).type === "text" + ) + .map((part) => (part as Record).text || "") + .join("\n") + .trim(), + }; + return next; + }); + } + } + + // --- Message update (message.updated) — inline upsert --- + if (classification.hasMessageUpdate) { + const info = extractMessageFromEvent(eventLines); + if (info && typeof info.id === "string" && typeof info.role === "string") { + const infoId = info.id; + const infoRole = info.role as ChatMessage["role"]; + setMessages((current) => { + const existingIdx = current.findIndex((m) => m.id === infoId); + if (existingIdx >= 0) { + const freshParts = Array.isArray(info.parts) ? info.parts as ChatMessage["parts"] : undefined; + const freshText = typeof info.text === "string" ? info.text : undefined; + const next = [...current]; + next[existingIdx] = { + ...current[existingIdx], + ...(freshParts ? { parts: freshParts } : {}), + ...(freshText !== undefined ? { text: freshText } : {}), + }; + return next; + } + // New message — validate required fields before casting + const msg = { + id: infoId, + role: infoRole, + text: typeof info.text === "string" ? info.text : "", + createdAt: typeof info.time === "object" && info.time + ? new Date((info.time as Record).created as number).toISOString() + : new Date().toISOString(), + parts: Array.isArray(info.parts) ? info.parts as ChatMessage["parts"] : [], + }; + return [...current, msg]; + }); + } + } + + // --- Diff update --- + if (classification.hasDiffUpdate) { + const diffData = extractDiffFromEvent(eventLines); + if (diffData) { + setDiffEntries(diffData.diff as unknown as SessionDiffEntry[]); + } + } + + // --- Session status --- + if (classification.hasSessionUpdate) { + const statusData = extractSessionStatusFromEvent(eventLines); + if (statusData && statusData.status) { + const statusType = String((statusData.status as Record).type || ""); + if (statusType === "idle") { + void loadMessages(activeProjectId, { silent: true }); + } + } + } + + // Full refresh only for compaction or message deletion (structural changes) + const needsFullRefresh = classification.hasCompactionUpdate || classification.hasMessageDelete; if (needsFullRefresh) { scheduleStreamRefresh(activeProjectId); } @@ -3449,15 +3572,21 @@ async function loadDiff(projectId: string) { awaitingFinalReplyNotificationByChatRef.current[`${activeProjectId}:${result.sessionId}`] = true; } - setMessages((current) => [...current, result.message]); + setMessages((current) => [ + ...current.filter((m) => m.id !== result.message.id && !m.id.startsWith("local-")), + result.message, + ]); await refreshProjectsAndStatus(activeProjectId); await Promise.all([ loadProjectSessions(activeProjectId, { silent: true }), - loadMessages(activeProjectId, { sessionId: result.sessionId }), loadPendingApprovals(activeProjectId), loadPendingQuestions(activeProjectId), ]); void loadDiff(activeProjectId); + // Reconcile with server after a brief delay — don't mask send errors + setTimeout(() => { + void loadMessages(activeProjectId, { silent: true, sessionId: result.sessionId }); + }, 800); } catch (error) { setProjectError(error instanceof Error ? error.message : "Failed to send message"); setMessages((current) => current.filter((message) => message.id !== userEcho.id)); diff --git a/frontend/src/utils/streamUtils.ts b/frontend/src/utils/streamUtils.ts index 2563570..9a0d3f9 100644 --- a/frontend/src/utils/streamUtils.ts +++ b/frontend/src/utils/streamUtils.ts @@ -103,15 +103,18 @@ export function parseApprovalFromStreamData(data: string): { return { request: null, resolvedPermissionId: null }; } - const typeValue = String(payload.type || "").toLowerCase(); + // Unwrap GlobalEvent envelope + const inner = (payload.payload as Record) || payload; + + const typeValue = String(inner.type || "").toLowerCase(); if (!typeValue.includes("permission")) { return { request: null, resolvedPermissionId: null }; } const properties = - payload.properties && typeof payload.properties === "object" - ? (payload.properties as Record) - : payload; + inner.properties && typeof inner.properties === "object" + ? (inner.properties as Record) + : inner; const permissionId = findPermissionId(properties); if (!permissionId) { @@ -150,11 +153,14 @@ export function parseQuestionFromStreamData(data: string): { return { request: null, resolvedQuestionId: null }; } - const typeValue = String(payload.type || "").toLowerCase(); + // Unwrap GlobalEvent envelope + const inner = (payload.payload as Record) || payload; + + const typeValue = String(inner.type || "").toLowerCase(); const properties = - payload.properties && typeof payload.properties === "object" - ? (payload.properties as Record) - : payload; + inner.properties && typeof inner.properties === "object" + ? (inner.properties as Record) + : inner; if (typeValue === "question.asked") { const id = typeof properties.id === "string" ? properties.id : null; @@ -201,8 +207,14 @@ export type StreamEventClassification = { hasMessageUpdate: boolean; hasQuestionUpdate: boolean; hasApprovalUpdate: boolean; + hasPartDelta: boolean; hasPartUpdate: boolean; + hasPartRemove: boolean; hasCompactionUpdate: boolean; + hasSessionUpdate: boolean; + hasTodoUpdate: boolean; + hasDiffUpdate: boolean; + hasMessageDelete: boolean; isHeartbeat: boolean; rawEventType: string | null; }; @@ -212,8 +224,14 @@ export function classifyStreamEvent(data: string): StreamEventClassification { hasMessageUpdate: false, hasQuestionUpdate: false, hasApprovalUpdate: false, + hasPartDelta: false, hasPartUpdate: false, + hasPartRemove: false, hasCompactionUpdate: false, + hasSessionUpdate: false, + hasTodoUpdate: false, + hasDiffUpdate: false, + hasMessageDelete: false, isHeartbeat: false, rawEventType: null, }; @@ -233,27 +251,32 @@ export function classifyStreamEvent(data: string): StreamEventClassification { result.hasApprovalUpdate = true; } else if (eventType === "question.asked" || eventType === "question.replied" || eventType === "question.rejected") { result.hasQuestionUpdate = true; + } else if (eventType === "message.part.delta") { + result.hasPartDelta = true; } else if (eventType === "message.part.updated") { result.hasPartUpdate = true; + } else if (eventType === "message.part.removed") { + result.hasPartRemove = true; + } else if (eventType === "message.updated") { + result.hasMessageUpdate = true; + } else if (eventType === "message.removed") { + result.hasMessageDelete = true; + } else if (eventType === "todo.updated") { + result.hasTodoUpdate = true; + } else if (eventType === "session.diff") { + result.hasDiffUpdate = true; + } else if ( + eventType === "session.status" || + eventType === "session.idle" || + eventType === "session.compacted" + ) { + result.hasSessionUpdate = true; } else if ( - eventType === "session.compacted" || eventType === "session.next.compaction.started" || eventType === "session.next.compaction.delta" || eventType === "session.next.compaction.ended" ) { result.hasCompactionUpdate = true; - } else if ( - eventType === "text.delta" || - eventType === "text.ended" || - eventType === "prompted" || - eventType === "tool.called" || - eventType === "tool.result" || - eventType === "tool.error" || - eventType === "message.created" || - eventType === "message.updated" || - eventType === "message.completed" - ) { - result.hasMessageUpdate = true; } } @@ -266,18 +289,32 @@ export function classifyStreamEvent(data: string): StreamEventClassification { result.hasApprovalUpdate = true; } else if (directType.startsWith("question.")) { result.hasQuestionUpdate = true; + } else if (directType === "message.part.delta") { + result.hasPartDelta = true; + } else if (directType === "message.part.updated") { + result.hasPartUpdate = true; + } else if (directType === "message.part.removed") { + result.hasPartRemove = true; + } else if (directType === "message.updated") { + result.hasMessageUpdate = true; + } else if (directType === "message.removed") { + result.hasMessageDelete = true; + } else if (directType === "todo.updated") { + result.hasTodoUpdate = true; + } else if (directType === "session.diff") { + result.hasDiffUpdate = true; } else if ( - directType === "session.compacted" || - directType.startsWith("session.next.compaction.") + directType === "session.status" || + directType === "session.idle" || + directType === "session.compacted" ) { - result.hasCompactionUpdate = true; + result.hasSessionUpdate = true; } else if ( - directType.startsWith("text.") || - directType.startsWith("tool.") || - directType.startsWith("message.") || - directType === "prompted" + directType === "session.next.compaction.started" || + directType === "session.next.compaction.delta" || + directType === "session.next.compaction.ended" ) { - result.hasMessageUpdate = true; + result.hasCompactionUpdate = true; } } } catch { @@ -295,6 +332,143 @@ export type MessagePartPayload = { text: string | null; }; +export function extractPayloadFromDataLines(eventLines: string[]): Record | null { + const payloadLines = eventLines + .filter((line) => line.startsWith("data:")) + .map((line) => line.slice(5).trim()) + .filter(Boolean); + + if (payloadLines.length === 0) return null; + + const payload = payloadLines.join("\n"); + try { + const parsed = JSON.parse(payload) as Record; + // Unwrap GlobalEvent envelope: { directory, project?, workspace?, payload: { type, properties } } + const inner = parsed.payload as Record | undefined; + return (inner && typeof inner.type === "string") ? inner : parsed; + } catch { + return null; + } +} + +export type ExtractedPart = { + sessionID: string | null; + part: Record | null; + rawEventType: string | null; +}; + +export type ExtractedPartRemoval = { + sessionID: string | null; + messageID: string | null; + partID: string | null; + rawEventType: string | null; +}; + +export function extractPartFromEvent(eventLines: string[]): ExtractedPart | null { + const payload = extractPayloadFromDataLines(eventLines); + if (!payload) return null; + + const properties = payload.properties as Record | undefined; + if (!properties) return null; + + const sessionID = typeof properties.sessionID === "string" ? properties.sessionID : null; + const msgType = typeof payload.type === "string" ? payload.type : null; + + const part = properties.part as Record | undefined; + if (part) { + return { sessionID, part, rawEventType: msgType }; + } + + if (msgType === "message.part.delta") { + const deltaPart: Record = { + id: properties.partID, + messageID: properties.messageID, + type: properties.field || "text", + text: properties.delta, + }; + return { sessionID, part: deltaPart, rawEventType: msgType }; + } + + return null; +} + +export function extractPartRemovalFromEvent(eventLines: string[]): ExtractedPartRemoval | null { + const payload = extractPayloadFromDataLines(eventLines); + if (!payload || payload.type !== "message.part.removed") return null; + + const properties = payload.properties as Record | undefined; + if (!properties) return null; + + const sessionID = typeof properties.sessionID === "string" ? properties.sessionID : null; + const part = properties.part as Record | undefined; + const messageID = + typeof properties.messageID === "string" + ? properties.messageID + : part && typeof part.messageID === "string" + ? part.messageID + : null; + const partID = + typeof properties.partID === "string" + ? properties.partID + : part && typeof part.id === "string" + ? part.id + : null; + const rawEventType = typeof payload.type === "string" ? payload.type : null; + + return { + sessionID, + messageID, + partID, + rawEventType, + }; +} + +export function extractMessageFromEvent(eventLines: string[]): Record | null { + const payload = extractPayloadFromDataLines(eventLines); + if (!payload) return null; + + const properties = payload.properties as Record | undefined; + if (!properties) return null; + + return (properties.info || properties) as Record | null; +} + +export function extractTodosFromEvent(eventLines: string[]): { sessionID: string | null; todos: unknown[] } | null { + const payload = extractPayloadFromDataLines(eventLines); + if (!payload) return null; + + const properties = payload.properties as Record | undefined; + if (!properties) return null; + + const sessionID = typeof properties.sessionID === "string" ? properties.sessionID : null; + const todos = Array.isArray(properties.todos) ? properties.todos : []; + return { sessionID, todos }; +} + +export function extractDiffFromEvent(eventLines: string[]): { sessionID: string | null; diff: unknown[] } | null { + const payload = extractPayloadFromDataLines(eventLines); + if (!payload) return null; + + const properties = payload.properties as Record | undefined; + if (!properties) return null; + + const sessionID = typeof properties.sessionID === "string" ? properties.sessionID : null; + const diff = Array.isArray(properties.diff) ? properties.diff : []; + return { sessionID, diff }; +} + +export function extractSessionStatusFromEvent(eventLines: string[]): { sessionID: string | null; status: Record | null } | null { + const payload = extractPayloadFromDataLines(eventLines); + if (!payload) return null; + + const properties = payload.properties as Record | undefined; + if (!properties) return null; + + const sessionID = typeof properties.sessionID === "string" ? properties.sessionID : null; + const status = properties.status as Record | undefined; + return { sessionID, status: status || null }; +} + export function extractMessagePartText(eventLines: string[]): MessagePartPayload | null { const payloadLines = eventLines .filter((line) => line.startsWith("data:")) @@ -305,21 +479,49 @@ export function extractMessagePartText(eventLines: string[]): MessagePartPayload return null; } - const payload = payloadLines.join("\n"); + const raw = payloadLines.join("\n"); try { - const parsed = JSON.parse(payload) as Record; - const part = parsed.part && typeof parsed.part === "object" ? parsed.part as Record : null; - if (!part) { - return null; + const parsed = JSON.parse(raw) as Record; + + // Unwrap GlobalEvent envelope + const wrapper = (parsed.payload as Record) || parsed; + + // Try the properties.part path (message.part.updated) + const part = wrapper.part as Record | undefined; + if (part) { + const sessionID = typeof wrapper.sessionID === "string" ? wrapper.sessionID : null; + const messageID = typeof part.messageID === "string" ? part.messageID : null; + const partID = typeof part.id === "string" ? part.id : null; + const partType = typeof part.type === "string" ? part.type : null; + const text = typeof part.text === "string" ? part.text : null; + return { sessionID, messageID, partID, partType, text }; } - const sessionID = typeof parsed.sessionID === "string" ? parsed.sessionID : null; - const messageID = typeof part.messageID === "string" ? part.messageID : null; - const partID = typeof part.id === "string" ? part.id : null; - const partType = typeof part.type === "string" ? part.type : null; - const text = typeof part.text === "string" ? part.text : null; + // Try the nested properties path (common in events) + const properties = wrapper.properties as Record | undefined; + if (properties) { + const msgType = typeof wrapper.type === "string" ? wrapper.type : null; + if (msgType === "message.part.delta") { + const sessionID = typeof properties.sessionID === "string" ? properties.sessionID : null; + const messageID = typeof properties.messageID === "string" ? properties.messageID : null; + const partID = typeof properties.partID === "string" ? properties.partID : null; + const field = typeof properties.field === "string" ? properties.field : null; + const delta = typeof properties.delta === "string" ? properties.delta : null; + return { sessionID, messageID, partID, partType: field, text: delta }; + } - return { sessionID, messageID, partID, partType, text }; + const partObj = properties.part as Record | undefined; + if (partObj) { + const sessionID = typeof properties.sessionID === "string" ? properties.sessionID : null; + const messageID = typeof partObj.messageID === "string" ? partObj.messageID : null; + const partID = typeof partObj.id === "string" ? partObj.id : null; + const partType = typeof partObj.type === "string" ? partObj.type : null; + const text = typeof partObj.text === "string" ? partObj.text : null; + return { sessionID, messageID, partID, partType, text }; + } + } + + return null; } catch { return null; } diff --git a/frontend/vite.config.ts b/frontend/vite.config.ts index c5dabb0..94bd8ff 100644 --- a/frontend/vite.config.ts +++ b/frontend/vite.config.ts @@ -18,6 +18,26 @@ export default defineConfig({ "/api": { target: `http://localhost:${backendPort}`, changeOrigin: true, + proxyTimeout: 30000, + timeout: 30000, + configure: (proxy) => { + proxy.on("proxyReq", (proxyReq, req) => { + if (req.url?.includes("/stream")) { + proxyReq.setHeader("Connection", "keep-alive"); + } + }); + proxy.on("proxyRes", (proxyRes, req) => { + if (req.url?.includes("/stream")) { + delete proxyRes.headers["content-length"]; + } + }); + proxy.on("error", (err, _req, res) => { + if (res && !res.headersSent) { + res.writeHead(502, { "Content-Type": "application/json" }); + res.end(JSON.stringify({ error: `Backend unavailable: ${err.message}` })); + } + }); + }, }, }, }, diff --git a/scripts/install-autostart-ubuntu.sh b/scripts/install-autostart-ubuntu.sh index b76a205..5063042 100755 --- a/scripts/install-autostart-ubuntu.sh +++ b/scripts/install-autostart-ubuntu.sh @@ -79,6 +79,7 @@ resolve_opencode_bin() { fi for candidate in \ + "$APP_HOME/.opencode/bin/opencode" \ "$APP_HOME/.local/bin/opencode" \ "/usr/local/bin/opencode" \ "/usr/bin/opencode"; do diff --git a/scripts/run-backend-service.sh b/scripts/run-backend-service.sh index cc107e4..0f74ef4 100755 --- a/scripts/run-backend-service.sh +++ b/scripts/run-backend-service.sh @@ -46,4 +46,5 @@ fi exec env \ OPENCODE_BASE_URL="$OPENCODE_BASE_URL" \ BACKEND_PORT="$BACKEND_PORT" \ + FRONTEND_ORIGINS="${FRONTEND_ORIGINS:-http://localhost:5173}" \ "$PYTHON_BIN" "$ROOT_DIR/backend/run.py" diff --git a/scripts/start-app.sh b/scripts/start-app.sh index fa9a507..e737201 100755 --- a/scripts/start-app.sh +++ b/scripts/start-app.sh @@ -22,19 +22,19 @@ is_running() { if [[ ! -f "$pid_file" ]]; then return 1 fi - local pid pid="$(cat "$pid_file")" - if [[ -z "$pid" ]]; then - return 1 - fi - + [[ -n "$pid" ]] || return 1 kill -0 "$pid" >/dev/null 2>&1 } +systemd_active() { + local unit="$1" + systemctl is-active --quiet "$unit" 2>/dev/null +} + require_dep() { - local dep="$1" - local hint="$2" + local dep="$1" hint="$2" if ! command -v "$dep" >/dev/null 2>&1; then echo "Missing dependency: $dep" echo "$hint" @@ -57,24 +57,32 @@ if [[ ! -d "$ROOT_DIR/frontend/node_modules" ]]; then npm --prefix "$ROOT_DIR/frontend" install fi -if is_running "$OPENCODE_PID_FILE" || is_running "$BACKEND_PID_FILE" || is_running "$FRONTEND_PID_FILE"; then - echo "One or more services are already running. Use scripts/stop-app.sh first." +BACKEND_SYSTEMD_UNIT="mobile-opencode-control-backend.service" +FRONTEND_SYSTEMD_UNIT="mobile-opencode-control-frontend.service" +OPENCODE_SYSTEMD_UNIT="mobile-opencode-control-opencode.service" + +BACKEND_USE_SYSTEMD=false +FRONTEND_USE_SYSTEMD=false +OPENCODE_USE_SYSTEMD=false + +systemd_active "$BACKEND_SYSTEMD_UNIT" && BACKEND_USE_SYSTEMD=true +systemd_active "$FRONTEND_SYSTEMD_UNIT" && FRONTEND_USE_SYSTEMD=true +systemd_active "$OPENCODE_SYSTEMD_UNIT" && OPENCODE_USE_SYSTEMD=true + +if is_running "$BACKEND_PID_FILE" || is_running "$FRONTEND_PID_FILE" || is_running "$OPENCODE_PID_FILE"; then + echo "One or more services are already running (manual scripts)." + echo "Use scripts/stop-app.sh first, or check systemd services." exit 1 fi +# ─── port picking helper ─── pick_port() { - local base_port="$1" - local attempts="${2:-60}" - local sleep_seconds="${4:-0.5}" + local base_port="$1" attempts="${2:-60}" sleep_seconds="${3:-0.5}" "$ROOT_DIR/.venv/bin/python" - "$base_port" "$attempts" "$sleep_seconds" <<'PY' -import socket -import time -import sys - +import socket, time, sys base_port = int(sys.argv[1]) attempts = int(sys.argv[2]) sleep_seconds = float(sys.argv[3]) - for attempt in range(attempts): for offset in range(3): candidate = base_port + offset @@ -82,22 +90,18 @@ for attempt in range(attempts): sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) try: sock.bind(("127.0.0.1", candidate)) - except OSError as exc: + except OSError: pass else: print(candidate) sock.close() sys.exit(0) finally: - try: - sock.close() - except Exception: - pass - + try: sock.close() + except Exception: pass if attempt < attempts - 1: time.sleep(sleep_seconds) - -print(f"No free port available in range {base_port}-{base_port + 2}", file=sys.stderr) +print(f"No free port in range {base_port}-{base_port + 2}", file=sys.stderr) sys.exit(1) PY } @@ -105,55 +109,79 @@ PY wait_for_opencode() { local port="$1" local url="http://127.0.0.1:${port}/global/health" - local max_tries=40 - local attempt=1 + local max_tries=40 attempt=1 while (( attempt <= max_tries )); do - if curl -fsS "$url" >/dev/null 2>&1; then - return 0 - fi + curl -fsS "$url" >/dev/null 2>&1 && return 0 sleep 0.25 (( attempt += 1 )) done return 1 } -OPENCODE_BASE_PORT="${OPENCODE_APP_PORT:-40961}" -OPENCODE_PORT="$(pick_port "$OPENCODE_BASE_PORT")" -OPENCODE_BASE_URL="http://127.0.0.1:${OPENCODE_PORT}" - -BACKEND_BASE_PORT="${BACKEND_APP_PORT:-38473}" -BACKEND_PORT="$(pick_port "$BACKEND_BASE_PORT")" -BACKEND_BASE_URL="http://127.0.0.1:${BACKEND_PORT}" - -FRONTEND_BASE_PORT="${FRONTEND_APP_PORT:-5173}" -FRONTEND_PORT="$(pick_port "$FRONTEND_BASE_PORT")" -FRONTEND_BASE_URL="http://127.0.0.1:${FRONTEND_PORT}" +# ─── 1. Resolve frontend port first (needed by OpenCode CORS and backend) ─── +if $FRONTEND_USE_SYSTEMD; then + FRONTEND_PORT="$(cat "$FRONTEND_PORT_FILE" 2>/dev/null || echo "${FRONTEND_APP_PORT:-5173}")" +else + FRONTEND_PORT="$(pick_port "${FRONTEND_APP_PORT:-5173}")" +fi FRONTEND_ALLOWED_HOSTS="${FRONTEND_ALLOWED_HOSTS:-localhost,127.0.0.1}" FRONTEND_ORIGINS="${FRONTEND_ORIGINS:-http://localhost:${FRONTEND_PORT}}" +FRONTEND_BASE_URL="http://localhost:${FRONTEND_PORT}" OPENCODE_CORS_ORIGINS="${OPENCODE_CORS_ORIGINS:-$FRONTEND_ORIGINS}" -echo "Starting OpenCode server..." -opencode serve --hostname 127.0.0.1 --port "$OPENCODE_PORT" --cors "$OPENCODE_CORS_ORIGINS" >"$LOG_DIR/opencode.log" 2>&1 & -echo $! >"$OPENCODE_PID_FILE" -echo "$OPENCODE_PORT" >"$OPENCODE_PORT_FILE" -echo "$OPENCODE_BASE_URL" >"$OPENCODE_URL_FILE" +# ─── 2. OpenCode ─── +if $OPENCODE_USE_SYSTEMD; then + echo "OpenCode is managed by systemd ($OPENCODE_SYSTEMD_UNIT)." + OPENCODE_PORT="$(cat "$OPENCODE_PORT_FILE" 2>/dev/null || echo "${OPENCODE_APP_PORT:-40961}")" +else + OPENCODE_PORT="$(pick_port "${OPENCODE_APP_PORT:-40961}")" + echo "Starting OpenCode server..." + opencode serve \ + --hostname 127.0.0.1 \ + --port "$OPENCODE_PORT" \ + --cors "$OPENCODE_CORS_ORIGINS" \ + >"$LOG_DIR/opencode.log" 2>&1 & + echo $! >"$OPENCODE_PID_FILE" + echo "$OPENCODE_PORT" >"$OPENCODE_PORT_FILE" + echo "http://127.0.0.1:${OPENCODE_PORT}" >"$OPENCODE_URL_FILE" + + if ! wait_for_opencode "$OPENCODE_PORT"; then + echo "OpenCode failed to start on port ${OPENCODE_PORT}. Check: $LOG_DIR/opencode.log" + exit 1 + fi +fi +OPENCODE_BASE_URL="http://127.0.0.1:${OPENCODE_PORT}" -if ! wait_for_opencode "$OPENCODE_PORT"; then - echo "OpenCode failed to start on ${OPENCODE_BASE_URL}. Check: $LOG_DIR/opencode.log" - exit 1 +# ─── 3. Backend ─── +if $BACKEND_USE_SYSTEMD; then + echo "Backend is managed by systemd ($BACKEND_SYSTEMD_UNIT)." + BACKEND_PORT="$(cat "$BACKEND_PORT_FILE" 2>/dev/null || echo "${BACKEND_APP_PORT:-38473}")" +else + BACKEND_PORT="$(pick_port "${BACKEND_APP_PORT:-38473}")" + echo "Starting backend..." + OPENCODE_BASE_URL="$OPENCODE_BASE_URL" \ + FRONTEND_ORIGINS="$FRONTEND_ORIGINS" \ + FRONTEND_ORIGIN="${FRONTEND_ORIGIN:-$(printf '%s' "$FRONTEND_ORIGINS" | cut -d',' -f1)}" \ + BACKEND_PORT="$BACKEND_PORT" \ + "$ROOT_DIR/.venv/bin/python" "$ROOT_DIR/backend/run.py" \ + >"$LOG_DIR/backend.log" 2>&1 & + echo $! >"$BACKEND_PID_FILE" + echo "$BACKEND_PORT" >"$BACKEND_PORT_FILE" fi +BACKEND_BASE_URL="http://127.0.0.1:${BACKEND_PORT}" -echo "Starting backend with project venv..." -OPENCODE_BASE_URL="$OPENCODE_BASE_URL" FRONTEND_ORIGINS="$FRONTEND_ORIGINS" FRONTEND_ORIGIN="${FRONTEND_ORIGIN:-$(printf '%s' "$FRONTEND_ORIGINS" | cut -d',' -f1)}" BACKEND_PORT="$BACKEND_PORT" "$ROOT_DIR/.venv/bin/python" "$ROOT_DIR/backend/run.py" >"$LOG_DIR/backend.log" 2>&1 & -echo $! >"$BACKEND_PID_FILE" -echo "$BACKEND_PORT" >"$BACKEND_PORT_FILE" -echo "$BACKEND_BASE_URL" >"$BACKEND_URL_FILE" - -echo "Starting frontend dev server..." -BACKEND_PORT="$BACKEND_PORT" FRONTEND_ALLOWED_HOSTS="$FRONTEND_ALLOWED_HOSTS" npm --prefix "$ROOT_DIR/frontend" run dev -- --host 0.0.0.0 --port "$FRONTEND_PORT" >"$LOG_DIR/frontend.log" 2>&1 & -echo $! >"$FRONTEND_PID_FILE" -echo "$FRONTEND_PORT" >"$FRONTEND_PORT_FILE" -echo "$FRONTEND_BASE_URL" >"$FRONTEND_URL_FILE" +# ─── 4. Frontend ─── +if $FRONTEND_USE_SYSTEMD; then + echo "Frontend is managed by systemd ($FRONTEND_SYSTEMD_UNIT)." +else + echo "Starting frontend dev server..." + BACKEND_PORT="$BACKEND_PORT" \ + FRONTEND_ALLOWED_HOSTS="$FRONTEND_ALLOWED_HOSTS" \ + npm --prefix "$ROOT_DIR/frontend" run dev -- --host 0.0.0.0 --port "$FRONTEND_PORT" \ + >"$LOG_DIR/frontend.log" 2>&1 & + echo $! >"$FRONTEND_PID_FILE" + echo "$FRONTEND_PORT" >"$FRONTEND_PORT_FILE" +fi sleep 2 diff --git a/scripts/stop-app.sh b/scripts/stop-app.sh index 09bc2e5..73305c7 100755 --- a/scripts/stop-app.sh +++ b/scripts/stop-app.sh @@ -14,87 +14,97 @@ FRONTEND_PID_FILE="$RUNTIME_DIR/frontend.pid" FRONTEND_PORT_FILE="$RUNTIME_DIR/frontend.port" FRONTEND_URL_FILE="$RUNTIME_DIR/frontend.url" -stop_matching_processes() { - local name="$1" - local pattern="$2" +BACKEND_SYSTEMD_UNIT="mobile-opencode-control-backend.service" +FRONTEND_SYSTEMD_UNIT="mobile-opencode-control-frontend.service" +OPENCODE_SYSTEMD_UNIT="mobile-opencode-control-opencode.service" +systemd_active() { + local unit="$1" + systemctl is-active --quiet "$unit" 2>/dev/null +} + +stop_by_pattern() { + local name="$1" pattern="$2" local pids - pids="$(pgrep -f "$pattern" || true)" + pids="$(pgrep -f "$pattern" 2>/dev/null || true)" if [[ -z "$pids" ]]; then return 1 fi - - while IFS= read -r pid; do + echo "$pids" | while read -r pid; do [[ -z "$pid" ]] && continue - kill "$pid" >/dev/null 2>&1 || true - done <<< "$pids" - + kill "$pid" 2>/dev/null || true + done sleep 1 - - while IFS= read -r pid; do + echo "$pids" | while read -r pid; do [[ -z "$pid" ]] && continue - if kill -0 "$pid" >/dev/null 2>&1; then - kill -9 "$pid" >/dev/null 2>&1 || true - fi - done <<< "$pids" - + kill -0 "$pid" 2>/dev/null && kill -9 "$pid" 2>/dev/null || true + done return 0 } -stop_from_pid_file() { - local name="$1" - local pid_file="$2" - - if [[ ! -f "$pid_file" ]]; then - return 1 +stop_by_port() { + local name="$1" port_file="$2" default_port="$3" + local port + if [[ -f "$port_file" ]]; then + port="$(cat "$port_file")" + else + port="$default_port" fi - + # kill only the process listening on the specific port local pid - pid="$(cat "$pid_file")" - if [[ -z "$pid" ]]; then - rm -f "$pid_file" - return 1 - fi - - if ! kill -0 "$pid" >/dev/null 2>&1; then - rm -f "$pid_file" - return 1 + pid="$(ss -tlnp "sport = :${port}" 2>/dev/null | grep -oP 'pid=\K[0-9]+' | head -1)" + if [[ -n "$pid" ]]; then + kill "$pid" 2>/dev/null || true + sleep 0.5 + kill -0 "$pid" 2>/dev/null && kill -9 "$pid" 2>/dev/null || true fi +} - kill "$pid" >/dev/null 2>&1 || true - sleep 1 +stop_service() { + local name="$1" pid_file="$2" pattern="$3" port_file="$4" default_port="$5" systemd_unit="$6" + local stopped=false - if kill -0 "$pid" >/dev/null 2>&1; then - kill -9 "$pid" >/dev/null 2>&1 || true + # If systemd manages this service, skip — user should use systemctl + if systemd_active "$systemd_unit"; then + echo "$name: managed by systemd ($systemd_unit) — use 'sudo systemctl stop $systemd_unit'" + return 0 fi - rm -f "$pid_file" - return 0 -} - -stop_service() { - local name="$1" - local pid_file="$2" - local pattern="$3" - local stopped=1 + # Try PID file first + if [[ -f "$pid_file" ]]; then + local pid + pid="$(cat "$pid_file" 2>/dev/null || echo "")" + if [[ -n "$pid" ]]; then + if kill -0 "$pid" 2>/dev/null; then + kill "$pid" 2>/dev/null || true + sleep 1 + kill -0 "$pid" 2>/dev/null && kill -9 "$pid" 2>/dev/null || true + stopped=true + fi + fi + rm -f "$pid_file" + fi - if stop_from_pid_file "$name" "$pid_file"; then - stopped=0 + # Fallback: kill by pattern + if stop_by_pattern "$name" "$pattern"; then + stopped=true fi - if stop_matching_processes "$name" "$pattern"; then - stopped=0 + # Fallback: kill by port + if ! $stopped; then + stop_by_port "$name" "$port_file" "$default_port" + stopped=true fi - if [[ "$stopped" -eq 0 ]]; then + if $stopped; then echo "$name: stopped" else echo "$name: not running" fi } -stop_service "frontend" "$FRONTEND_PID_FILE" "$ROOT_DIR/frontend/node_modules/.bin/vite --host 0.0.0.0 --port" -stop_service "backend" "$BACKEND_PID_FILE" "$ROOT_DIR/backend/run.py" -stop_service "opencode" "$OPENCODE_PID_FILE" "opencode serve --hostname 127.0.0.1 --port 4096" +stop_service "frontend" "$FRONTEND_PID_FILE" "$ROOT_DIR/frontend/node_modules/.bin/vite" "$FRONTEND_PORT_FILE" 5173 "$FRONTEND_SYSTEMD_UNIT" +stop_service "backend" "$BACKEND_PID_FILE" "$ROOT_DIR/backend/run.py" "$BACKEND_PORT_FILE" 38473 "$BACKEND_SYSTEMD_UNIT" +stop_service "opencode" "$OPENCODE_PID_FILE" "opencode serve" "$OPENCODE_PORT_FILE" 40961 "$OPENCODE_SYSTEMD_UNIT" rm -f "$OPENCODE_PORT_FILE" "$OPENCODE_URL_FILE" "$BACKEND_PORT_FILE" "$BACKEND_URL_FILE" "$FRONTEND_PORT_FILE" "$FRONTEND_URL_FILE"