diff --git a/go/go_core/internal/gatewayruntime/chat_run.go b/go/go_core/internal/gatewayruntime/chat_run.go new file mode 100644 index 00000000..2e6d2592 --- /dev/null +++ b/go/go_core/internal/gatewayruntime/chat_run.go @@ -0,0 +1,98 @@ +package gatewayruntime + +import "strings" + +func normalizeChatRunEvent(event string, payload map[string]any) map[string]any { + switch event { + case "chat": + runID := strings.TrimSpace(stringValue(payload["runId"])) + state := strings.TrimSpace(stringValue(payload["state"])) + if runID == "" && state == "" { + return nil + } + message := asMap(payload["message"]) + assistantText := "" + if strings.EqualFold(strings.TrimSpace(stringValue(message["role"])), "assistant") { + assistantText = extractMessageText(message) + } + normalized := map[string]any{ + "runId": runID, + "sessionKey": strings.TrimSpace(stringValue(payload["sessionKey"])), + "state": state, + "source": "chat", + "terminal": state == "final" || state == "aborted" || state == "error", + } + if assistantText != "" { + normalized["assistantText"] = assistantText + } + if errorMessage := strings.TrimSpace(stringValue(payload["errorMessage"])); errorMessage != "" { + normalized["errorMessage"] = errorMessage + } + return normalized + case "agent": + runID := strings.TrimSpace(stringValue(payload["runId"])) + if runID == "" { + return nil + } + stream := strings.TrimSpace(stringValue(payload["stream"])) + if !strings.EqualFold(stream, "assistant") { + return nil + } + data := asMap(payload["data"]) + assistantText := strings.TrimSpace(stringValue(data["text"])) + if assistantText == "" { + assistantText = extractMessageText(data) + } + if assistantText == "" { + return nil + } + sessionKey := strings.TrimSpace(stringValue(payload["sessionKey"])) + if sessionKey == "" { + sessionKey = strings.TrimSpace(stringValue(data["sessionKey"])) + } + return map[string]any{ + "runId": runID, + "sessionKey": sessionKey, + "state": "delta", + "source": "agent", + "stream": stream, + "assistantText": assistantText, + "terminal": false, + } + default: + return nil + } +} + +func asList(value any) []any { + switch typed := value.(type) { + case []any: + return typed + default: + return nil + } +} + +func extractMessageText(message map[string]any) string { + directContent, ok := message["content"].(string) + if ok { + return strings.TrimSpace(directContent) + } + parts := make([]string, 0, 4) + for _, part := range asList(message["content"]) { + segment := asMap(part) + text := strings.TrimSpace(firstNonEmpty( + stringValue(segment["text"]), + stringValue(segment["thinking"]), + )) + if text != "" { + parts = append(parts, text) + continue + } + nestedContent := strings.TrimSpace(stringValue(segment["content"])) + if nestedContent != "" { + parts = append(parts, nestedContent) + } + } + return strings.TrimSpace(strings.Join(parts, "\n")) +} diff --git a/go/go_core/internal/gatewayruntime/runtime.go b/go/go_core/internal/gatewayruntime/runtime.go index 31bba3b2..0fcf39b7 100644 --- a/go/go_core/internal/gatewayruntime/runtime.go +++ b/go/go_core/internal/gatewayruntime/runtime.go @@ -597,6 +597,19 @@ func (s *session) handleEvent( case "seqGap": s.appendLog("warn", "sync", "sequence gap detected") } + if normalized := normalizeChatRunEvent(event, payload); len(normalized) > 0 { + s.emitNotification( + "xworkmate.gateway.push", + map[string]any{ + "runtimeId": s.runtimeID, + "event": map[string]any{ + "event": "chat.run", + "payload": normalized, + "sequence": intValue(decoded["seq"]), + }, + }, + ) + } s.emitNotification( "xworkmate.gateway.push", map[string]any{ diff --git a/go/go_core/internal/gatewayruntime/runtime_test.go b/go/go_core/internal/gatewayruntime/runtime_test.go index a2aa65cb..95845cb2 100644 --- a/go/go_core/internal/gatewayruntime/runtime_test.go +++ b/go/go_core/internal/gatewayruntime/runtime_test.go @@ -110,6 +110,71 @@ func TestManagerSuppressesReconnectForPairingRequired(t *testing.T) { } } +func TestSessionEmitsNormalizedChatRunPushEvents(t *testing.T) { + manager := NewManager() + session := newSession(manager, "runtime-1") + notifications := make([]map[string]any, 0, 8) + session.setNotify(func(message map[string]any) { + notifications = append(notifications, message) + }) + + session.handleEvent( + "chat", + map[string]any{"seq": 7}, + map[string]any{ + "runId": "run-1", + "sessionKey": "agent:main:main", + "state": "final", + "message": map[string]any{ + "role": "assistant", + "content": []any{ + map[string]any{"type": "text", "text": "XWORKMATE_OK"}, + }, + }, + }, + ) + session.handleEvent( + "agent", + map[string]any{"seq": 8}, + map[string]any{ + "runId": "run-1", + "stream": "assistant", + "data": map[string]any{ + "text": "DELTA_TEXT", + }, + }, + ) + + normalized := make([]map[string]any, 0, 2) + for _, notification := range notifications { + if strings.TrimSpace(stringValue(notification["method"])) != "xworkmate.gateway.push" { + continue + } + params := asMap(notification["params"]) + event := asMap(params["event"]) + if strings.TrimSpace(stringValue(event["event"])) != "chat.run" { + continue + } + normalized = append(normalized, asMap(event["payload"])) + } + + if len(normalized) != 2 { + t.Fatalf("expected 2 normalized chat.run notifications, got %#v", normalized) + } + if normalized[0]["runId"] != "run-1" || normalized[0]["state"] != "final" { + t.Fatalf("unexpected normalized chat payload %#v", normalized[0]) + } + if normalized[0]["assistantText"] != "XWORKMATE_OK" { + t.Fatalf("expected final assistant text, got %#v", normalized[0]) + } + if normalized[0]["terminal"] != true { + t.Fatalf("expected terminal final chat.run, got %#v", normalized[0]) + } + if normalized[1]["assistantText"] != "DELTA_TEXT" || normalized[1]["state"] != "delta" { + t.Fatalf("unexpected normalized agent payload %#v", normalized[1]) + } +} + type fakeGatewayServer struct { server *http.Server listener net.Listener diff --git a/lib/runtime/runtime_controllers_gateway.dart b/lib/runtime/runtime_controllers_gateway.dart index 474cab8f..a923238d 100644 --- a/lib/runtime/runtime_controllers_gateway.dart +++ b/lib/runtime/runtime_controllers_gateway.dart @@ -293,6 +293,10 @@ class GatewayChatController extends ChangeNotifier { } void handleEvent(GatewayPushEvent event) { + if (event.event == 'chat.run') { + handleChatRunEventInternal(asMap(event.payload)); + return; + } if (event.event == 'chat') { handleChatEventInternal(asMap(event.payload)); return; @@ -310,7 +314,7 @@ class GatewayChatController extends ChangeNotifier { notifyListeners(); } - void handleChatEventInternal(Map payload) { + void handleChatRunEventInternal(Map payload) { final runId = stringValue(payload['runId']); final state = stringValue(payload['state']) ?? ''; final incomingSessionKey = @@ -321,18 +325,20 @@ class GatewayChatController extends ChangeNotifier { return; } - final message = asMap(payload['message']); - final role = (stringValue(message['role']) ?? '').toLowerCase(); - final text = extractMessageText(message); - if (role == 'assistant' && - text.isNotEmpty && + final assistantText = stringValue(payload['assistantText']) ?? ''; + if (assistantText.isNotEmpty && (state == 'delta' || state == 'final')) { - streamingAssistantTextInternal = text; + streamingAssistantTextInternal = assistantText; } if (state == 'error') { errorInternal = stringValue(payload['errorMessage']) ?? 'Chat failed'; } - if (state == 'final' || state == 'aborted' || state == 'error') { + final terminal = + boolValue(payload['terminal']) ?? false || + state == 'final' || + state == 'aborted' || + state == 'error'; + if (terminal) { if (runId != null) { pendingRunsInternal.remove(runId); } else { @@ -345,6 +351,19 @@ class GatewayChatController extends ChangeNotifier { notifyListeners(); } + void handleChatEventInternal(Map payload) { + final message = asMap(payload['message']); + final role = (stringValue(message['role']) ?? '').toLowerCase(); + handleChatRunEventInternal({ + 'runId': payload['runId'], + 'sessionKey': payload['sessionKey'], + 'state': payload['state'], + if (role == 'assistant') 'assistantText': extractMessageText(message), + 'errorMessage': payload['errorMessage'], + 'terminal': false, + }); + } + void handleAgentEventInternal(Map payload) { final runId = stringValue(payload['runId']); if (runId == null || !pendingRunsInternal.contains(runId)) { @@ -355,8 +374,14 @@ class GatewayChatController extends ChangeNotifier { if (stream == 'assistant') { final nextText = stringValue(data['text']) ?? extractMessageText(data); if (nextText.isNotEmpty) { - streamingAssistantTextInternal = nextText; - notifyListeners(); + handleChatRunEventInternal({ + 'runId': runId, + 'sessionKey': payload['sessionKey'] ?? data['sessionKey'], + 'state': 'delta', + 'assistantText': nextText, + 'source': 'agent', + 'terminal': false, + }); } } } diff --git a/test/runtime/gateway_runtime_suite.dart b/test/runtime/gateway_runtime_suite.dart index 41f144b5..14c5746f 100644 --- a/test/runtime/gateway_runtime_suite.dart +++ b/test/runtime/gateway_runtime_suite.dart @@ -10,6 +10,7 @@ import 'package:shared_preferences/shared_preferences.dart'; import 'package:xworkmate/runtime/device_identity_store.dart'; import 'package:xworkmate/runtime/gateway_runtime.dart'; import 'package:xworkmate/runtime/gateway_runtime_session_client.dart'; +import 'package:xworkmate/runtime/runtime_controllers.dart'; import 'package:xworkmate/runtime/runtime_models.dart'; import '../test_support.dart'; @@ -224,6 +225,62 @@ void main() { }, ); + test( + 'GatewayChatController applies normalized chat.run updates from go-core', + () async { + SharedPreferences.setMockInitialValues({}); + final store = createIsolatedTestStore(); + final runtime = _FakeGatewayRuntimeForChatController(store: store); + final controller = GatewayChatController(runtime); + addTearDown(controller.dispose); + + await controller.loadSession('agent:main:main'); + await controller.sendMessage( + sessionKey: 'agent:main:main', + message: 'hello', + thinking: 'low', + ); + + expect(controller.hasPendingRun, isTrue); + runtime.addAssistantMessage('HELLO'); + controller.handleEvent( + const GatewayPushEvent( + event: 'chat.run', + payload: { + 'runId': 'run-1', + 'sessionKey': 'agent:main:main', + 'state': 'delta', + 'assistantText': 'HELLO', + 'terminal': false, + }, + ), + ); + expect(controller.streamingAssistantText, 'HELLO'); + + controller.handleEvent( + const GatewayPushEvent( + event: 'chat.run', + payload: { + 'runId': 'run-1', + 'sessionKey': 'agent:main:main', + 'state': 'final', + 'assistantText': 'HELLO', + 'terminal': true, + }, + ), + ); + + await Future.delayed(Duration.zero); + expect(controller.hasPendingRun, isFalse); + expect( + controller.messages.any( + (message) => message.role == 'assistant' && message.text == 'HELLO', + ), + isTrue, + ); + }, + ); + test( 'GatewayRuntime does not silently fall back to direct websocket when go-core bridge is unavailable', () async { @@ -520,6 +577,54 @@ class _FakeGatewayRuntimeSessionClient implements GatewayRuntimeSessionClient { } } +class _FakeGatewayRuntimeForChatController extends GatewayRuntime { + _FakeGatewayRuntimeForChatController({required super.store}) + : super(identityStore: DeviceIdentityStore(store)); + + final List> _history = >[]; + + @override + bool get isConnected => true; + + void addAssistantMessage(String text) { + _history.add({ + 'role': 'assistant', + 'content': >[ + {'type': 'text', 'text': text}, + ], + 'timestamp': DateTime.now().millisecondsSinceEpoch, + }); + } + + @override + Future request( + String method, { + Map? params, + Duration timeout = const Duration(seconds: 15), + }) async { + switch (method) { + case 'chat.history': + return {'messages': List.from(_history)}; + case 'chat.send': + final text = params?['message']?.toString().trim() ?? ''; + if (text.isNotEmpty) { + _history.add({ + 'role': 'user', + 'content': >[ + {'type': 'text', 'text': text}, + ], + 'timestamp': DateTime.now().millisecondsSinceEpoch, + }); + } + return {'runId': 'run-1'}; + case 'chat.abort': + return const {}; + default: + return const {}; + } + } +} + class FakeGatewayRuntimeServerInternal { FakeGatewayRuntimeServerInternal._( this.serverInternal, {