ai-workspace-lab/xworkmate-app
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Batch 6: normalize gateway chat run streaming

Browse Source
This commit is contained in:
Haitao Pan 2026-03-29 16:56:58 +08:00
parent 7d6d966071
commit bd99ac2f88
5 changed files with 316 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

98
go/go_core/internal/gatewayruntime/chat_run.go Normal file
View File

@ -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"))
}

13
go/go_core/internal/gatewayruntime/runtime.go
View File

@ -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{

65
go/go_core/internal/gatewayruntime/runtime_test.go
View File

@ -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

45
lib/runtime/runtime_controllers_gateway.dart
View File

@ -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<String, dynamic> payload) {
void handleChatRunEventInternal(Map<String, dynamic> 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<String, dynamic> payload) {
final message = asMap(payload['message']);
final role = (stringValue(message['role']) ?? '').toLowerCase();
handleChatRunEventInternal(<String, dynamic>{
'runId': payload['runId'],
'sessionKey': payload['sessionKey'],
'state': payload['state'],
if (role == 'assistant') 'assistantText': extractMessageText(message),
'errorMessage': payload['errorMessage'],
'terminal': false,
});
}
void handleAgentEventInternal(Map<String, dynamic> 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(<String, dynamic>{
'runId': runId,
'sessionKey': payload['sessionKey'] ?? data['sessionKey'],
'state': 'delta',
'assistantText': nextText,
'source': 'agent',
'terminal': false,
});
}
}
}

105
test/runtime/gateway_runtime_suite.dart
View File

@ -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(<String, Object>{});
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: <String, dynamic>{
'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: <String, dynamic>{
'runId': 'run-1',
'sessionKey': 'agent:main:main',
'state': 'final',
'assistantText': 'HELLO',
'terminal': true,
},
),
);
await Future<void>.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<Map<String, dynamic>> _history = <Map<String, dynamic>>[];
@override
bool get isConnected => true;
void addAssistantMessage(String text) {
_history.add(<String, dynamic>{
'role': 'assistant',
'content': <Map<String, dynamic>>[
<String, dynamic>{'type': 'text', 'text': text},
],
'timestamp': DateTime.now().millisecondsSinceEpoch,
});
}
@override
Future<dynamic> request(
String method, {
Map<String, dynamic>? params,
Duration timeout = const Duration(seconds: 15),
}) async {
switch (method) {
case 'chat.history':
return <String, dynamic>{'messages': List<Object>.from(_history)};
case 'chat.send':
final text = params?['message']?.toString().trim() ?? '';
if (text.isNotEmpty) {
_history.add(<String, dynamic>{
'role': 'user',
'content': <Map<String, dynamic>>[
<String, dynamic>{'type': 'text', 'text': text},
],
'timestamp': DateTime.now().millisecondsSinceEpoch,
});
}
return <String, dynamic>{'runId': 'run-1'};
case 'chat.abort':
return const <String, dynamic>{};
default:
return const <String, dynamic>{};
}
}
}
class FakeGatewayRuntimeServerInternal {
FakeGatewayRuntimeServerInternal._(
this.serverInternal, {