Fix OpenCode single-agent ACP transport
This commit is contained in:
Haitao Pan
parent
1ad60c6461
commit
2d1d8ecb42
@ -75,6 +75,7 @@ class DirectSingleAgentAppServerClient {
|
||||
final Map<String, _DirectAppServerConnection> _activeConnections =
|
||||
<String, _DirectAppServerConnection>{};
|
||||
final Map<String, String> _threadIds = <String, String>{};
|
||||
final Map<String, String> _restSessionIds = <String, String>{};
|
||||
final Set<String> _abortedSessions = <String>{};
|
||||
|
||||
final Map<SingleAgentProvider, DirectSingleAgentCapabilities>
|
||||
@ -97,6 +98,37 @@ class DirectSingleAgentAppServerClient {
|
||||
}
|
||||
|
||||
final endpoint = _resolveWebSocketEndpoint(provider);
|
||||
if (_usesRestSessionApi(provider)) {
|
||||
final base = endpointResolver(provider);
|
||||
if (base == null) {
|
||||
final unavailable = const DirectSingleAgentCapabilities.unavailable(
|
||||
endpoint: '',
|
||||
errorMessage: 'Single-agent app-server endpoint is not configured.',
|
||||
);
|
||||
_cachedCapabilities[provider] = unavailable;
|
||||
_capabilitiesRefreshedAt[provider] = DateTime.now();
|
||||
return unavailable;
|
||||
}
|
||||
try {
|
||||
await _fetchJson(
|
||||
_buildRestUri(base, '/global/health'),
|
||||
gatewayToken: gatewayToken,
|
||||
);
|
||||
_cachedCapabilities[provider] = DirectSingleAgentCapabilities(
|
||||
available: true,
|
||||
supportedProviders: <SingleAgentProvider>[provider],
|
||||
endpoint: base.toString(),
|
||||
);
|
||||
} catch (error) {
|
||||
_cachedCapabilities[provider] = DirectSingleAgentCapabilities.unavailable(
|
||||
endpoint: base.toString(),
|
||||
errorMessage: error.toString(),
|
||||
);
|
||||
} finally {
|
||||
_capabilitiesRefreshedAt[provider] = DateTime.now();
|
||||
}
|
||||
return _cachedCapabilities[provider]!;
|
||||
}
|
||||
if (endpoint == null) {
|
||||
final unavailable = const DirectSingleAgentCapabilities.unavailable(
|
||||
endpoint: '',
|
||||
@ -135,6 +167,9 @@ class DirectSingleAgentAppServerClient {
|
||||
Future<DirectSingleAgentRunResult> run(
|
||||
DirectSingleAgentRunRequest request,
|
||||
) async {
|
||||
if (_usesRestSessionApi(request.provider)) {
|
||||
return _runViaRestApi(request);
|
||||
}
|
||||
final endpoint = _resolveWebSocketEndpoint(request.provider);
|
||||
if (endpoint == null) {
|
||||
return const DirectSingleAgentRunResult(
|
||||
@ -302,6 +337,22 @@ class DirectSingleAgentAppServerClient {
|
||||
return;
|
||||
}
|
||||
_abortedSessions.add(normalizedSessionId);
|
||||
final restSessionId = _restSessionIds[normalizedSessionId]?.trim() ?? '';
|
||||
if (restSessionId.isNotEmpty) {
|
||||
final provider = SingleAgentProvider.opencode;
|
||||
final base = endpointResolver(provider);
|
||||
if (base != null) {
|
||||
try {
|
||||
await _postJson(
|
||||
_buildRestUri(base, '/session/$restSessionId/abort'),
|
||||
body: null,
|
||||
gatewayToken: '',
|
||||
);
|
||||
} catch (_) {
|
||||
// Best effort only.
|
||||
}
|
||||
}
|
||||
}
|
||||
final connection = _activeConnections[normalizedSessionId];
|
||||
final threadId = _threadIds[normalizedSessionId];
|
||||
if (connection == null || threadId == null || threadId.isEmpty) {
|
||||
@ -365,6 +416,442 @@ class DirectSingleAgentAppServerClient {
|
||||
return threadId;
|
||||
}
|
||||
|
||||
Future<DirectSingleAgentRunResult> _runViaRestApi(
|
||||
DirectSingleAgentRunRequest request,
|
||||
) async {
|
||||
final base = endpointResolver(request.provider);
|
||||
if (base == null) {
|
||||
return const DirectSingleAgentRunResult(
|
||||
success: false,
|
||||
output: '',
|
||||
errorMessage: 'Single-agent REST endpoint is missing.',
|
||||
);
|
||||
}
|
||||
final normalizedSessionId = request.sessionId.trim();
|
||||
if (normalizedSessionId.isEmpty) {
|
||||
return const DirectSingleAgentRunResult(
|
||||
success: false,
|
||||
output: '',
|
||||
errorMessage: 'Single-agent session id is missing.',
|
||||
);
|
||||
}
|
||||
|
||||
_abortedSessions.remove(normalizedSessionId);
|
||||
final remoteSessionId = await _ensureRestSession(
|
||||
base,
|
||||
sessionId: normalizedSessionId,
|
||||
workingDirectory: request.workingDirectory,
|
||||
gatewayToken: request.gatewayToken,
|
||||
);
|
||||
|
||||
final output = StringBuffer();
|
||||
final completion = Completer<DirectSingleAgentRunResult>();
|
||||
String? activeAssistantMessageId;
|
||||
String? lastAssistantText;
|
||||
var busySeen = false;
|
||||
|
||||
final eventClient = HttpClient()
|
||||
..connectionTimeout = const Duration(seconds: 8);
|
||||
late final HttpClientRequest eventRequest;
|
||||
late final HttpClientResponse eventResponse;
|
||||
StreamSubscription<String>? lineSubscription;
|
||||
|
||||
void completeSuccess() {
|
||||
if (completion.isCompleted) {
|
||||
return;
|
||||
}
|
||||
final resolvedOutput = output.toString().trim().isNotEmpty
|
||||
? output.toString()
|
||||
: (lastAssistantText ?? '');
|
||||
completion.complete(
|
||||
DirectSingleAgentRunResult(
|
||||
success: true,
|
||||
output: resolvedOutput,
|
||||
errorMessage: '',
|
||||
resolvedModel: request.model,
|
||||
),
|
||||
);
|
||||
}
|
||||
|
||||
try {
|
||||
final eventUri = _buildRestUri(base, '/global/event');
|
||||
eventRequest = await eventClient.getUrl(eventUri);
|
||||
eventRequest.headers.set(
|
||||
HttpHeaders.acceptHeader,
|
||||
'text/event-stream',
|
||||
);
|
||||
final normalizedToken = request.gatewayToken.trim();
|
||||
if (normalizedToken.isNotEmpty) {
|
||||
eventRequest.headers.set(
|
||||
HttpHeaders.authorizationHeader,
|
||||
'Bearer $normalizedToken',
|
||||
);
|
||||
}
|
||||
eventResponse = await eventRequest.close();
|
||||
lineSubscription = eventResponse
|
||||
.transform(utf8.decoder)
|
||||
.transform(const LineSplitter())
|
||||
.listen(
|
||||
(line) {
|
||||
if (!line.startsWith('data: ')) {
|
||||
return;
|
||||
}
|
||||
final event = _decodeMap(line.substring(6));
|
||||
final payload = _asMap(event['payload']);
|
||||
final type = payload['type']?.toString().trim() ?? '';
|
||||
final properties = _asMap(payload['properties']);
|
||||
if (properties['sessionID']?.toString().trim() !=
|
||||
remoteSessionId) {
|
||||
return;
|
||||
}
|
||||
if (type == 'session.status') {
|
||||
final status = _asMap(properties['status']);
|
||||
final statusType = status['type']?.toString().trim() ?? '';
|
||||
if (statusType == 'busy') {
|
||||
busySeen = true;
|
||||
}
|
||||
if (statusType == 'idle' && busySeen) {
|
||||
completeSuccess();
|
||||
}
|
||||
return;
|
||||
}
|
||||
if (type == 'session.idle' && busySeen) {
|
||||
completeSuccess();
|
||||
return;
|
||||
}
|
||||
if (type == 'session.error' && !completion.isCompleted) {
|
||||
final error = _asMap(properties['error']);
|
||||
completion.complete(
|
||||
DirectSingleAgentRunResult(
|
||||
success: false,
|
||||
output: output.toString(),
|
||||
errorMessage:
|
||||
error['message']?.toString() ??
|
||||
error['name']?.toString() ??
|
||||
'OpenCode session failed.',
|
||||
aborted: _abortedSessions.contains(normalizedSessionId),
|
||||
resolvedModel: request.model,
|
||||
),
|
||||
);
|
||||
return;
|
||||
}
|
||||
if (type == 'message.updated') {
|
||||
final info = _asMap(properties['info']);
|
||||
if (info['role']?.toString().trim() == 'assistant') {
|
||||
activeAssistantMessageId = info['id']?.toString().trim();
|
||||
}
|
||||
return;
|
||||
}
|
||||
if (type == 'message.part.delta') {
|
||||
final part = _asMap(properties['part']);
|
||||
if (activeAssistantMessageId != null &&
|
||||
part['messageID']?.toString().trim() ==
|
||||
activeAssistantMessageId) {
|
||||
final delta = properties['text']?.toString() ??
|
||||
properties['delta']?.toString() ??
|
||||
'';
|
||||
if (delta.isNotEmpty) {
|
||||
output.write(delta);
|
||||
request.onOutput?.call(delta);
|
||||
}
|
||||
}
|
||||
return;
|
||||
}
|
||||
if (type == 'message.part.updated') {
|
||||
final part = _asMap(properties['part']);
|
||||
if (activeAssistantMessageId != null &&
|
||||
part['messageID']?.toString().trim() ==
|
||||
activeAssistantMessageId &&
|
||||
part['type']?.toString().trim() == 'text') {
|
||||
lastAssistantText = part['text']?.toString();
|
||||
if ((lastAssistantText?.trim().isNotEmpty ?? false)) {
|
||||
completeSuccess();
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
onError: (Object error, StackTrace stackTrace) {
|
||||
// OpenCode event streams can disconnect independently from the
|
||||
// backing session lifecycle. Keep polling session state instead.
|
||||
},
|
||||
onDone: () {},
|
||||
cancelOnError: true,
|
||||
);
|
||||
|
||||
await _postJson(
|
||||
_buildRestUri(
|
||||
base,
|
||||
'/session/$remoteSessionId/message',
|
||||
queryParameters: <String, String>{
|
||||
'directory': request.workingDirectory,
|
||||
},
|
||||
),
|
||||
body: <String, dynamic>{
|
||||
'agent': 'build',
|
||||
'parts': <Map<String, dynamic>>[
|
||||
<String, dynamic>{'type': 'text', 'text': request.prompt},
|
||||
],
|
||||
},
|
||||
gatewayToken: request.gatewayToken,
|
||||
);
|
||||
unawaited(
|
||||
_pollRestAssistantMessage(
|
||||
base,
|
||||
remoteSessionId: remoteSessionId,
|
||||
workingDirectory: request.workingDirectory,
|
||||
gatewayToken: request.gatewayToken,
|
||||
onResolved: (text) {
|
||||
if (text.trim().isNotEmpty) {
|
||||
lastAssistantText = text;
|
||||
if (output.toString().trim().isEmpty) {
|
||||
output.write(text);
|
||||
request.onOutput?.call(text);
|
||||
}
|
||||
completeSuccess();
|
||||
}
|
||||
},
|
||||
onError: (message) {
|
||||
if (!completion.isCompleted) {
|
||||
completion.complete(
|
||||
DirectSingleAgentRunResult(
|
||||
success: false,
|
||||
output: output.toString(),
|
||||
errorMessage: message,
|
||||
aborted: _abortedSessions.contains(normalizedSessionId),
|
||||
resolvedModel: request.model,
|
||||
),
|
||||
);
|
||||
}
|
||||
},
|
||||
),
|
||||
);
|
||||
|
||||
return await completion.future.timeout(
|
||||
const Duration(minutes: 10),
|
||||
onTimeout: () => DirectSingleAgentRunResult(
|
||||
success: false,
|
||||
output: output.toString(),
|
||||
errorMessage: 'OpenCode REST request timed out.',
|
||||
aborted: _abortedSessions.contains(normalizedSessionId),
|
||||
resolvedModel: request.model,
|
||||
),
|
||||
);
|
||||
} catch (error) {
|
||||
return DirectSingleAgentRunResult(
|
||||
success: false,
|
||||
output: output.toString(),
|
||||
errorMessage: error.toString(),
|
||||
aborted: _abortedSessions.contains(normalizedSessionId),
|
||||
resolvedModel: request.model,
|
||||
);
|
||||
} finally {
|
||||
unawaited(lineSubscription?.cancel());
|
||||
eventClient.close(force: true);
|
||||
_abortedSessions.remove(normalizedSessionId);
|
||||
}
|
||||
}
|
||||
|
||||
Future<String> _ensureRestSession(
|
||||
Uri base, {
|
||||
required String sessionId,
|
||||
required String workingDirectory,
|
||||
required String gatewayToken,
|
||||
}) async {
|
||||
final existing = _restSessionIds[sessionId]?.trim() ?? '';
|
||||
if (existing.isNotEmpty) {
|
||||
return existing;
|
||||
}
|
||||
final created = await _postJson(
|
||||
_buildRestUri(
|
||||
base,
|
||||
'/session',
|
||||
queryParameters: <String, String>{'directory': workingDirectory},
|
||||
),
|
||||
body: <String, dynamic>{'title': sessionId},
|
||||
gatewayToken: gatewayToken,
|
||||
);
|
||||
final createdId = created['id']?.toString().trim() ?? '';
|
||||
if (createdId.isEmpty) {
|
||||
throw StateError('OpenCode REST endpoint returned an empty session id.');
|
||||
}
|
||||
_restSessionIds[sessionId] = createdId;
|
||||
return createdId;
|
||||
}
|
||||
|
||||
Future<void> _pollRestAssistantMessage(
|
||||
Uri base, {
|
||||
required String remoteSessionId,
|
||||
required String workingDirectory,
|
||||
required String gatewayToken,
|
||||
required void Function(String text) onResolved,
|
||||
required void Function(String message) onError,
|
||||
}) async {
|
||||
String? previousText;
|
||||
var stableCount = 0;
|
||||
for (var attempt = 0; attempt < 100; attempt++) {
|
||||
try {
|
||||
final items = await _fetchJsonList(
|
||||
_buildRestUri(
|
||||
base,
|
||||
'/session/$remoteSessionId/message',
|
||||
queryParameters: <String, String>{
|
||||
'directory': workingDirectory,
|
||||
'limit': '20',
|
||||
},
|
||||
),
|
||||
gatewayToken: gatewayToken,
|
||||
);
|
||||
final text = _latestAssistantTextFromRestMessages(items);
|
||||
if (text.trim().isNotEmpty) {
|
||||
if (text == previousText) {
|
||||
stableCount += 1;
|
||||
} else {
|
||||
previousText = text;
|
||||
stableCount = 1;
|
||||
}
|
||||
if (stableCount >= 2) {
|
||||
onResolved(text);
|
||||
return;
|
||||
}
|
||||
}
|
||||
} catch (error) {
|
||||
onError(error.toString());
|
||||
return;
|
||||
}
|
||||
await Future<void>.delayed(const Duration(milliseconds: 200));
|
||||
}
|
||||
}
|
||||
|
||||
String _latestAssistantTextFromRestMessages(List<Object?> items) {
|
||||
for (final raw in items.reversed) {
|
||||
final item = _asMap(raw);
|
||||
final info = _asMap(item['info']);
|
||||
if (info['role']?.toString().trim() != 'assistant') {
|
||||
continue;
|
||||
}
|
||||
final parts = item['parts'];
|
||||
if (parts is! List) {
|
||||
continue;
|
||||
}
|
||||
for (final rawPart in parts) {
|
||||
final part = _asMap(rawPart);
|
||||
if (part['type']?.toString().trim() == 'text') {
|
||||
final text = part['text']?.toString() ?? '';
|
||||
if (text.trim().isNotEmpty) {
|
||||
return text;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
return '';
|
||||
}
|
||||
|
||||
bool _usesRestSessionApi(SingleAgentProvider provider) {
|
||||
if (provider.providerId != SingleAgentProvider.opencode.providerId) {
|
||||
return false;
|
||||
}
|
||||
final base = endpointResolver(provider);
|
||||
final scheme = base?.scheme.toLowerCase() ?? '';
|
||||
return scheme == 'http' || scheme == 'https';
|
||||
}
|
||||
|
||||
Uri _buildRestUri(
|
||||
Uri base,
|
||||
String path, {
|
||||
Map<String, String>? queryParameters,
|
||||
}) {
|
||||
final normalizedPath = path.startsWith('/') ? path : '/$path';
|
||||
return base.replace(
|
||||
path: normalizedPath,
|
||||
queryParameters: queryParameters,
|
||||
fragment: null,
|
||||
);
|
||||
}
|
||||
|
||||
Future<Map<String, dynamic>> _fetchJson(
|
||||
Uri uri, {
|
||||
required String gatewayToken,
|
||||
}) async {
|
||||
final client = HttpClient()..connectionTimeout = const Duration(seconds: 8);
|
||||
try {
|
||||
final request = await client.getUrl(uri);
|
||||
final normalizedToken = gatewayToken.trim();
|
||||
if (normalizedToken.isNotEmpty) {
|
||||
request.headers.set(
|
||||
HttpHeaders.authorizationHeader,
|
||||
'Bearer $normalizedToken',
|
||||
);
|
||||
}
|
||||
final response = await request.close();
|
||||
final body = await response.transform(utf8.decoder).join();
|
||||
return _decodeMap(body);
|
||||
} finally {
|
||||
client.close(force: true);
|
||||
}
|
||||
}
|
||||
|
||||
Future<Map<String, dynamic>> _postJson(
|
||||
Uri uri, {
|
||||
required Object? body,
|
||||
required String gatewayToken,
|
||||
}) async {
|
||||
final client = HttpClient()..connectionTimeout = const Duration(seconds: 8);
|
||||
try {
|
||||
final request = await client.postUrl(uri);
|
||||
request.headers.set(
|
||||
HttpHeaders.contentTypeHeader,
|
||||
'application/json; charset=utf-8',
|
||||
);
|
||||
final normalizedToken = gatewayToken.trim();
|
||||
if (normalizedToken.isNotEmpty) {
|
||||
request.headers.set(
|
||||
HttpHeaders.authorizationHeader,
|
||||
'Bearer $normalizedToken',
|
||||
);
|
||||
}
|
||||
if (body != null) {
|
||||
request.add(utf8.encode(jsonEncode(body)));
|
||||
}
|
||||
final response = await request.close();
|
||||
final text = await response.transform(utf8.decoder).join();
|
||||
if (text.trim().isEmpty) {
|
||||
return const <String, dynamic>{};
|
||||
}
|
||||
return _decodeMap(text);
|
||||
} finally {
|
||||
client.close(force: true);
|
||||
}
|
||||
}
|
||||
|
||||
Future<List<Object?>> _fetchJsonList(
|
||||
Uri uri, {
|
||||
required String gatewayToken,
|
||||
}) async {
|
||||
final client = HttpClient()..connectionTimeout = const Duration(seconds: 8);
|
||||
try {
|
||||
final request = await client.getUrl(uri);
|
||||
final normalizedToken = gatewayToken.trim();
|
||||
if (normalizedToken.isNotEmpty) {
|
||||
request.headers.set(
|
||||
HttpHeaders.authorizationHeader,
|
||||
'Bearer $normalizedToken',
|
||||
);
|
||||
}
|
||||
final response = await request.close();
|
||||
final body = await response.transform(utf8.decoder).join();
|
||||
final decoded = jsonDecode(body);
|
||||
if (decoded is List<Object?>) {
|
||||
return decoded;
|
||||
}
|
||||
if (decoded is List) {
|
||||
return decoded.cast<Object?>();
|
||||
}
|
||||
return const <Object?>[];
|
||||
} finally {
|
||||
client.close(force: true);
|
||||
}
|
||||
}
|
||||
|
||||
String? _extractThreadId(Map<String, dynamic> payload) {
|
||||
final topLevelId = payload['id']?.toString().trim() ?? '';
|
||||
if (topLevelId.isNotEmpty) {
|
||||
|
||||
@ -50,7 +50,7 @@ void main() {
|
||||
).copyWith(onOutput: deltas.add),
|
||||
);
|
||||
|
||||
expect(result.success, isTrue);
|
||||
expect(result.success, isTrue, reason: result.errorMessage);
|
||||
expect(result.output, 'hello world from app server');
|
||||
expect(result.resolvedModel, 'codex-sonnet');
|
||||
expect(server.lastTurnInput, <Object?>[
|
||||
@ -175,6 +175,54 @@ void main() {
|
||||
expect(result.resolvedModel, 'codex-sonnet');
|
||||
},
|
||||
);
|
||||
|
||||
test('probes OpenCode REST endpoint and reports provider support', () async {
|
||||
final server = await _FakeOpenCodeRestServer.start();
|
||||
addTearDown(server.close);
|
||||
|
||||
final client = DirectSingleAgentAppServerClient(
|
||||
endpointResolver: (_) => server.baseHttpUri,
|
||||
);
|
||||
|
||||
final capabilities = await client.loadCapabilities(
|
||||
provider: SingleAgentProvider.opencode,
|
||||
);
|
||||
|
||||
expect(capabilities.available, isTrue);
|
||||
expect(
|
||||
capabilities.supportsProvider(SingleAgentProvider.opencode),
|
||||
isTrue,
|
||||
);
|
||||
expect(server.healthRequested, isTrue);
|
||||
});
|
||||
|
||||
test('runs OpenCode turns over REST session api', () async {
|
||||
final server = await _FakeOpenCodeRestServer.start();
|
||||
addTearDown(server.close);
|
||||
|
||||
final client = DirectSingleAgentAppServerClient(
|
||||
endpointResolver: (_) => server.baseHttpUri,
|
||||
);
|
||||
addTearDown(client.dispose);
|
||||
|
||||
final deltas = <String>[];
|
||||
final result = await client.run(
|
||||
const DirectSingleAgentRunRequest(
|
||||
sessionId: 'session-opencode',
|
||||
provider: SingleAgentProvider.opencode,
|
||||
prompt: 'hello opencode',
|
||||
model: '',
|
||||
workingDirectory: '/tmp',
|
||||
gatewayToken: '',
|
||||
).copyWith(onOutput: deltas.add),
|
||||
);
|
||||
|
||||
expect(result.success, isTrue);
|
||||
expect(result.output, 'hello world from opencode');
|
||||
expect(deltas.join(), 'hello world from opencode');
|
||||
expect(server.createdSessionCount, 1);
|
||||
expect(server.lastPromptText, 'hello opencode');
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
@ -424,6 +472,211 @@ class _FakeAppServer {
|
||||
}
|
||||
}
|
||||
|
||||
class _FakeOpenCodeRestServer {
|
||||
_FakeOpenCodeRestServer._(this._server);
|
||||
|
||||
final HttpServer _server;
|
||||
final List<HttpResponse> _eventResponses = <HttpResponse>[];
|
||||
var _sessionCounter = 0;
|
||||
var _messageCounter = 0;
|
||||
bool healthRequested = false;
|
||||
int createdSessionCount = 0;
|
||||
String lastPromptText = '';
|
||||
final Map<String, String> _assistantTextBySession = <String, String>{};
|
||||
|
||||
Uri get baseHttpUri => Uri.parse('http://127.0.0.1:${_server.port}');
|
||||
|
||||
static Future<_FakeOpenCodeRestServer> start() async {
|
||||
final server = await HttpServer.bind(InternetAddress.loopbackIPv4, 0);
|
||||
final fake = _FakeOpenCodeRestServer._(server);
|
||||
unawaited(fake._listen());
|
||||
return fake;
|
||||
}
|
||||
|
||||
Future<void> close() async {
|
||||
for (final response in _eventResponses.toList(growable: false)) {
|
||||
try {
|
||||
await response.close();
|
||||
} catch (_) {
|
||||
// Best effort.
|
||||
}
|
||||
}
|
||||
await _server.close(force: true);
|
||||
}
|
||||
|
||||
Future<void> _listen() async {
|
||||
await for (final request in _server) {
|
||||
if (request.uri.path == '/global/health') {
|
||||
healthRequested = true;
|
||||
request.response.headers.contentType = ContentType.json;
|
||||
request.response.write(
|
||||
jsonEncode(<String, dynamic>{'healthy': true, 'version': '1.3.3'}),
|
||||
);
|
||||
await request.response.close();
|
||||
continue;
|
||||
}
|
||||
if (request.uri.path == '/global/event') {
|
||||
request.response.headers.set(
|
||||
HttpHeaders.contentTypeHeader,
|
||||
'text/event-stream',
|
||||
);
|
||||
request.response.headers.set(HttpHeaders.cacheControlHeader, 'no-cache');
|
||||
request.response.write(
|
||||
'data: ${jsonEncode(<String, dynamic>{'payload': <String, dynamic>{'type': 'server.connected', 'properties': <String, dynamic>{}}})}\n\n',
|
||||
);
|
||||
await request.response.flush();
|
||||
_eventResponses.add(request.response);
|
||||
continue;
|
||||
}
|
||||
if (request.uri.path == '/session' && request.method == 'POST') {
|
||||
createdSessionCount += 1;
|
||||
final sessionId = 'ses-${_sessionCounter++}';
|
||||
request.response.headers.contentType = ContentType.json;
|
||||
request.response.write(
|
||||
jsonEncode(<String, dynamic>{
|
||||
'id': sessionId,
|
||||
'title': 'test',
|
||||
'directory':
|
||||
request.uri.queryParameters['directory'] ?? Directory.current.path,
|
||||
}),
|
||||
);
|
||||
await request.response.close();
|
||||
continue;
|
||||
}
|
||||
final sessionMatch = RegExp(r'^/session/([^/]+)/message$').firstMatch(
|
||||
request.uri.path,
|
||||
);
|
||||
if (sessionMatch != null && request.method == 'GET') {
|
||||
final sessionId = sessionMatch.group(1)!;
|
||||
final text = _assistantTextBySession[sessionId] ?? '';
|
||||
request.response.headers.contentType = ContentType.json;
|
||||
request.response.write(
|
||||
jsonEncode(<Map<String, dynamic>>[
|
||||
<String, dynamic>{
|
||||
'info': <String, dynamic>{'id': 'msg-user', 'role': 'user'},
|
||||
'parts': <Map<String, dynamic>>[
|
||||
<String, dynamic>{'type': 'text', 'text': lastPromptText},
|
||||
],
|
||||
},
|
||||
if (text.isNotEmpty)
|
||||
<String, dynamic>{
|
||||
'info': <String, dynamic>{
|
||||
'id': 'msg-assistant',
|
||||
'role': 'assistant',
|
||||
},
|
||||
'parts': <Map<String, dynamic>>[
|
||||
<String, dynamic>{'type': 'text', 'text': text},
|
||||
],
|
||||
},
|
||||
]),
|
||||
);
|
||||
await request.response.close();
|
||||
continue;
|
||||
}
|
||||
if (sessionMatch != null && request.method == 'POST') {
|
||||
final sessionId = sessionMatch.group(1)!;
|
||||
final body = jsonDecode(await utf8.decodeStream(request));
|
||||
final parts = (body as Map<String, dynamic>)['parts'] as List<dynamic>? ??
|
||||
const <dynamic>[];
|
||||
if (parts.isNotEmpty) {
|
||||
lastPromptText =
|
||||
(parts.first as Map<String, dynamic>)['text']?.toString() ?? '';
|
||||
}
|
||||
final assistantMessageId = 'msg-assistant-${_messageCounter++}';
|
||||
await _broadcastEvent(
|
||||
<String, dynamic>{
|
||||
'payload': <String, dynamic>{
|
||||
'type': 'session.status',
|
||||
'properties': <String, dynamic>{
|
||||
'sessionID': sessionId,
|
||||
'status': <String, dynamic>{'type': 'busy'},
|
||||
},
|
||||
},
|
||||
},
|
||||
);
|
||||
await _broadcastEvent(
|
||||
<String, dynamic>{
|
||||
'payload': <String, dynamic>{
|
||||
'type': 'message.updated',
|
||||
'properties': <String, dynamic>{
|
||||
'sessionID': sessionId,
|
||||
'info': <String, dynamic>{
|
||||
'id': assistantMessageId,
|
||||
'role': 'assistant',
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
);
|
||||
for (final delta in <String>['hello ', 'world ', 'from ', 'opencode']) {
|
||||
await _broadcastEvent(
|
||||
<String, dynamic>{
|
||||
'payload': <String, dynamic>{
|
||||
'type': 'message.part.delta',
|
||||
'properties': <String, dynamic>{
|
||||
'sessionID': sessionId,
|
||||
'part': <String, dynamic>{'messageID': assistantMessageId},
|
||||
'text': delta,
|
||||
},
|
||||
},
|
||||
},
|
||||
);
|
||||
}
|
||||
await _broadcastEvent(
|
||||
<String, dynamic>{
|
||||
'payload': <String, dynamic>{
|
||||
'type': 'message.part.updated',
|
||||
'properties': <String, dynamic>{
|
||||
'sessionID': sessionId,
|
||||
'part': <String, dynamic>{
|
||||
'messageID': assistantMessageId,
|
||||
'type': 'text',
|
||||
'text': 'hello world from opencode',
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
);
|
||||
_assistantTextBySession[sessionId] = 'hello world from opencode';
|
||||
await _broadcastEvent(
|
||||
<String, dynamic>{
|
||||
'payload': <String, dynamic>{
|
||||
'type': 'session.status',
|
||||
'properties': <String, dynamic>{
|
||||
'sessionID': sessionId,
|
||||
'status': <String, dynamic>{'type': 'idle'},
|
||||
},
|
||||
},
|
||||
},
|
||||
);
|
||||
request.response.headers.contentType = ContentType.json;
|
||||
request.response.write('');
|
||||
await request.response.close();
|
||||
continue;
|
||||
}
|
||||
final abortMatch = RegExp(r'^/session/([^/]+)/abort$').firstMatch(
|
||||
request.uri.path,
|
||||
);
|
||||
if (abortMatch != null && request.method == 'POST') {
|
||||
request.response.headers.contentType = ContentType.json;
|
||||
request.response.write('{}');
|
||||
await request.response.close();
|
||||
continue;
|
||||
}
|
||||
request.response.statusCode = HttpStatus.notFound;
|
||||
await request.response.close();
|
||||
}
|
||||
}
|
||||
|
||||
Future<void> _broadcastEvent(Map<String, dynamic> event) async {
|
||||
final payload = 'data: ${jsonEncode(event)}\n\n';
|
||||
for (final response in _eventResponses.toList(growable: false)) {
|
||||
response.write(payload);
|
||||
await response.flush();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Map<String, dynamic> _decodeMap(Object raw) {
|
||||
if (raw is Map<String, dynamic>) {
|
||||
return raw;
|
||||
|
||||
Loading…
Reference in New Issue
Block a user