diff --git a/android/src/screens/ChatScreen.tsx b/android/src/screens/ChatScreen.tsx index 158c8e8..89dc5ef 100644 --- a/android/src/screens/ChatScreen.tsx +++ b/android/src/screens/ChatScreen.tsx @@ -301,6 +301,9 @@ const ChatScreen: React.FC = () => { const [searchIndex, setSearchIndex] = useState(0); // welcher Treffer aktiv ist const [pendingAttachments, setPendingAttachments] = useState<{file: any, isPhoto: boolean}[]>([]); const [agentActivity, setAgentActivity] = useState<{activity: string, tool: string}>({activity: 'idle', tool: ''}); + // Multi-Threading: Activity pro Kontext (key = projectId, '' = Hauptchat). + // Der Indikator zeigt nur den fokussierten Kontext — nicht global. + const [agentActivityByCtx, setAgentActivityByCtx] = useState>({}); // Gedanken-Stream: chronologisches Log dessen was ARIA intern macht. // Wird aus agent_activity-Events gefuettert und in AsyncStorage persistiert. const [thoughts, setThoughts] = useState([]); @@ -1209,7 +1212,11 @@ const ChatScreen: React.FC = () => { if (message.type === 'agent_activity') { const activity = (message.payload.activity as string) || 'idle'; const tool = (message.payload.tool as string) || ''; + const actPid = ((message.payload as any).projectId as string) || ''; + // Global (fuer die bestehende ACK-/Watchdog-Logik) UND per-Kontext + // (fuer den fokussierten Indikator) fuehren. setAgentActivity({ activity, tool }); + setAgentActivityByCtx(prev => ({ ...prev, [actPid]: { activity, tool } })); // Implizite ACK-Bestaetigung: Brain hat angefangen zu arbeiten → // unsere Nachricht ist offensichtlich angekommen, auch wenn das // chat_ack aus irgendeinem Grund nicht durchkam. Alle laufenden @@ -1930,11 +1937,13 @@ const ChatScreen: React.FC = () => { }); }, [inputText, getCurrentLocation, pendingAttachments, sendPendingAttachments, interruptAriaIfBusy, dispatchWithAck]); - // Anfrage abbrechen — sofort lokalen Indicator weg, Bridge triggert doctor --fix + // Anfrage abbrechen — nur den fokussierten Kontext (kontext-scoped Cancel). const cancelRequest = useCallback(() => { + const pid = focusedProjectIdRef.current || ''; setAgentActivity({ activity: 'idle', tool: '' }); + setAgentActivityByCtx(prev => ({ ...prev, [pid]: { activity: 'idle', tool: '' } })); clearStuckWatchdog(); - rvs.send('cancel_request' as any, {}); + rvs.send('cancel_request' as any, { projectId: pid }); }, []); // Barge-In: wenn der User waehrend ARIA arbeitet/spricht eine neue Sprach- @@ -2759,13 +2768,18 @@ const ChatScreen: React.FC = () => { } /> - {/* Thinking-Indicator */} - {agentActivity.activity !== 'idle' && ( + {/* Thinking-Indicator \u2014 NUR fuer den fokussierten Kontext (Multi-Threading). + ARIA kann in anderen Kontexten parallel arbeiten, ohne dass hier ein + Indikator flackert der nicht zum sichtbaren Chat gehoert. */} + {(() => { + const focusAct = agentActivityByCtx[focusedProjectId] || { activity: 'idle', tool: '' }; + if (focusAct.activity === 'idle') return null; + return ( - {agentActivity.activity === 'tool' && agentActivity.tool - ? `\uD83D\uDD27 ${agentActivity.tool}` - : agentActivity.activity === 'assistant' + {focusAct.activity === 'tool' && focusAct.tool + ? `\uD83D\uDD27 ${focusAct.tool}` + : focusAct.activity === 'assistant' ? '\u270D\uFE0F ARIA schreibt...' : '\uD83D\uDCAD ARIA denkt...'} @@ -2775,7 +2789,8 @@ const ChatScreen: React.FC = () => { - )} + ); + })()} {/* Pending Anhaenge Vorschau */} {pendingAttachments.length > 0 && ( diff --git a/aria-brain/agent.py b/aria-brain/agent.py index 61944ff..a69dbf9 100644 --- a/aria-brain/agent.py +++ b/aria-brain/agent.py @@ -1210,7 +1210,8 @@ class Agent: final_reply = "" try: for iteration in range(self.MAX_TOOL_ITERATIONS): - result = self.proxy.chat_full(messages, tools=tools) + result = self.proxy.chat_full(messages, tools=tools, + project_id=active_project_id) if result.tool_calls: # Assistant-Turn mit tool_calls in messages anhaengen (nicht in Conversation!) messages.append(ProxyMessage( diff --git a/aria-brain/proxy_client.py b/aria-brain/proxy_client.py index d5c690d..c3c9942 100644 --- a/aria-brain/proxy_client.py +++ b/aria-brain/proxy_client.py @@ -94,6 +94,7 @@ class ProxyClient: messages: List[Message], tools: Optional[list] = None, model: Optional[str] = None, + project_id: str = "", ) -> ProxyResult: """Full chat — kann Tool-Calls liefern (wenn tools mitgegeben). @@ -108,6 +109,11 @@ class ProxyClient: } if tools: payload["tools"] = tools + # Projekt-Kontext an den Proxy: routes.js taggt damit die agent_activity- + # /agent_stream-Hooks und trackt den Subprocess pro Kontext (fuer + # kontext-scoped Cancel). Leer = Hauptchat. + if project_id: + payload["aria_project_id"] = project_id logger.info("Proxy → %s (%d Messages, %d tools, model=%s)", url, len(messages), len(tools or []), payload["model"]) try: diff --git a/bridge/aria_bridge.py b/bridge/aria_bridge.py index 37b7b5f..e21fc1f 100644 --- a/bridge/aria_bridge.py +++ b/bridge/aria_bridge.py @@ -1357,7 +1357,7 @@ class ARIABridge: # _last_chat_final_at bewusst NICHT setzen: die 3s-Cooldown war fuer # trailing OpenClaw-Activity-Events; bei Voice-Chat wuerde sie die # naechste thinking-Welle unterdruecken. - await self._emit_activity("idle", "") + await self._emit_activity("idle", "", project_id=turn_pid) # ── Mode Persistence (global, nicht pro Geraet) ────── _MODE_FILE = "/shared/config/mode.json" @@ -1572,7 +1572,7 @@ class ARIABridge: # agent_activity → thinking. _emit_activity statt direktem _send_to_rvs # damit der State-Cache fuer die spaetere idle-Dedup richtig steht. - await self._emit_activity("thinking", "") + await self._emit_activity("thinking", "", project_id=project_id) def _do_call(): try: @@ -1591,7 +1591,7 @@ class ARIABridge: status, body = await asyncio.get_event_loop().run_in_executor(None, _do_call) if status != 200: logger.error("[brain] /chat fehlgeschlagen: status=%s body=%s", status, body[:200]) - await self._emit_activity("idle", "") + await self._emit_activity("idle", "", project_id=project_id) await self._send_to_rvs({ "type": "chat", "payload": { @@ -1606,13 +1606,13 @@ class ARIABridge: data = json.loads(body) except Exception: logger.error("[brain] /chat lieferte ungueltiges JSON: %s", body[:200]) - await self._emit_activity("idle", "") + await self._emit_activity("idle", "", project_id=project_id) return reply = (data.get("reply") or "").strip() if not reply: logger.warning("[brain] /chat: leerer Reply") - await self._emit_activity("idle", "") + await self._emit_activity("idle", "", project_id=project_id) return # Projekt-Kontext des Turns — wird an _process_core_response weiter- @@ -1688,7 +1688,7 @@ class ARIABridge: await self._process_core_response(reply, {"projectId": turn_project_id}) except Exception: logger.exception("[brain] _process_core_response Fehler") - await self._emit_activity("idle", "") + await self._emit_activity("idle", "", project_id=project_id) # Originaler Fallback-Send (toter Code, _emit_activity uebernimmt jetzt) await self._send_to_rvs({ "type": "agent_activity", @@ -2008,10 +2008,14 @@ class ARIABridge: logger.warning("[rvs] NOT-AUS — hard cancel: Diagnostic /api/cancel + Proxy /cancel-all") await self._cancel_via_diagnostic() await self._cancel_proxy_subprocesses() + await self._emit_activity("idle", "") else: - logger.info("[rvs] Cancel-Request von App — rufe Diagnostic /api/cancel auf") - await self._cancel_via_diagnostic() - await self._emit_activity("idle", "") + # Barge-In: nur den fokussierten Kontext abbrechen (projectId von + # der App), damit parallele Arbeit in anderen Kontexten weiterlaeuft. + cancel_pid = str(payload.get("projectId") or "") + logger.info("[rvs] Cancel-Request (kontext-scoped) project=%s", cancel_pid or "(main)") + await self._cancel_proxy_for_project(cancel_pid) + await self._emit_activity("idle", "", project_id=cancel_pid) return elif msg_type == "audio_pcm": @@ -3518,7 +3522,30 @@ class ARIABridge: status, body = await asyncio.get_event_loop().run_in_executor(None, _do_request) logger.warning("[NOT-AUS] proxy /cancel-all: %s %s", status, body) - async def _emit_activity(self, activity: str, tool: str = "", force: bool = False) -> None: + async def _cancel_proxy_for_project(self, project_id: str) -> None: + """Kontext-scoped Barge-In: killt NUR die Subprozesse EINES Kontexts + (leer = Hauptchat) ueber den proxy-internen /cancel-Endpoint. So bricht + eine Nachricht in Kontext A nicht die parallele Arbeit in Kontext B ab.""" + url = os.environ.get("PROXY_INTERNAL_URL", "http://aria-proxy:3457") + "/cancel" + data = json.dumps({"projectId": project_id or ""}).encode("utf-8") + + def _do_request(): + try: + req = urllib.request.Request( + url, method="POST", data=data, + headers={"Content-Type": "application/json"}, + ) + with urllib.request.urlopen(req, timeout=3) as resp: + return resp.status, resp.read().decode("utf-8", "ignore")[:200] + except Exception as e: + return f"error: {e}", "" + + status, body = await asyncio.get_event_loop().run_in_executor(None, _do_request) + logger.info("[cancel] proxy /cancel project=%s: %s %s", + project_id or "(main)", status, body) + + async def _emit_activity(self, activity: str, tool: str = "", force: bool = False, + project_id: str = "") -> None: """Sendet agent_activity an die App — nur wenn sich der State geaendert hat. Trailing Agent-Events nach chat:final werden 3s lang unterdrueckt @@ -3527,18 +3554,23 @@ class ARIABridge: force=True: kein State-Dedup — wird vom Proxy-Tool-Hook genutzt damit auch wiederholte gleiche Tool-Aufrufe (z.B. 3x Bash hintereinander) im Gedanken-Stream als eigene Eintraege sichtbar - bleiben.""" + bleiben. + + project_id: welcher Kontext arbeitet (leer = Hauptchat). App/Diagnostic + zeigen den Indikator damit pro Kontext statt global (Multi-Threading).""" if activity != "idle" and self._last_chat_final_at > 0: since_final = asyncio.get_event_loop().time() - self._last_chat_final_at if since_final < 3.0: return - state = (activity, tool) + # Dedup schliesst project_id ein — sonst wuerde ein Kontext-Wechsel bei + # gleichem (activity, tool) verschluckt. + state = (activity, tool, project_id) if not force and state == self._last_activity_state: return self._last_activity_state = state await self._send_to_rvs({ "type": "agent_activity", - "payload": {"activity": activity, "tool": tool}, + "payload": {"activity": activity, "tool": tool, "projectId": project_id or ""}, "timestamp": int(asyncio.get_event_loop().time() * 1000), }) @@ -3695,9 +3727,11 @@ class ARIABridge: if not tool: await _send_response(writer, 400, {"error": "tool erforderlich"}) return + tool_pid = str(data.get("projectId") or "") # Force-emit (kein Dedup): User soll JEDEN Tool-Call sehen # selbst wenn derselbe Name zweimal in Folge kommt. - asyncio.create_task(self._emit_activity("tool", tool, force=True)) + asyncio.create_task(self._emit_activity("tool", tool, force=True, + project_id=tool_pid)) await _send_response(writer, 200, {"ok": True}) elif method == "POST" and path == "/internal/agent-stream": # Vom Proxy gefeuert: voller Live-Stream der Claude-Code- diff --git a/proxy-patches/routes.js b/proxy-patches/routes.js index 2542188..b954351 100644 --- a/proxy-patches/routes.js +++ b/proxy-patches/routes.js @@ -70,9 +70,9 @@ function _postJson(url, body) { /** * Pusht einen Tool-Use-Event an die Bridge (alter Gedanken-Stream-Pfad). */ -function _emitToolEvent(toolName) { +function _emitToolEvent(toolName, projectId) { if (!toolName) return; - _postJson(TOOL_HOOK_URL, { tool: String(toolName) }); + _postJson(TOOL_HOOK_URL, { tool: String(toolName), projectId: projectId || "" }); } /** @@ -92,9 +92,11 @@ function _truncate(str, max) { // ── Subprocess-Tracking fuer Not-Aus ────────────────────────── // requestId → ClaudeSubprocess. Eintraege werden beim close/result-Event // wieder entfernt. /v1/cancel-all iteriert und ruft .kill() auf jeden. +// Wert: { subprocess, projectId }. projectId erlaubt kontext-scoped Cancel +// (nur die Subprozesse EINES Projekts killen statt aller). const _activeSubprocesses = new Map(); -function _trackSubprocess(requestId, subprocess) { - _activeSubprocesses.set(requestId, subprocess); +function _trackSubprocess(requestId, subprocess, projectId) { + _activeSubprocesses.set(requestId, { subprocess, projectId: projectId || "" }); const cleanup = () => _activeSubprocesses.delete(requestId); subprocess.on("close", cleanup); subprocess.on("error", cleanup); @@ -149,24 +151,25 @@ function _attachIdleWatchdog(subprocess, requestId) { * - Alt-API: nur Tool-Namen an /internal/agent-activity (Gedanken-Stream) * - Neu-API: voller Stream (text/tool_use/tool_result) an /internal/agent-stream */ -function _attachToolHook(subprocess, requestId) { +function _attachToolHook(subprocess, requestId, projectId) { subprocess.on("assistant", (message) => { try { const blocks = message?.message?.content || []; for (const b of blocks) { if (!b) continue; if (b.type === "tool_use") { - if (b.name) _emitToolEvent(b.name); + if (b.name) _emitToolEvent(b.name, projectId); const inputStr = b.input ? JSON.stringify(b.input) : ""; const inp = _truncate(inputStr, TOOL_INPUT_MAX_CHARS); _emitStreamEvent(requestId, "tool_use", { + projectId: projectId || "", id: b.id || null, name: b.name || "", input: inp.text, inputTruncatedBytes: inp.truncatedBytes, }); } else if (b.type === "text" && b.text) { - _emitStreamEvent(requestId, "text", { text: b.text }); + _emitStreamEvent(requestId, "text", { projectId: projectId || "", text: b.text }); } else if (b.type === "thinking" && b.thinking) { // Wenn das Modell Extended Thinking emittiert — selten in // Claude Code CLI, aber moeglich. Markieren wir extra. @@ -227,15 +230,18 @@ export async function handleChatCompletions(req, res) { } // Convert to CLI input format const cliInput = openaiToCli(body); + // ARIA: Projekt-Kontext (vom Brain via aria_project_id). Fuer + // kontext-getaggte Activity-/Stream-Events + kontext-scoped Cancel. + const ariaProjectId = String(body.aria_project_id || ""); const subprocess = new ClaudeSubprocess(); // ARIA-Patch: Tool-Use-Events + voller Live-Stream an die Bridge. // Plus: Subprocess fuer Not-Aus tracken (Hard-Kill via /v1/cancel-all). // Plus: Idle-Watchdog — Subprocess darf ewig laufen solange Events // kommen, wird aber gekillt nach IDLE_TIMEOUT_MS Inaktivitaet. - _attachToolHook(subprocess, requestId); - _trackSubprocess(requestId, subprocess); + _attachToolHook(subprocess, requestId, ariaProjectId); + _trackSubprocess(requestId, subprocess, ariaProjectId); _attachIdleWatchdog(subprocess, requestId); - _emitStreamEvent(requestId, "start", { model: body.model || null }); + _emitStreamEvent(requestId, "start", { model: body.model || null, projectId: ariaProjectId }); subprocess.on("result", () => _emitStreamEvent(requestId, "end", { reason: "result" })); subprocess.on("close", (code) => _emitStreamEvent(requestId, "end", { reason: "close", code })); subprocess.on("error", (err) => _emitStreamEvent(requestId, "end", { reason: "error", error: String(err?.message || err) })); @@ -497,9 +503,9 @@ const INTERNAL_HOST = "0.0.0.0"; // im aria-net erreichbar, nicht nach extern e function _cancelAll() { const ids = Array.from(_activeSubprocesses.keys()); let killed = 0; - for (const [id, subp] of _activeSubprocesses) { + for (const [id, entry] of _activeSubprocesses) { try { - subp.kill(); + entry.subprocess.kill(); killed++; } catch (e) { console.error("[aria-not-aus] kill failed for", id, e?.message); @@ -509,6 +515,27 @@ function _cancelAll() { return { killed, requestIds: ids }; } +// Kontext-scoped Cancel: killt NUR die Subprozesse eines Projekts (leer = +// Hauptchat). Fuer Barge-In in einem Kontext ohne die parallele Arbeit in +// anderen Kontexten abzuwuergen. +function _cancelByProject(projectId) { + const pid = String(projectId || ""); + const ids = []; + let killed = 0; + for (const [id, entry] of Array.from(_activeSubprocesses)) { + if (entry.projectId !== pid) continue; + ids.push(id); + try { + entry.subprocess.kill(); + killed++; + } catch (e) { + console.error("[aria-cancel] kill failed for", id, e?.message); + } + _activeSubprocesses.delete(id); + } + return { killed, requestIds: ids, projectId: pid }; +} + try { const internalServer = http.createServer((req, res) => { if (req.method === "POST" && req.url === "/cancel-all") { @@ -518,6 +545,21 @@ try { res.end(JSON.stringify({ ok: true, ...result })); return; } + if (req.method === "POST" && req.url === "/cancel") { + // Body: {projectId}. Kontext-scoped Barge-In — killt nur die + // Subprozesse dieses Kontexts (leer = Hauptchat). + let raw = ""; + req.on("data", (c) => { raw += c; if (raw.length > 4096) req.destroy(); }); + req.on("end", () => { + let projectId = ""; + try { projectId = String((JSON.parse(raw || "{}")).projectId || ""); } catch (_) {} + const result = _cancelByProject(projectId); + console.warn("[aria-cancel] /cancel project=%s — killed %d", projectId || "(main)", result.killed); + res.writeHead(200, { "Content-Type": "application/json" }); + res.end(JSON.stringify({ ok: true, ...result })); + }); + return; + } if (req.method === "GET" && req.url === "/health") { res.writeHead(200, { "Content-Type": "application/json" }); res.end(JSON.stringify({ ok: true, active: _activeSubprocesses.size }));