feat: implement tea/gosmee issue source for #15
This commit is contained in:
parent
089b7bffc1
commit
61b1bdca63
|
|
@ -50,6 +50,11 @@ class _CodeWorkSpawnerCommandRunner extends CommandRunner<void> {
|
|||
'tea-command',
|
||||
help: 'Path to the tea executable.',
|
||||
defaultsTo: 'tea',
|
||||
)
|
||||
..addOption(
|
||||
'gosmee-command',
|
||||
help: 'Path to the gosmee executable.',
|
||||
defaultsTo: 'gosmee',
|
||||
);
|
||||
|
||||
addCommand(_InitConfigCommand());
|
||||
|
|
@ -88,6 +93,7 @@ class _CodeWorkSpawnerCommandRunner extends CommandRunner<void> {
|
|||
config: config,
|
||||
databasePath: topLevelResults['db'] as String,
|
||||
ghCommand: topLevelResults['gh-command'] as String,
|
||||
gosmeeCommand: topLevelResults['gosmee-command'] as String,
|
||||
teaCommand: topLevelResults['tea-command'] as String,
|
||||
);
|
||||
|
||||
|
|
|
|||
|
|
@ -8,11 +8,14 @@ event source.
|
|||
|
||||
GitHub projects default to `gh/polling`, and Gitea projects default to
|
||||
`tea/polling`. GitHub projects can switch their long-running runtime to
|
||||
`gh/webhook-forward`, which starts
|
||||
`gh/webhook-forward`, and Gitea projects can switch to `tea/gosmee`.
|
||||
`gh/webhook-forward` starts
|
||||
`gh webhook forward` for `issues` and `issue_comment` events while keeping
|
||||
periodic polling reconciliation for startup sync and missed-event recovery.
|
||||
`--once` still uses a single polling reconciliation cycle so existing backlog
|
||||
handling keeps working.
|
||||
`tea/gosmee` creates a temporary Gitea webhook that targets a generated gosmee
|
||||
URL, then keeps the same polling reconciliation path active for startup sync
|
||||
and missed-event recovery. `--once` still uses a single polling reconciliation
|
||||
cycle so existing backlog handling keeps working.
|
||||
|
||||
Generated comments include both a visible prefix and footer by default so it is
|
||||
clear the reply came from automation even when `gh` or `tea` is authenticated
|
||||
|
|
|
|||
|
|
@ -7,6 +7,9 @@ worktreeDir: ~/.worktrees
|
|||
defaults:
|
||||
issueAssistant:
|
||||
enabled: true
|
||||
# Use tea/gosmee for long-running runs while keeping --once and missed-event
|
||||
# recovery on polling reconciliation.
|
||||
# eventSource: tea/gosmee
|
||||
pollInterval: 30s
|
||||
maxConcurrentIssues: 3
|
||||
mentionTriggers: ["@codex", "@helper"]
|
||||
|
|
|
|||
|
|
@ -218,10 +218,12 @@ class ProjectConfig {
|
|||
);
|
||||
}
|
||||
case IssueAssistantEventSourceKind.teaGosmee:
|
||||
throw FormatException(
|
||||
'projects.$key.issueAssistant.eventSource tea/gosmee '
|
||||
'is not implemented yet.',
|
||||
);
|
||||
if (provider != IssueTrackerProvider.gitea) {
|
||||
throw FormatException(
|
||||
'projects.$key.issueAssistant.eventSource tea/gosmee '
|
||||
'requires provider: gitea.',
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
return ProjectConfig(
|
||||
|
|
|
|||
|
|
@ -8,6 +8,7 @@ import 'cli_responder.dart';
|
|||
import 'config.dart';
|
||||
import 'database.dart';
|
||||
import 'issue_event_source.dart';
|
||||
import 'issue_event_source_gitea_gosmee.dart';
|
||||
import 'issue_event_source_github_webhook_forward.dart';
|
||||
import 'issue_tracker_client.dart';
|
||||
import 'notifier.dart';
|
||||
|
|
@ -40,6 +41,7 @@ class IssueAssistantApp {
|
|||
required AppConfig config,
|
||||
required String databasePath,
|
||||
String ghCommand = 'gh',
|
||||
String gosmeeCommand = 'gosmee',
|
||||
String teaCommand = 'tea',
|
||||
}) async {
|
||||
final database = AppDatabase.open(databasePath);
|
||||
|
|
@ -65,6 +67,11 @@ class IssueAssistantApp {
|
|||
issueTrackerClient: issueTrackerClient,
|
||||
logger: _logger,
|
||||
),
|
||||
IssueAssistantEventSourceKind.teaGosmee: GiteaGosmeeIssueEventSource(
|
||||
issueTrackerClient: issueTrackerClient,
|
||||
gosmeeCommand: gosmeeCommand,
|
||||
logger: _logger,
|
||||
),
|
||||
},
|
||||
),
|
||||
responderRunner: CliResponderRunner(),
|
||||
|
|
|
|||
|
|
@ -110,6 +110,10 @@ class RoutedIssueEventSource implements IssueEventSource {
|
|||
yield IssueAssistantEventSourceKind.ghPolling;
|
||||
yield IssueAssistantEventSourceKind.ghWebhookForward;
|
||||
case IssueAssistantEventSourceKind.teaGosmee:
|
||||
// Keep reconciliation polling active for startup sync and missed
|
||||
// webhook recovery while gosmee handles low-latency issue/comment
|
||||
// events for Gitea projects.
|
||||
yield IssueAssistantEventSourceKind.teaPolling;
|
||||
yield IssueAssistantEventSourceKind.teaGosmee;
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -0,0 +1,397 @@
|
|||
import 'dart:async';
|
||||
import 'dart:convert';
|
||||
import 'dart:io';
|
||||
|
||||
import 'package:github/github.dart';
|
||||
import 'package:logging/logging.dart';
|
||||
|
||||
import 'config.dart';
|
||||
import 'issue_event_source.dart';
|
||||
import 'issue_tracker_client.dart';
|
||||
|
||||
class GiteaGosmeeIssueEventSource implements IssueEventSource {
|
||||
GiteaGosmeeIssueEventSource({
|
||||
required this.issueTrackerClient,
|
||||
required this.gosmeeCommand,
|
||||
required Logger logger,
|
||||
}) : _logger = logger;
|
||||
|
||||
static const Duration _webhookCoalescingWindow = Duration(milliseconds: 500);
|
||||
|
||||
final IssueTrackerClient issueTrackerClient;
|
||||
final String gosmeeCommand;
|
||||
final Logger _logger;
|
||||
final Map<String, _QueuedGiteaIssue> _queuedIssues = {};
|
||||
final Set<Future<void>> _activeDispatches = <Future<void>>{};
|
||||
final Map<String, Process> _gosmeeProcesses = {};
|
||||
final Map<String, int> _webhookIdsByProject = {};
|
||||
final Map<String, RepositorySlug> _repoSlugsByProject = {};
|
||||
|
||||
HttpServer? _server;
|
||||
Timer? _flushTimer;
|
||||
Completer<void>? _runCompleter;
|
||||
bool _isClosing = false;
|
||||
bool _isClosed = false;
|
||||
|
||||
@override
|
||||
Future<void> run({
|
||||
required List<ProjectConfig> projects,
|
||||
required IssueEventBatchHandler onBatch,
|
||||
}) async {
|
||||
if (_isClosed) {
|
||||
throw StateError('IssueEventSource is already closed.');
|
||||
}
|
||||
if (_runCompleter != null) {
|
||||
return _runCompleter!.future;
|
||||
}
|
||||
|
||||
final giteaProjects = projects
|
||||
.where((project) => project.provider == IssueTrackerProvider.gitea)
|
||||
.toList(growable: false);
|
||||
if (giteaProjects.isEmpty) {
|
||||
return;
|
||||
}
|
||||
|
||||
final completer = Completer<void>();
|
||||
_runCompleter = completer;
|
||||
final server = await HttpServer.bind(InternetAddress.loopbackIPv4, 0);
|
||||
_server = server;
|
||||
final repoToProject = {
|
||||
for (final project in giteaProjects) project.repo: project,
|
||||
};
|
||||
for (final project in giteaProjects) {
|
||||
_repoSlugsByProject[project.key] = project.repoSlug;
|
||||
}
|
||||
server.listen(
|
||||
(request) => unawaited(
|
||||
_handleWebhookRequest(
|
||||
request: request,
|
||||
repoToProject: repoToProject,
|
||||
onBatch: onBatch,
|
||||
),
|
||||
),
|
||||
onError: (Object error, StackTrace stackTrace) {
|
||||
_logError('gitea/gosmee server error=$error');
|
||||
if (!_isClosing && !completer.isCompleted) {
|
||||
completer.completeError(error, stackTrace);
|
||||
}
|
||||
},
|
||||
);
|
||||
|
||||
final localUrl =
|
||||
'http://${server.address.address}:${server.port}/gitea/webhooks';
|
||||
for (final project in giteaProjects) {
|
||||
final publicUrl = await _createGosmeeChannelUrl(localUrl);
|
||||
final webhookId = await issueTrackerClient.createGiteaIssueWebhook(
|
||||
repo: project.repoSlug,
|
||||
url: publicUrl,
|
||||
);
|
||||
_webhookIdsByProject[project.key] = webhookId;
|
||||
final process = await Process.start(gosmeeCommand, [
|
||||
'client',
|
||||
publicUrl,
|
||||
localUrl,
|
||||
]);
|
||||
_gosmeeProcesses[project.key] = process;
|
||||
_pipeProcessLogs(
|
||||
project: project,
|
||||
stream: process.stdout,
|
||||
level: 'stdout',
|
||||
);
|
||||
_pipeProcessLogs(
|
||||
project: project,
|
||||
stream: process.stderr,
|
||||
level: 'stderr',
|
||||
);
|
||||
unawaited(
|
||||
process.exitCode.then((exitCode) {
|
||||
_gosmeeProcesses.remove(project.key);
|
||||
_log(
|
||||
'gitea/gosmee process project=${project.key} exit_code=$exitCode',
|
||||
);
|
||||
if (!_isClosing && !completer.isCompleted) {
|
||||
completer.completeError(
|
||||
StateError(
|
||||
'gosmee client exited for project=${project.key} '
|
||||
'with code $exitCode.',
|
||||
),
|
||||
);
|
||||
}
|
||||
}),
|
||||
);
|
||||
}
|
||||
|
||||
try {
|
||||
await completer.future;
|
||||
} finally {
|
||||
await close();
|
||||
}
|
||||
}
|
||||
|
||||
@override
|
||||
Future<void> runOnce({
|
||||
required List<ProjectConfig> projects,
|
||||
required IssueEventBatchHandler onBatch,
|
||||
}) async {
|
||||
throw UnsupportedError(
|
||||
'gitea/gosmee does not support runOnce; use polling instead.',
|
||||
);
|
||||
}
|
||||
|
||||
@override
|
||||
Future<void> close() async {
|
||||
if (_isClosed || _isClosing) {
|
||||
return;
|
||||
}
|
||||
|
||||
_isClosing = true;
|
||||
_flushTimer?.cancel();
|
||||
_flushTimer = null;
|
||||
final completer = _runCompleter;
|
||||
if (completer != null && !completer.isCompleted) {
|
||||
completer.complete();
|
||||
}
|
||||
_runCompleter = null;
|
||||
|
||||
final server = _server;
|
||||
_server = null;
|
||||
await server?.close(force: true);
|
||||
|
||||
final processes = _gosmeeProcesses.values.toList(growable: false);
|
||||
_gosmeeProcesses.clear();
|
||||
for (final process in processes) {
|
||||
process.kill();
|
||||
}
|
||||
for (final process in processes) {
|
||||
await process.exitCode.catchError((_) => 0);
|
||||
}
|
||||
|
||||
final webhookIds = Map<String, int>.from(_webhookIdsByProject);
|
||||
_webhookIdsByProject.clear();
|
||||
for (final entry in webhookIds.entries) {
|
||||
try {
|
||||
await issueTrackerClient.deleteGiteaWebhook(
|
||||
repo: _repoSlugsByProject[entry.key]!,
|
||||
webhookId: entry.value,
|
||||
);
|
||||
} catch (error) {
|
||||
_logError(
|
||||
'gitea/gosmee delete webhook project=${entry.key} '
|
||||
'webhook_id=${entry.value} error=$error',
|
||||
);
|
||||
}
|
||||
}
|
||||
_repoSlugsByProject.clear();
|
||||
|
||||
await _flushQueuedIssues();
|
||||
await Future.wait(_activeDispatches.toList(growable: false));
|
||||
|
||||
_isClosed = true;
|
||||
_isClosing = false;
|
||||
}
|
||||
|
||||
Future<void> _handleWebhookRequest({
|
||||
required HttpRequest request,
|
||||
required Map<String, ProjectConfig> repoToProject,
|
||||
required IssueEventBatchHandler onBatch,
|
||||
}) async {
|
||||
if (request.method != 'POST') {
|
||||
request.response
|
||||
..statusCode = HttpStatus.methodNotAllowed
|
||||
..write('Only POST is supported.');
|
||||
await request.response.close();
|
||||
return;
|
||||
}
|
||||
|
||||
try {
|
||||
final payloadText = await utf8.decoder.bind(request).join();
|
||||
final payload = jsonDecode(payloadText);
|
||||
if (payload is! Map<String, dynamic>) {
|
||||
throw const FormatException('Webhook payload must be a JSON object.');
|
||||
}
|
||||
final repository = payload['repository'];
|
||||
final fullName = repository is Map<String, dynamic>
|
||||
? repository['full_name']?.toString()
|
||||
: null;
|
||||
final project = fullName == null ? null : repoToProject[fullName];
|
||||
final issueNode = payload['issue'];
|
||||
if (project != null &&
|
||||
issueNode is Map<String, dynamic> &&
|
||||
(issueNode['pull_request'] == null)) {
|
||||
final issue = GitHubIssueSummary.fromRepositoryJson(
|
||||
project.repoSlug,
|
||||
issueNode,
|
||||
);
|
||||
if (issue.state == 'open') {
|
||||
_queuedIssues['${project.key}#${issue.number}'] = _QueuedGiteaIssue(
|
||||
project: project,
|
||||
issue: issue,
|
||||
onBatch: onBatch,
|
||||
);
|
||||
_scheduleFlush();
|
||||
}
|
||||
}
|
||||
|
||||
request.response.statusCode = HttpStatus.accepted;
|
||||
await request.response.close();
|
||||
} catch (error) {
|
||||
_logError('gitea/gosmee payload error=$error');
|
||||
request.response
|
||||
..statusCode = HttpStatus.badRequest
|
||||
..write('Invalid webhook payload.');
|
||||
await request.response.close();
|
||||
}
|
||||
}
|
||||
|
||||
Future<String> _createGosmeeChannelUrl(String localUrl) async {
|
||||
final result = await Process.run(gosmeeCommand, [
|
||||
'--output',
|
||||
'json',
|
||||
'client',
|
||||
'--new-url',
|
||||
localUrl,
|
||||
]);
|
||||
if (result.exitCode != 0) {
|
||||
throw ProcessException(
|
||||
gosmeeCommand,
|
||||
['--output', 'json', 'client', '--new-url', localUrl],
|
||||
'${result.stdout}\n${result.stderr}',
|
||||
result.exitCode,
|
||||
);
|
||||
}
|
||||
final stdoutText = result.stdout.toString().trim();
|
||||
if (stdoutText.isEmpty) {
|
||||
throw const FormatException('gosmee did not print a public URL.');
|
||||
}
|
||||
try {
|
||||
final decodedJson = jsonDecode(stdoutText);
|
||||
if (decodedJson is Map<String, dynamic>) {
|
||||
final publicUrl = decodedJson['url']?.toString().trim();
|
||||
if (publicUrl == null || publicUrl.isEmpty) {
|
||||
throw const FormatException(
|
||||
'gosmee JSON output did not contain a valid public URL.',
|
||||
);
|
||||
}
|
||||
return publicUrl;
|
||||
}
|
||||
if (decodedJson is String) {
|
||||
final publicUrl = decodedJson.trim();
|
||||
if (publicUrl.isEmpty) {
|
||||
throw const FormatException('gosmee did not print a public URL.');
|
||||
}
|
||||
return publicUrl;
|
||||
}
|
||||
} on FormatException {
|
||||
if (Uri.tryParse(stdoutText)?.hasScheme ?? false) {
|
||||
return stdoutText;
|
||||
}
|
||||
}
|
||||
throw const FormatException('gosmee did not print a valid public URL.');
|
||||
}
|
||||
|
||||
void _scheduleFlush() {
|
||||
_flushTimer ??= Timer(_webhookCoalescingWindow, () {
|
||||
_flushTimer = null;
|
||||
unawaited(_flushQueuedIssues());
|
||||
});
|
||||
}
|
||||
|
||||
Future<void> _flushQueuedIssues() async {
|
||||
if (_queuedIssues.isEmpty) {
|
||||
return;
|
||||
}
|
||||
|
||||
final queuedIssues = _queuedIssues.values.toList(growable: false);
|
||||
_queuedIssues.clear();
|
||||
final groupedIssues = <String, _QueuedGiteaBatch>{};
|
||||
for (final queuedIssue in queuedIssues) {
|
||||
groupedIssues.update(
|
||||
queuedIssue.project.key,
|
||||
(existing) => existing.add(queuedIssue.issue),
|
||||
ifAbsent: () => _QueuedGiteaBatch(
|
||||
project: queuedIssue.project,
|
||||
onBatch: queuedIssue.onBatch,
|
||||
issues: [queuedIssue.issue],
|
||||
),
|
||||
);
|
||||
}
|
||||
|
||||
late final Future<void> dispatchFuture;
|
||||
dispatchFuture = () async {
|
||||
for (final batch in groupedIssues.values) {
|
||||
batch.issues.sort(
|
||||
(left, right) => left.updatedAt.compareTo(right.updatedAt),
|
||||
);
|
||||
_log(
|
||||
'gitea/gosmee project=${batch.project.key} '
|
||||
'fetched_issues=${batch.issues.length}',
|
||||
);
|
||||
await batch.onBatch(
|
||||
IssueEventBatch(
|
||||
project: batch.project,
|
||||
issues: batch.issues,
|
||||
latestSeenUpdatedAt: batch.issues.last.updatedAt,
|
||||
sourceKind: 'gitea/gosmee',
|
||||
),
|
||||
);
|
||||
}
|
||||
}();
|
||||
|
||||
_activeDispatches.add(dispatchFuture);
|
||||
try {
|
||||
await dispatchFuture;
|
||||
} finally {
|
||||
_activeDispatches.remove(dispatchFuture);
|
||||
}
|
||||
}
|
||||
|
||||
void _pipeProcessLogs({
|
||||
required ProjectConfig project,
|
||||
required Stream<List<int>> stream,
|
||||
required String level,
|
||||
}) {
|
||||
utf8.decoder
|
||||
.bind(stream)
|
||||
.transform(const LineSplitter())
|
||||
.listen(
|
||||
(line) => _log('gitea/gosmee project=${project.key} $level=$line'),
|
||||
);
|
||||
}
|
||||
|
||||
void _log(String message) {
|
||||
_logger.info(message);
|
||||
}
|
||||
|
||||
void _logError(String message) {
|
||||
_logger.warning(message);
|
||||
}
|
||||
}
|
||||
|
||||
class _QueuedGiteaIssue {
|
||||
_QueuedGiteaIssue({
|
||||
required this.project,
|
||||
required this.issue,
|
||||
required this.onBatch,
|
||||
});
|
||||
|
||||
final ProjectConfig project;
|
||||
final GitHubIssueSummary issue;
|
||||
final IssueEventBatchHandler onBatch;
|
||||
}
|
||||
|
||||
class _QueuedGiteaBatch {
|
||||
_QueuedGiteaBatch({
|
||||
required this.project,
|
||||
required this.onBatch,
|
||||
required this.issues,
|
||||
});
|
||||
|
||||
final ProjectConfig project;
|
||||
final IssueEventBatchHandler onBatch;
|
||||
final List<GitHubIssueSummary> issues;
|
||||
|
||||
_QueuedGiteaBatch add(GitHubIssueSummary issue) {
|
||||
issues.add(issue);
|
||||
return this;
|
||||
}
|
||||
}
|
||||
|
|
@ -121,6 +121,46 @@ class IssueTrackerClient {
|
|||
};
|
||||
}
|
||||
|
||||
Future<int> createGiteaIssueWebhook({
|
||||
required RepositorySlug repo,
|
||||
required String url,
|
||||
}) async {
|
||||
final json = await _runTeaJson([
|
||||
'webhooks',
|
||||
'create',
|
||||
'--type',
|
||||
'gitea',
|
||||
'--events',
|
||||
'issues,issue_comment',
|
||||
'--active',
|
||||
'-o',
|
||||
'json',
|
||||
'-r',
|
||||
repo.fullName,
|
||||
url,
|
||||
]);
|
||||
if (json is! Map<String, dynamic>) {
|
||||
throw const FormatException(
|
||||
'tea webhooks create returned an unexpected response.',
|
||||
);
|
||||
}
|
||||
return _readInt(json, 'id');
|
||||
}
|
||||
|
||||
Future<void> deleteGiteaWebhook({
|
||||
required RepositorySlug repo,
|
||||
required int webhookId,
|
||||
}) async {
|
||||
await _runTeaJson([
|
||||
'webhooks',
|
||||
'delete',
|
||||
'--confirm',
|
||||
'-r',
|
||||
repo.fullName,
|
||||
'$webhookId',
|
||||
]);
|
||||
}
|
||||
|
||||
Future<List<GitHubIssueSummary>> _fetchUpdatedGitHubIssues({
|
||||
required Iterable<RepositorySlug> repos,
|
||||
required DateTime? since,
|
||||
|
|
|
|||
|
|
@ -430,6 +430,78 @@ projects:
|
|||
);
|
||||
});
|
||||
|
||||
/// ```gherkin
|
||||
/// Scenario: Load tea/gosmee as the issue event source
|
||||
/// Given a Gitea project config that opts into tea/gosmee
|
||||
/// When loading the app config from disk
|
||||
/// Then the project keeps tea/gosmee as its event source
|
||||
/// And the Gitea provider remains valid for that project
|
||||
/// ```
|
||||
test('Load tea/gosmee as the issue event source', () async {
|
||||
// Given a Gitea project config that opts into tea/gosmee.
|
||||
final tempDir = await Directory.systemTemp.createTemp('cws-gosmee-');
|
||||
final configFile = File(p.join(tempDir.path, 'agent-orchestrator.yaml'));
|
||||
await configFile.writeAsString('''
|
||||
projects:
|
||||
sample:
|
||||
provider: gitea
|
||||
repo: owner/sample
|
||||
path: ${tempDir.path}
|
||||
issueAssistant:
|
||||
eventSource: tea/gosmee
|
||||
''');
|
||||
|
||||
// When loading the app config from disk.
|
||||
final config = await AppConfig.load(configFile.path);
|
||||
|
||||
// Then the project keeps tea/gosmee as its event source.
|
||||
expect(
|
||||
config.projects['sample']!.issueAssistant.eventSource,
|
||||
IssueAssistantEventSourceKind.teaGosmee,
|
||||
);
|
||||
|
||||
// And the Gitea provider remains valid for that project.
|
||||
expect(config.projects['sample']!.provider, IssueTrackerProvider.gitea);
|
||||
});
|
||||
|
||||
/// ```gherkin
|
||||
/// Scenario: Reject tea/gosmee for non-Gitea providers
|
||||
/// Given a GitHub project config that opts into tea/gosmee
|
||||
/// When loading the app config from disk
|
||||
/// Then loading fails with a clear provider compatibility error
|
||||
/// ```
|
||||
test('Reject tea/gosmee for non-Gitea providers', () async {
|
||||
// Given a GitHub project config that opts into tea/gosmee.
|
||||
final tempDir = await Directory.systemTemp.createTemp(
|
||||
'cws-gosmee-github-',
|
||||
);
|
||||
final configFile = File(p.join(tempDir.path, 'agent-orchestrator.yaml'));
|
||||
await configFile.writeAsString('''
|
||||
projects:
|
||||
sample:
|
||||
provider: github
|
||||
repo: owner/sample
|
||||
path: ${tempDir.path}
|
||||
issueAssistant:
|
||||
eventSource: tea/gosmee
|
||||
''');
|
||||
|
||||
// When loading the app config from disk.
|
||||
final load = AppConfig.load(configFile.path);
|
||||
|
||||
// Then loading fails with a clear provider compatibility error.
|
||||
await expectLater(
|
||||
load,
|
||||
throwsA(
|
||||
isA<FormatException>().having(
|
||||
(error) => error.message,
|
||||
'message',
|
||||
contains('tea/gosmee requires provider: gitea'),
|
||||
),
|
||||
),
|
||||
);
|
||||
});
|
||||
|
||||
/// ```gherkin
|
||||
/// Scenario: Reject snake_case config keys
|
||||
/// Given a config that still uses snake_case keys
|
||||
|
|
|
|||
|
|
@ -34,6 +34,7 @@ void main() {
|
|||
final stdoutText = result.stdout as String;
|
||||
expect(stdoutText, contains('Available commands:'));
|
||||
expect(stdoutText, contains('init-config'));
|
||||
expect(stdoutText, contains('--gosmee-command'));
|
||||
});
|
||||
|
||||
/// ```gherkin
|
||||
|
|
|
|||
|
|
@ -0,0 +1,418 @@
|
|||
import 'dart:convert';
|
||||
import 'dart:io';
|
||||
|
||||
import 'package:code_work_spawner/code_work_spawner.dart';
|
||||
import 'package:drift/drift.dart' show driftRuntimeOptions;
|
||||
import 'package:path/path.dart' as p;
|
||||
import 'package:test/test.dart';
|
||||
|
||||
import 'test_support.dart';
|
||||
|
||||
void registerIssueAssistantAppRunGosmeeTests() {
|
||||
/// ```gherkin
|
||||
/// Feature: Gitea gosmee issue events
|
||||
///
|
||||
/// As a maintainer running the long-lived assistant for Gitea
|
||||
/// I want tea/gosmee events to trigger issue evaluation without relying only on polling
|
||||
/// So that issue and comment activity can be handled through a channel-based source
|
||||
/// ```
|
||||
group('IssueAssistantApp.run gosmee', () {
|
||||
/// ```gherkin
|
||||
/// Scenario: Reply to a mention received through tea/gosmee
|
||||
/// Given a temporary project checkout that can be used as the local repo path
|
||||
/// And a gosmee script that forwards one Gitea issue_comment webhook event
|
||||
/// And a tea script that can manage webhooks and post issue comments
|
||||
/// And a responder that emits a planning reply
|
||||
/// And an orchestrator-style config that uses tea/gosmee
|
||||
/// When the long-running app starts and receives the forwarded webhook
|
||||
/// Then it reads issue comments and posts exactly one reply for that issue
|
||||
/// And the tea webhook lifecycle commands are invoked for the Gitea project
|
||||
/// ```
|
||||
test('Reply to a mention received through tea/gosmee', () async {
|
||||
// Given a temporary project checkout that can be used as the local repo path.
|
||||
final sandbox = await Directory.systemTemp.createTemp('cws-gosmee-run-');
|
||||
final repoDir = await Directory(p.join(sandbox.path, 'repo')).create();
|
||||
await initGitRepo(repoDir);
|
||||
|
||||
// And a gosmee script that forwards one Gitea issue_comment webhook event.
|
||||
final gosmeeScript = File(p.join(sandbox.path, 'gosmee'));
|
||||
final gosmeeLog = File(p.join(sandbox.path, 'gosmee-log.jsonl'));
|
||||
final issue = {
|
||||
'index': 12,
|
||||
'title': 'Need channel-based trigger handling',
|
||||
'body': 'Please react to gosmee.',
|
||||
'state': 'open',
|
||||
'url': 'https://gitea.example.test/owner/sample/issues/12',
|
||||
'updated_at': '2026-04-08T04:00:00Z',
|
||||
'poster': {'login': 'reporter'},
|
||||
'labels': const <Map<String, Object?>>[],
|
||||
};
|
||||
final comments = [
|
||||
{
|
||||
'id': 51,
|
||||
'body': '@helper can you handle gosmee comments?',
|
||||
'url':
|
||||
'https://gitea.example.test/owner/sample/issues/12#issuecomment-51',
|
||||
'created_at': '2026-04-08T04:00:00Z',
|
||||
'updated_at': '2026-04-08T04:00:00Z',
|
||||
'poster': {'login': 'reporter'},
|
||||
},
|
||||
];
|
||||
await writeGosmeeForwardScript(
|
||||
gosmeeScript: gosmeeScript,
|
||||
publicUrl: 'https://gosmee.example.test/sample',
|
||||
payload: {
|
||||
'action': 'created',
|
||||
'issue': issue,
|
||||
'comment': comments.last,
|
||||
'repository': {'full_name': 'owner/sample'},
|
||||
},
|
||||
gosmeeLog: gosmeeLog,
|
||||
);
|
||||
|
||||
// And a tea script that can manage webhooks and post issue comments.
|
||||
final teaScript = File(p.join(sandbox.path, 'tea'));
|
||||
final teaLog = File(p.join(sandbox.path, 'tea-log.jsonl'));
|
||||
final postedBody = File(p.join(sandbox.path, 'posted-body.md'));
|
||||
await teaScript.writeAsString('''#!/usr/bin/env bash
|
||||
set -euo pipefail
|
||||
printf '%s\n' "\$*" >> "${teaLog.path}"
|
||||
|
||||
if [[ "\$1" == "api" && "\$2" == "user" ]]; then
|
||||
cat <<'JSON'
|
||||
${jsonEncode({'login': 'tea-octocat'})}
|
||||
JSON
|
||||
exit 0
|
||||
fi
|
||||
|
||||
if [[ "\$1" == "api" && "\$4" == repos/owner/sample/issues\\?state=open* ]]; then
|
||||
cat <<'JSON'
|
||||
[]
|
||||
JSON
|
||||
exit 0
|
||||
fi
|
||||
|
||||
if [[ "\$1" == "api" && "\$4" == "repos/owner/sample/issues/12/comments?limit=100" ]]; then
|
||||
cat <<'JSON'
|
||||
${jsonEncode(comments)}
|
||||
JSON
|
||||
exit 0
|
||||
fi
|
||||
|
||||
if [[ "\$1" == "api" && "\$4" == "repos/owner/sample/issues/12/comments" ]]; then
|
||||
for arg in "\$@"; do
|
||||
if [[ "\$arg" == body=* ]]; then
|
||||
printf '%s' "\${arg#body=}" > "${postedBody.path}"
|
||||
fi
|
||||
done
|
||||
cat <<'JSON'
|
||||
${jsonEncode({'id': 801, 'url': 'https://gitea.example.test/comment/801', 'body': 'posted'})}
|
||||
JSON
|
||||
exit 0
|
||||
fi
|
||||
|
||||
if [[ "\$1" == "webhooks" && "\$2" == "create" ]]; then
|
||||
cat <<'JSON'
|
||||
${jsonEncode({'id': 81})}
|
||||
JSON
|
||||
exit 0
|
||||
fi
|
||||
|
||||
if [[ "\$1" == "webhooks" && "\$2" == "delete" ]]; then
|
||||
exit 0
|
||||
fi
|
||||
|
||||
echo "unexpected tea args: \$*" >&2
|
||||
exit 1
|
||||
''');
|
||||
await Process.run('chmod', ['+x', teaScript.path]);
|
||||
|
||||
// And a responder that emits a planning reply.
|
||||
final responderScript = File(p.join(sandbox.path, 'responder.sh'));
|
||||
await writeResponderScript(responderScript, {
|
||||
'decision': 'reply',
|
||||
'mode': 'planning',
|
||||
'markdown': 'Plan:\n- consume gosmee issue_comment events',
|
||||
'summary': 'posted channel-based planning advice',
|
||||
});
|
||||
|
||||
// And an orchestrator-style config that uses tea/gosmee.
|
||||
final configFile = File(p.join(sandbox.path, 'agent-orchestrator.yaml'));
|
||||
await configFile.writeAsString('''
|
||||
defaults:
|
||||
issueAssistant:
|
||||
enabled: true
|
||||
pollInterval: 5m
|
||||
mentionTriggers: ["@helper"]
|
||||
responders:
|
||||
- id: primary
|
||||
command: ${responderScript.path}
|
||||
timeout: 1m
|
||||
projects:
|
||||
sample:
|
||||
provider: gitea
|
||||
repo: owner/sample
|
||||
path: ${repoDir.path}
|
||||
issueAssistant:
|
||||
eventSource: tea/gosmee
|
||||
''');
|
||||
final app = await IssueAssistantApp.open(
|
||||
config: await AppConfig.load(configFile.path),
|
||||
databasePath: p.join(sandbox.path, 'state.sqlite3'),
|
||||
gosmeeCommand: gosmeeScript.path,
|
||||
teaCommand: teaScript.path,
|
||||
);
|
||||
|
||||
// When the long-running app starts and receives the forwarded webhook.
|
||||
final runFuture = app.run();
|
||||
for (var attempt = 0; attempt < 50; attempt += 1) {
|
||||
if (await postedBody.exists()) {
|
||||
break;
|
||||
}
|
||||
await Future<void>.delayed(const Duration(milliseconds: 100));
|
||||
}
|
||||
await app.close();
|
||||
await runFuture;
|
||||
|
||||
// Then it reads issue comments and posts exactly one reply for that issue.
|
||||
expect(await postedBody.exists(), isTrue);
|
||||
final postedComment = await postedBody.readAsString();
|
||||
expect(
|
||||
postedComment,
|
||||
contains('Plan:\n- consume gosmee issue_comment events'),
|
||||
);
|
||||
expect(postedComment, contains('@tea-octocat'));
|
||||
|
||||
// And the tea webhook lifecycle commands are invoked for the Gitea project.
|
||||
final teaLogLines = await teaLog.readAsLines();
|
||||
expect(
|
||||
teaLogLines.any(
|
||||
(line) => line.contains(
|
||||
'webhooks create --type gitea --events issues,issue_comment --active -o json -r owner/sample https://gosmee.example.test/sample',
|
||||
),
|
||||
),
|
||||
isTrue,
|
||||
);
|
||||
expect(
|
||||
teaLogLines.any(
|
||||
(line) =>
|
||||
line.contains('webhooks delete --confirm -r owner/sample 81'),
|
||||
),
|
||||
isTrue,
|
||||
);
|
||||
});
|
||||
|
||||
/// ```gherkin
|
||||
/// Scenario: Reconcile a missed Gitea gosmee event through polling
|
||||
/// Given a temporary project checkout that can be used as the local repo path
|
||||
/// And a gosmee script that starts forwarding without delivering an event
|
||||
/// And a tea script that returns a new issue only on a later reconciliation poll
|
||||
/// And a responder that emits a planning reply
|
||||
/// And an orchestrator-style config that uses tea/gosmee with a short poll interval
|
||||
/// When the long-running app starts and enough time passes for reconciliation polling
|
||||
/// Then it posts exactly one reply for the issue discovered by polling
|
||||
/// And the app performs more than one Gitea issue listing request while running
|
||||
/// ```
|
||||
test('Reconcile a missed Gitea gosmee event through polling', () async {
|
||||
// Given a temporary project checkout that can be used as the local repo path.
|
||||
final sandbox = await Directory.systemTemp.createTemp(
|
||||
'cws-gosmee-reconcile-',
|
||||
);
|
||||
final repoDir = await Directory(p.join(sandbox.path, 'repo')).create();
|
||||
await initGitRepo(repoDir);
|
||||
|
||||
// And a gosmee script that starts forwarding without delivering an event.
|
||||
final gosmeeScript = File(p.join(sandbox.path, 'gosmee'));
|
||||
final gosmeeLog = File(p.join(sandbox.path, 'gosmee-log.jsonl'));
|
||||
await gosmeeScript.writeAsString('''#!/usr/bin/env bash
|
||||
set -euo pipefail
|
||||
printf '%s\n' "\$*" >> "${gosmeeLog.path}"
|
||||
|
||||
if [[ "\$1" == "--output" && "\$2" == "json" && "\$3" == "client" && "\$4" == "--new-url" ]]; then
|
||||
printf '%s\n' 'https://gosmee.example.test/reconcile'
|
||||
exit 0
|
||||
fi
|
||||
|
||||
if [[ "\$1" == "client" ]]; then
|
||||
while true; do
|
||||
sleep 60
|
||||
done
|
||||
fi
|
||||
|
||||
echo "unexpected gosmee args: \$*" >&2
|
||||
exit 1
|
||||
''');
|
||||
await Process.run('chmod', ['+x', gosmeeScript.path]);
|
||||
|
||||
// And a tea script that returns a new issue only on a later reconciliation poll.
|
||||
final teaScript = File(p.join(sandbox.path, 'tea'));
|
||||
final teaLog = File(p.join(sandbox.path, 'tea-log.jsonl'));
|
||||
final postedBody = File(p.join(sandbox.path, 'posted-body.md'));
|
||||
final issueListCount = File(p.join(sandbox.path, 'issue-list-count'));
|
||||
final issue = {
|
||||
'index': 18,
|
||||
'title': 'Need reconciliation after a missed webhook',
|
||||
'body': 'Please recover this mention from polling.',
|
||||
'state': 'open',
|
||||
'url': 'https://gitea.example.test/owner/sample/issues/18',
|
||||
'updated_at': '2026-04-08T05:00:00Z',
|
||||
'poster': {'login': 'reporter'},
|
||||
'labels': const <Map<String, Object?>>[],
|
||||
};
|
||||
final comments = [
|
||||
{
|
||||
'id': 181,
|
||||
'body': '@helper please recover this even if gosmee missed it',
|
||||
'url':
|
||||
'https://gitea.example.test/owner/sample/issues/18#issuecomment-181',
|
||||
'created_at': '2026-04-08T05:00:00Z',
|
||||
'updated_at': '2026-04-08T05:00:00Z',
|
||||
'poster': {'login': 'reporter'},
|
||||
},
|
||||
];
|
||||
await teaScript.writeAsString('''#!/usr/bin/env bash
|
||||
set -euo pipefail
|
||||
printf '%s\n' "\$*" >> "${teaLog.path}"
|
||||
|
||||
if [[ "\$1" == "api" && "\$2" == "user" ]]; then
|
||||
cat <<'JSON'
|
||||
${jsonEncode({'login': 'tea-octocat'})}
|
||||
JSON
|
||||
exit 0
|
||||
fi
|
||||
|
||||
if [[ "\$1" == "api" && "\$4" == repos/owner/sample/issues\\?state=open* ]]; then
|
||||
count=0
|
||||
if [[ -f "${issueListCount.path}" ]]; then
|
||||
count="\$(cat "${issueListCount.path}")"
|
||||
fi
|
||||
count=\$((count + 1))
|
||||
printf '%s' "\$count" > "${issueListCount.path}"
|
||||
if [[ "\$count" -eq 1 ]]; then
|
||||
cat <<'JSON'
|
||||
[]
|
||||
JSON
|
||||
else
|
||||
cat <<'JSON'
|
||||
${jsonEncode([issue])}
|
||||
JSON
|
||||
fi
|
||||
exit 0
|
||||
fi
|
||||
|
||||
if [[ "\$1" == "api" && "\$4" == "repos/owner/sample/issues/18/comments?limit=100" ]]; then
|
||||
cat <<'JSON'
|
||||
${jsonEncode(comments)}
|
||||
JSON
|
||||
exit 0
|
||||
fi
|
||||
|
||||
if [[ "\$1" == "api" && "\$4" == "repos/owner/sample/issues/18/comments" ]]; then
|
||||
for arg in "\$@"; do
|
||||
if [[ "\$arg" == body=* ]]; then
|
||||
printf '%s' "\${arg#body=}" > "${postedBody.path}"
|
||||
fi
|
||||
done
|
||||
cat <<'JSON'
|
||||
${jsonEncode({'id': 802, 'url': 'https://gitea.example.test/comment/802', 'body': 'posted'})}
|
||||
JSON
|
||||
exit 0
|
||||
fi
|
||||
|
||||
if [[ "\$1" == "webhooks" && "\$2" == "create" ]]; then
|
||||
cat <<'JSON'
|
||||
${jsonEncode({'id': 82})}
|
||||
JSON
|
||||
exit 0
|
||||
fi
|
||||
|
||||
if [[ "\$1" == "webhooks" && "\$2" == "delete" ]]; then
|
||||
exit 0
|
||||
fi
|
||||
|
||||
echo "unexpected tea args: \$*" >&2
|
||||
exit 1
|
||||
''');
|
||||
await Process.run('chmod', ['+x', teaScript.path]);
|
||||
|
||||
// And a responder that emits a planning reply.
|
||||
final responderScript = File(p.join(sandbox.path, 'responder.sh'));
|
||||
await writeResponderScript(responderScript, {
|
||||
'decision': 'reply',
|
||||
'mode': 'planning',
|
||||
'markdown': 'Plan:\n- reconcile missed gosmee events through polling',
|
||||
'summary': 'posted reconciliation advice',
|
||||
});
|
||||
|
||||
// And an orchestrator-style config that uses tea/gosmee with a short poll interval.
|
||||
final configFile = File(p.join(sandbox.path, 'agent-orchestrator.yaml'));
|
||||
await configFile.writeAsString('''
|
||||
defaults:
|
||||
issueAssistant:
|
||||
enabled: true
|
||||
pollInterval: 1s
|
||||
mentionTriggers: ["@helper"]
|
||||
responders:
|
||||
- id: primary
|
||||
command: ${responderScript.path}
|
||||
timeout: 1m
|
||||
projects:
|
||||
sample:
|
||||
provider: gitea
|
||||
repo: owner/sample
|
||||
path: ${repoDir.path}
|
||||
issueAssistant:
|
||||
eventSource: tea/gosmee
|
||||
''');
|
||||
final app = await IssueAssistantApp.open(
|
||||
config: await AppConfig.load(configFile.path),
|
||||
databasePath: p.join(sandbox.path, 'state.sqlite3'),
|
||||
gosmeeCommand: gosmeeScript.path,
|
||||
teaCommand: teaScript.path,
|
||||
);
|
||||
|
||||
// When the long-running app starts and enough time passes for reconciliation polling.
|
||||
final runFuture = app.run();
|
||||
for (var attempt = 0; attempt < 50; attempt += 1) {
|
||||
if (await postedBody.exists()) {
|
||||
break;
|
||||
}
|
||||
await Future<void>.delayed(const Duration(milliseconds: 100));
|
||||
}
|
||||
await app.close();
|
||||
await runFuture;
|
||||
|
||||
// Then it posts exactly one reply for the issue discovered by polling.
|
||||
expect(await postedBody.exists(), isTrue);
|
||||
final postedComment = await postedBody.readAsString();
|
||||
expect(
|
||||
postedComment,
|
||||
contains('Plan:\n- reconcile missed gosmee events through polling'),
|
||||
);
|
||||
|
||||
// And the app performs more than one Gitea issue listing request while running.
|
||||
final teaLogLines = await teaLog.readAsLines();
|
||||
expect(
|
||||
teaLogLines
|
||||
.where(
|
||||
(line) => line.contains('repos/owner/sample/issues?state=open'),
|
||||
)
|
||||
.length,
|
||||
greaterThanOrEqualTo(2),
|
||||
);
|
||||
final gosmeeLogLines = await gosmeeLog.readAsLines();
|
||||
expect(
|
||||
gosmeeLogLines.any(
|
||||
(line) =>
|
||||
line.contains('client https://gosmee.example.test/reconcile'),
|
||||
),
|
||||
isTrue,
|
||||
);
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
void main() {
|
||||
driftRuntimeOptions.dontWarnAboutMultipleDatabases = true;
|
||||
registerIssueAssistantAppRunGosmeeTests();
|
||||
}
|
||||
|
|
@ -93,7 +93,25 @@ projects:
|
|||
|
||||
// When the long-running app starts and enough time passes for a retry.
|
||||
final runFuture = app.run();
|
||||
await Future<void>.delayed(const Duration(milliseconds: 2600));
|
||||
for (var attempt = 0; attempt < 60; attempt += 1) {
|
||||
final logLines = await ghLog.exists()
|
||||
? await ghLog.readAsLines()
|
||||
: const <String>[];
|
||||
final retrySearchCount = logLines
|
||||
.where(
|
||||
(line) =>
|
||||
line ==
|
||||
'api -X GET search/issues -f q=repo:owner/sample is:issue state:open -f sort=updated -f order=asc -f per_page=100 -f page=1',
|
||||
)
|
||||
.length;
|
||||
final sawRetryLog = logMessages.any(
|
||||
(message) => message.contains('will_retry=true'),
|
||||
);
|
||||
if (retrySearchCount >= 2 && sawRetryLog) {
|
||||
break;
|
||||
}
|
||||
await Future<void>.delayed(const Duration(milliseconds: 100));
|
||||
}
|
||||
await app.close();
|
||||
await runFuture;
|
||||
await subscription.cancel();
|
||||
|
|
|
|||
|
|
@ -570,6 +570,65 @@ Future<void> writeWebhookForwardGhScript({
|
|||
await Process.run('chmod', ['+x', ghScript.path]);
|
||||
}
|
||||
|
||||
Future<void> writeGosmeeForwardScript({
|
||||
required File gosmeeScript,
|
||||
required String publicUrl,
|
||||
required Map<String, Object?> payload,
|
||||
File? gosmeeLog,
|
||||
}) async {
|
||||
final buffer = StringBuffer()
|
||||
..writeln('#!/usr/bin/env bash')
|
||||
..writeln('set -euo pipefail');
|
||||
|
||||
if (gosmeeLog != null) {
|
||||
buffer.writeln(
|
||||
r'''printf '%s\n' "$*" >> '''
|
||||
'"${gosmeeLog.path}"',
|
||||
);
|
||||
}
|
||||
|
||||
buffer
|
||||
..writeln(
|
||||
r'''if [[ "$1" == "--output" && "$2" == "json" && "$3" == "client" && "$4" == "--new-url" ]]; then''',
|
||||
)
|
||||
..writeln(
|
||||
r''' printf '%s\n' '''
|
||||
"'$publicUrl'",
|
||||
)
|
||||
..writeln(' exit 0')
|
||||
..writeln('fi');
|
||||
|
||||
buffer
|
||||
..writeln(r'''if [[ "$1" == "client" ]]; then''')
|
||||
..writeln(r''' url="$2"''')
|
||||
..writeln(r''' target="$3"''')
|
||||
..writeln(
|
||||
r''' if [[ "$url" != '''
|
||||
"'$publicUrl'"
|
||||
r''' ]]; then''',
|
||||
)
|
||||
..writeln(r''' echo "unexpected gosmee url: $url" >&2''')
|
||||
..writeln(' exit 1')
|
||||
..writeln(r''' fi''')
|
||||
..writeln(r''' curl -sS -X POST \''')
|
||||
..writeln(r''' -H "Content-Type: application/json" \''')
|
||||
..writeln(r''' -H "X-Gitea-Event: issue_comment" \''')
|
||||
..writeln(r''' --data-binary @- "$target" <<'JSON' >/dev/null''')
|
||||
..writeln(jsonEncode(payload))
|
||||
..writeln('JSON')
|
||||
..writeln(r''' while true; do''')
|
||||
..writeln(r''' sleep 60''')
|
||||
..writeln(r''' done''')
|
||||
..writeln('fi');
|
||||
|
||||
buffer
|
||||
..writeln(r'''echo "unexpected gosmee args: $*" >&2''')
|
||||
..writeln('exit 1');
|
||||
|
||||
await gosmeeScript.writeAsString(buffer.toString());
|
||||
await Process.run('chmod', ['+x', gosmeeScript.path]);
|
||||
}
|
||||
|
||||
Future<void> initGitRepo(Directory directory) async {
|
||||
await GitDir.init(directory.path, allowContent: true);
|
||||
await runGit([
|
||||
|
|
|
|||
Loading…
Reference in New Issue