feat(multitask): kontext-getaggte Activity + kontext-scoped Cancel (Restpunkte 1+2)
Beide Restpunkte teilen eine Verkabelung: die projectId fliesst jetzt bis zum
Proxy und zurueck in die agent_activity-Events.
Restpunkt 1 — per-Kontext Activity-Indikator:
- Brain: proxy_client.chat_full(project_id) → payload.aria_project_id;
agent.py gibt active_project_id mit.
- Proxy routes.js: liest aria_project_id, taggt Tool-/Stream-Hooks damit,
trackt Subprozesse pro Kontext.
- Bridge: _emit_activity(project_id) + payload.projectId; /internal/agent-activity
reicht projectId durch; send_to_core/_process_core_response taggen thinking/idle
mit dem Turn-Kontext.
- App: agentActivityByCtx-Map; „ARIA denkt"-Indikator zeigt nur den
fokussierten Kontext statt global zu flackern.
Restpunkt 2 — kontext-scoped Cancel (Barge-In):
- Proxy: neuer /cancel {projectId} killt NUR die Subprozesse eines Kontexts
(_cancelByProject); /cancel-all bleibt fuers NOT-AUS.
- Bridge: soft cancel_request → _cancel_proxy_for_project(projectId) statt des
toten Diagnostic /api/cancel. Hard bleibt /cancel-all.
- App: cancel_request + Abbrechen-Button tragen die fokussierte projectId.
Damit laufen Kontexte in der App echt parallel: Arbeit in Projekt A wird durch
Senden/Abbrechen in Hauptchat oder Projekt B nicht mehr abgewuergt, und der
Indikator gehoert zum sichtbaren Chat.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -301,6 +301,9 @@ const ChatScreen: React.FC = () => {
|
|||||||
const [searchIndex, setSearchIndex] = useState(0); // welcher Treffer aktiv ist
|
const [searchIndex, setSearchIndex] = useState(0); // welcher Treffer aktiv ist
|
||||||
const [pendingAttachments, setPendingAttachments] = useState<{file: any, isPhoto: boolean}[]>([]);
|
const [pendingAttachments, setPendingAttachments] = useState<{file: any, isPhoto: boolean}[]>([]);
|
||||||
const [agentActivity, setAgentActivity] = useState<{activity: string, tool: string}>({activity: 'idle', tool: ''});
|
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<Record<string, {activity: string; tool: string}>>({});
|
||||||
// Gedanken-Stream: chronologisches Log dessen was ARIA intern macht.
|
// Gedanken-Stream: chronologisches Log dessen was ARIA intern macht.
|
||||||
// Wird aus agent_activity-Events gefuettert und in AsyncStorage persistiert.
|
// Wird aus agent_activity-Events gefuettert und in AsyncStorage persistiert.
|
||||||
const [thoughts, setThoughts] = useState<ThoughtEntry[]>([]);
|
const [thoughts, setThoughts] = useState<ThoughtEntry[]>([]);
|
||||||
@@ -1209,7 +1212,11 @@ const ChatScreen: React.FC = () => {
|
|||||||
if (message.type === 'agent_activity') {
|
if (message.type === 'agent_activity') {
|
||||||
const activity = (message.payload.activity as string) || 'idle';
|
const activity = (message.payload.activity as string) || 'idle';
|
||||||
const tool = (message.payload.tool as string) || '';
|
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 });
|
setAgentActivity({ activity, tool });
|
||||||
|
setAgentActivityByCtx(prev => ({ ...prev, [actPid]: { activity, tool } }));
|
||||||
// Implizite ACK-Bestaetigung: Brain hat angefangen zu arbeiten →
|
// Implizite ACK-Bestaetigung: Brain hat angefangen zu arbeiten →
|
||||||
// unsere Nachricht ist offensichtlich angekommen, auch wenn das
|
// unsere Nachricht ist offensichtlich angekommen, auch wenn das
|
||||||
// chat_ack aus irgendeinem Grund nicht durchkam. Alle laufenden
|
// chat_ack aus irgendeinem Grund nicht durchkam. Alle laufenden
|
||||||
@@ -1930,11 +1937,13 @@ const ChatScreen: React.FC = () => {
|
|||||||
});
|
});
|
||||||
}, [inputText, getCurrentLocation, pendingAttachments, sendPendingAttachments, interruptAriaIfBusy, dispatchWithAck]);
|
}, [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 cancelRequest = useCallback(() => {
|
||||||
|
const pid = focusedProjectIdRef.current || '';
|
||||||
setAgentActivity({ activity: 'idle', tool: '' });
|
setAgentActivity({ activity: 'idle', tool: '' });
|
||||||
|
setAgentActivityByCtx(prev => ({ ...prev, [pid]: { activity: 'idle', tool: '' } }));
|
||||||
clearStuckWatchdog();
|
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-
|
// Barge-In: wenn der User waehrend ARIA arbeitet/spricht eine neue Sprach-
|
||||||
@@ -2759,13 +2768,18 @@ const ChatScreen: React.FC = () => {
|
|||||||
}
|
}
|
||||||
/>
|
/>
|
||||||
|
|
||||||
{/* Thinking-Indicator */}
|
{/* Thinking-Indicator \u2014 NUR fuer den fokussierten Kontext (Multi-Threading).
|
||||||
{agentActivity.activity !== 'idle' && (
|
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 (
|
||||||
<View style={styles.thinkingBar}>
|
<View style={styles.thinkingBar}>
|
||||||
<Text style={styles.thinkingText}>
|
<Text style={styles.thinkingText}>
|
||||||
{agentActivity.activity === 'tool' && agentActivity.tool
|
{focusAct.activity === 'tool' && focusAct.tool
|
||||||
? `\uD83D\uDD27 ${agentActivity.tool}`
|
? `\uD83D\uDD27 ${focusAct.tool}`
|
||||||
: agentActivity.activity === 'assistant'
|
: focusAct.activity === 'assistant'
|
||||||
? '\u270D\uFE0F ARIA schreibt...'
|
? '\u270D\uFE0F ARIA schreibt...'
|
||||||
: '\uD83D\uDCAD ARIA denkt...'}
|
: '\uD83D\uDCAD ARIA denkt...'}
|
||||||
</Text>
|
</Text>
|
||||||
@@ -2775,7 +2789,8 @@ const ChatScreen: React.FC = () => {
|
|||||||
</TouchableOpacity>
|
</TouchableOpacity>
|
||||||
</View>
|
</View>
|
||||||
</View>
|
</View>
|
||||||
)}
|
);
|
||||||
|
})()}
|
||||||
|
|
||||||
{/* Pending Anhaenge Vorschau */}
|
{/* Pending Anhaenge Vorschau */}
|
||||||
{pendingAttachments.length > 0 && (
|
{pendingAttachments.length > 0 && (
|
||||||
|
|||||||
+2
-1
@@ -1210,7 +1210,8 @@ class Agent:
|
|||||||
final_reply = ""
|
final_reply = ""
|
||||||
try:
|
try:
|
||||||
for iteration in range(self.MAX_TOOL_ITERATIONS):
|
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:
|
if result.tool_calls:
|
||||||
# Assistant-Turn mit tool_calls in messages anhaengen (nicht in Conversation!)
|
# Assistant-Turn mit tool_calls in messages anhaengen (nicht in Conversation!)
|
||||||
messages.append(ProxyMessage(
|
messages.append(ProxyMessage(
|
||||||
|
|||||||
@@ -94,6 +94,7 @@ class ProxyClient:
|
|||||||
messages: List[Message],
|
messages: List[Message],
|
||||||
tools: Optional[list] = None,
|
tools: Optional[list] = None,
|
||||||
model: Optional[str] = None,
|
model: Optional[str] = None,
|
||||||
|
project_id: str = "",
|
||||||
) -> ProxyResult:
|
) -> ProxyResult:
|
||||||
"""Full chat — kann Tool-Calls liefern (wenn tools mitgegeben).
|
"""Full chat — kann Tool-Calls liefern (wenn tools mitgegeben).
|
||||||
|
|
||||||
@@ -108,6 +109,11 @@ class ProxyClient:
|
|||||||
}
|
}
|
||||||
if tools:
|
if tools:
|
||||||
payload["tools"] = 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)",
|
logger.info("Proxy → %s (%d Messages, %d tools, model=%s)",
|
||||||
url, len(messages), len(tools or []), payload["model"])
|
url, len(messages), len(tools or []), payload["model"])
|
||||||
try:
|
try:
|
||||||
|
|||||||
+48
-14
@@ -1357,7 +1357,7 @@ class ARIABridge:
|
|||||||
# _last_chat_final_at bewusst NICHT setzen: die 3s-Cooldown war fuer
|
# _last_chat_final_at bewusst NICHT setzen: die 3s-Cooldown war fuer
|
||||||
# trailing OpenClaw-Activity-Events; bei Voice-Chat wuerde sie die
|
# trailing OpenClaw-Activity-Events; bei Voice-Chat wuerde sie die
|
||||||
# naechste thinking-Welle unterdruecken.
|
# 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 Persistence (global, nicht pro Geraet) ──────
|
||||||
_MODE_FILE = "/shared/config/mode.json"
|
_MODE_FILE = "/shared/config/mode.json"
|
||||||
@@ -1572,7 +1572,7 @@ class ARIABridge:
|
|||||||
|
|
||||||
# agent_activity → thinking. _emit_activity statt direktem _send_to_rvs
|
# agent_activity → thinking. _emit_activity statt direktem _send_to_rvs
|
||||||
# damit der State-Cache fuer die spaetere idle-Dedup richtig steht.
|
# 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():
|
def _do_call():
|
||||||
try:
|
try:
|
||||||
@@ -1591,7 +1591,7 @@ class ARIABridge:
|
|||||||
status, body = await asyncio.get_event_loop().run_in_executor(None, _do_call)
|
status, body = await asyncio.get_event_loop().run_in_executor(None, _do_call)
|
||||||
if status != 200:
|
if status != 200:
|
||||||
logger.error("[brain] /chat fehlgeschlagen: status=%s body=%s", status, body[: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({
|
await self._send_to_rvs({
|
||||||
"type": "chat",
|
"type": "chat",
|
||||||
"payload": {
|
"payload": {
|
||||||
@@ -1606,13 +1606,13 @@ class ARIABridge:
|
|||||||
data = json.loads(body)
|
data = json.loads(body)
|
||||||
except Exception:
|
except Exception:
|
||||||
logger.error("[brain] /chat lieferte ungueltiges JSON: %s", body[:200])
|
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
|
return
|
||||||
|
|
||||||
reply = (data.get("reply") or "").strip()
|
reply = (data.get("reply") or "").strip()
|
||||||
if not reply:
|
if not reply:
|
||||||
logger.warning("[brain] /chat: leerer Reply")
|
logger.warning("[brain] /chat: leerer Reply")
|
||||||
await self._emit_activity("idle", "")
|
await self._emit_activity("idle", "", project_id=project_id)
|
||||||
return
|
return
|
||||||
|
|
||||||
# Projekt-Kontext des Turns — wird an _process_core_response weiter-
|
# 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})
|
await self._process_core_response(reply, {"projectId": turn_project_id})
|
||||||
except Exception:
|
except Exception:
|
||||||
logger.exception("[brain] _process_core_response Fehler")
|
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)
|
# Originaler Fallback-Send (toter Code, _emit_activity uebernimmt jetzt)
|
||||||
await self._send_to_rvs({
|
await self._send_to_rvs({
|
||||||
"type": "agent_activity",
|
"type": "agent_activity",
|
||||||
@@ -2008,10 +2008,14 @@ class ARIABridge:
|
|||||||
logger.warning("[rvs] NOT-AUS — hard cancel: Diagnostic /api/cancel + Proxy /cancel-all")
|
logger.warning("[rvs] NOT-AUS — hard cancel: Diagnostic /api/cancel + Proxy /cancel-all")
|
||||||
await self._cancel_via_diagnostic()
|
await self._cancel_via_diagnostic()
|
||||||
await self._cancel_proxy_subprocesses()
|
await self._cancel_proxy_subprocesses()
|
||||||
|
await self._emit_activity("idle", "")
|
||||||
else:
|
else:
|
||||||
logger.info("[rvs] Cancel-Request von App — rufe Diagnostic /api/cancel auf")
|
# Barge-In: nur den fokussierten Kontext abbrechen (projectId von
|
||||||
await self._cancel_via_diagnostic()
|
# der App), damit parallele Arbeit in anderen Kontexten weiterlaeuft.
|
||||||
await self._emit_activity("idle", "")
|
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
|
return
|
||||||
|
|
||||||
elif msg_type == "audio_pcm":
|
elif msg_type == "audio_pcm":
|
||||||
@@ -3518,7 +3522,30 @@ class ARIABridge:
|
|||||||
status, body = await asyncio.get_event_loop().run_in_executor(None, _do_request)
|
status, body = await asyncio.get_event_loop().run_in_executor(None, _do_request)
|
||||||
logger.warning("[NOT-AUS] proxy /cancel-all: %s %s", status, body)
|
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.
|
"""Sendet agent_activity an die App — nur wenn sich der State geaendert hat.
|
||||||
|
|
||||||
Trailing Agent-Events nach chat:final werden 3s lang unterdrueckt
|
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
|
force=True: kein State-Dedup — wird vom Proxy-Tool-Hook genutzt
|
||||||
damit auch wiederholte gleiche Tool-Aufrufe (z.B. 3x Bash
|
damit auch wiederholte gleiche Tool-Aufrufe (z.B. 3x Bash
|
||||||
hintereinander) im Gedanken-Stream als eigene Eintraege sichtbar
|
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:
|
if activity != "idle" and self._last_chat_final_at > 0:
|
||||||
since_final = asyncio.get_event_loop().time() - self._last_chat_final_at
|
since_final = asyncio.get_event_loop().time() - self._last_chat_final_at
|
||||||
if since_final < 3.0:
|
if since_final < 3.0:
|
||||||
return
|
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:
|
if not force and state == self._last_activity_state:
|
||||||
return
|
return
|
||||||
self._last_activity_state = state
|
self._last_activity_state = state
|
||||||
await self._send_to_rvs({
|
await self._send_to_rvs({
|
||||||
"type": "agent_activity",
|
"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),
|
"timestamp": int(asyncio.get_event_loop().time() * 1000),
|
||||||
})
|
})
|
||||||
|
|
||||||
@@ -3695,9 +3727,11 @@ class ARIABridge:
|
|||||||
if not tool:
|
if not tool:
|
||||||
await _send_response(writer, 400, {"error": "tool erforderlich"})
|
await _send_response(writer, 400, {"error": "tool erforderlich"})
|
||||||
return
|
return
|
||||||
|
tool_pid = str(data.get("projectId") or "")
|
||||||
# Force-emit (kein Dedup): User soll JEDEN Tool-Call sehen
|
# Force-emit (kein Dedup): User soll JEDEN Tool-Call sehen
|
||||||
# selbst wenn derselbe Name zweimal in Folge kommt.
|
# 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})
|
await _send_response(writer, 200, {"ok": True})
|
||||||
elif method == "POST" and path == "/internal/agent-stream":
|
elif method == "POST" and path == "/internal/agent-stream":
|
||||||
# Vom Proxy gefeuert: voller Live-Stream der Claude-Code-
|
# Vom Proxy gefeuert: voller Live-Stream der Claude-Code-
|
||||||
|
|||||||
+54
-12
@@ -70,9 +70,9 @@ function _postJson(url, body) {
|
|||||||
/**
|
/**
|
||||||
* Pusht einen Tool-Use-Event an die Bridge (alter Gedanken-Stream-Pfad).
|
* Pusht einen Tool-Use-Event an die Bridge (alter Gedanken-Stream-Pfad).
|
||||||
*/
|
*/
|
||||||
function _emitToolEvent(toolName) {
|
function _emitToolEvent(toolName, projectId) {
|
||||||
if (!toolName) return;
|
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 ──────────────────────────
|
// ── Subprocess-Tracking fuer Not-Aus ──────────────────────────
|
||||||
// requestId → ClaudeSubprocess. Eintraege werden beim close/result-Event
|
// requestId → ClaudeSubprocess. Eintraege werden beim close/result-Event
|
||||||
// wieder entfernt. /v1/cancel-all iteriert und ruft .kill() auf jeden.
|
// 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();
|
const _activeSubprocesses = new Map();
|
||||||
function _trackSubprocess(requestId, subprocess) {
|
function _trackSubprocess(requestId, subprocess, projectId) {
|
||||||
_activeSubprocesses.set(requestId, subprocess);
|
_activeSubprocesses.set(requestId, { subprocess, projectId: projectId || "" });
|
||||||
const cleanup = () => _activeSubprocesses.delete(requestId);
|
const cleanup = () => _activeSubprocesses.delete(requestId);
|
||||||
subprocess.on("close", cleanup);
|
subprocess.on("close", cleanup);
|
||||||
subprocess.on("error", cleanup);
|
subprocess.on("error", cleanup);
|
||||||
@@ -149,24 +151,25 @@ function _attachIdleWatchdog(subprocess, requestId) {
|
|||||||
* - Alt-API: nur Tool-Namen an /internal/agent-activity (Gedanken-Stream)
|
* - Alt-API: nur Tool-Namen an /internal/agent-activity (Gedanken-Stream)
|
||||||
* - Neu-API: voller Stream (text/tool_use/tool_result) an /internal/agent-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) => {
|
subprocess.on("assistant", (message) => {
|
||||||
try {
|
try {
|
||||||
const blocks = message?.message?.content || [];
|
const blocks = message?.message?.content || [];
|
||||||
for (const b of blocks) {
|
for (const b of blocks) {
|
||||||
if (!b) continue;
|
if (!b) continue;
|
||||||
if (b.type === "tool_use") {
|
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 inputStr = b.input ? JSON.stringify(b.input) : "";
|
||||||
const inp = _truncate(inputStr, TOOL_INPUT_MAX_CHARS);
|
const inp = _truncate(inputStr, TOOL_INPUT_MAX_CHARS);
|
||||||
_emitStreamEvent(requestId, "tool_use", {
|
_emitStreamEvent(requestId, "tool_use", {
|
||||||
|
projectId: projectId || "",
|
||||||
id: b.id || null,
|
id: b.id || null,
|
||||||
name: b.name || "",
|
name: b.name || "",
|
||||||
input: inp.text,
|
input: inp.text,
|
||||||
inputTruncatedBytes: inp.truncatedBytes,
|
inputTruncatedBytes: inp.truncatedBytes,
|
||||||
});
|
});
|
||||||
} else if (b.type === "text" && b.text) {
|
} 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) {
|
} else if (b.type === "thinking" && b.thinking) {
|
||||||
// Wenn das Modell Extended Thinking emittiert — selten in
|
// Wenn das Modell Extended Thinking emittiert — selten in
|
||||||
// Claude Code CLI, aber moeglich. Markieren wir extra.
|
// Claude Code CLI, aber moeglich. Markieren wir extra.
|
||||||
@@ -227,15 +230,18 @@ export async function handleChatCompletions(req, res) {
|
|||||||
}
|
}
|
||||||
// Convert to CLI input format
|
// Convert to CLI input format
|
||||||
const cliInput = openaiToCli(body);
|
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();
|
const subprocess = new ClaudeSubprocess();
|
||||||
// ARIA-Patch: Tool-Use-Events + voller Live-Stream an die Bridge.
|
// ARIA-Patch: Tool-Use-Events + voller Live-Stream an die Bridge.
|
||||||
// Plus: Subprocess fuer Not-Aus tracken (Hard-Kill via /v1/cancel-all).
|
// Plus: Subprocess fuer Not-Aus tracken (Hard-Kill via /v1/cancel-all).
|
||||||
// Plus: Idle-Watchdog — Subprocess darf ewig laufen solange Events
|
// Plus: Idle-Watchdog — Subprocess darf ewig laufen solange Events
|
||||||
// kommen, wird aber gekillt nach IDLE_TIMEOUT_MS Inaktivitaet.
|
// kommen, wird aber gekillt nach IDLE_TIMEOUT_MS Inaktivitaet.
|
||||||
_attachToolHook(subprocess, requestId);
|
_attachToolHook(subprocess, requestId, ariaProjectId);
|
||||||
_trackSubprocess(requestId, subprocess);
|
_trackSubprocess(requestId, subprocess, ariaProjectId);
|
||||||
_attachIdleWatchdog(subprocess, requestId);
|
_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("result", () => _emitStreamEvent(requestId, "end", { reason: "result" }));
|
||||||
subprocess.on("close", (code) => _emitStreamEvent(requestId, "end", { reason: "close", code }));
|
subprocess.on("close", (code) => _emitStreamEvent(requestId, "end", { reason: "close", code }));
|
||||||
subprocess.on("error", (err) => _emitStreamEvent(requestId, "end", { reason: "error", error: String(err?.message || err) }));
|
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() {
|
function _cancelAll() {
|
||||||
const ids = Array.from(_activeSubprocesses.keys());
|
const ids = Array.from(_activeSubprocesses.keys());
|
||||||
let killed = 0;
|
let killed = 0;
|
||||||
for (const [id, subp] of _activeSubprocesses) {
|
for (const [id, entry] of _activeSubprocesses) {
|
||||||
try {
|
try {
|
||||||
subp.kill();
|
entry.subprocess.kill();
|
||||||
killed++;
|
killed++;
|
||||||
} catch (e) {
|
} catch (e) {
|
||||||
console.error("[aria-not-aus] kill failed for", id, e?.message);
|
console.error("[aria-not-aus] kill failed for", id, e?.message);
|
||||||
@@ -509,6 +515,27 @@ function _cancelAll() {
|
|||||||
return { killed, requestIds: ids };
|
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 {
|
try {
|
||||||
const internalServer = http.createServer((req, res) => {
|
const internalServer = http.createServer((req, res) => {
|
||||||
if (req.method === "POST" && req.url === "/cancel-all") {
|
if (req.method === "POST" && req.url === "/cancel-all") {
|
||||||
@@ -518,6 +545,21 @@ try {
|
|||||||
res.end(JSON.stringify({ ok: true, ...result }));
|
res.end(JSON.stringify({ ok: true, ...result }));
|
||||||
return;
|
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") {
|
if (req.method === "GET" && req.url === "/health") {
|
||||||
res.writeHead(200, { "Content-Type": "application/json" });
|
res.writeHead(200, { "Content-Type": "application/json" });
|
||||||
res.end(JSON.stringify({ ok: true, active: _activeSubprocesses.size }));
|
res.end(JSON.stringify({ ok: true, active: _activeSubprocesses.size }));
|
||||||
|
|||||||
Reference in New Issue
Block a user