diff --git a/bin/code_work_spawner.dart b/bin/code_work_spawner.dart index 5598baa..f8ccf84 100644 --- a/bin/code_work_spawner.dart +++ b/bin/code_work_spawner.dart @@ -50,6 +50,11 @@ class _CodeWorkSpawnerCommandRunner extends CommandRunner { '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 { 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, ); diff --git a/examples/README.md b/examples/README.md index 2a94853..b073b8e 100644 --- a/examples/README.md +++ b/examples/README.md @@ -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 diff --git a/examples/simple-gitea.yaml b/examples/simple-gitea.yaml index 0b0cbfc..47fce2f 100644 --- a/examples/simple-gitea.yaml +++ b/examples/simple-gitea.yaml @@ -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"] diff --git a/lib/src/config.dart b/lib/src/config.dart index 04acdfb..7e3347f 100644 --- a/lib/src/config.dart +++ b/lib/src/config.dart @@ -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( diff --git a/lib/src/issue_assistant_app.dart b/lib/src/issue_assistant_app.dart index c2af939..ae031de 100644 --- a/lib/src/issue_assistant_app.dart +++ b/lib/src/issue_assistant_app.dart @@ -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(), diff --git a/lib/src/issue_event_source.dart b/lib/src/issue_event_source.dart index 7b3f177..90eda17 100644 --- a/lib/src/issue_event_source.dart +++ b/lib/src/issue_event_source.dart @@ -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; } } diff --git a/lib/src/issue_event_source_gitea_gosmee.dart b/lib/src/issue_event_source_gitea_gosmee.dart new file mode 100644 index 0000000..7408f6f --- /dev/null +++ b/lib/src/issue_event_source_gitea_gosmee.dart @@ -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 _queuedIssues = {}; + final Set> _activeDispatches = >{}; + final Map _gosmeeProcesses = {}; + final Map _webhookIdsByProject = {}; + final Map _repoSlugsByProject = {}; + + HttpServer? _server; + Timer? _flushTimer; + Completer? _runCompleter; + bool _isClosing = false; + bool _isClosed = false; + + @override + Future run({ + required List 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(); + _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 runOnce({ + required List projects, + required IssueEventBatchHandler onBatch, + }) async { + throw UnsupportedError( + 'gitea/gosmee does not support runOnce; use polling instead.', + ); + } + + @override + Future 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.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 _handleWebhookRequest({ + required HttpRequest request, + required Map 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) { + throw const FormatException('Webhook payload must be a JSON object.'); + } + final repository = payload['repository']; + final fullName = repository is Map + ? repository['full_name']?.toString() + : null; + final project = fullName == null ? null : repoToProject[fullName]; + final issueNode = payload['issue']; + if (project != null && + issueNode is Map && + (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 _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) { + 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 _flushQueuedIssues() async { + if (_queuedIssues.isEmpty) { + return; + } + + final queuedIssues = _queuedIssues.values.toList(growable: false); + _queuedIssues.clear(); + final groupedIssues = {}; + 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 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> 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 issues; + + _QueuedGiteaBatch add(GitHubIssueSummary issue) { + issues.add(issue); + return this; + } +} diff --git a/lib/src/issue_tracker_client.dart b/lib/src/issue_tracker_client.dart index 99bda40..06b6ddd 100644 --- a/lib/src/issue_tracker_client.dart +++ b/lib/src/issue_tracker_client.dart @@ -121,6 +121,46 @@ class IssueTrackerClient { }; } + Future 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) { + throw const FormatException( + 'tea webhooks create returned an unexpected response.', + ); + } + return _readInt(json, 'id'); + } + + Future deleteGiteaWebhook({ + required RepositorySlug repo, + required int webhookId, + }) async { + await _runTeaJson([ + 'webhooks', + 'delete', + '--confirm', + '-r', + repo.fullName, + '$webhookId', + ]); + } + Future> _fetchUpdatedGitHubIssues({ required Iterable repos, required DateTime? since, diff --git a/test/app_config_test.dart b/test/app_config_test.dart index 7d05134..eea6cba 100644 --- a/test/app_config_test.dart +++ b/test/app_config_test.dart @@ -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().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 diff --git a/test/init_config_cli_test.dart b/test/init_config_cli_test.dart index 39fc156..4c22718 100644 --- a/test/init_config_cli_test.dart +++ b/test/init_config_cli_test.dart @@ -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 diff --git a/test/issue_assistant_app_run_gosmee_test.dart b/test/issue_assistant_app_run_gosmee_test.dart new file mode 100644 index 0000000..c88f249 --- /dev/null +++ b/test/issue_assistant_app_run_gosmee_test.dart @@ -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 >[], + }; + 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.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 >[], + }; + 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.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(); +} diff --git a/test/issue_assistant_app_run_retry_test.dart b/test/issue_assistant_app_run_retry_test.dart index 149269e..c0a40be 100644 --- a/test/issue_assistant_app_run_retry_test.dart +++ b/test/issue_assistant_app_run_retry_test.dart @@ -93,7 +93,25 @@ projects: // When the long-running app starts and enough time passes for a retry. final runFuture = app.run(); - await Future.delayed(const Duration(milliseconds: 2600)); + for (var attempt = 0; attempt < 60; attempt += 1) { + final logLines = await ghLog.exists() + ? await ghLog.readAsLines() + : const []; + 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.delayed(const Duration(milliseconds: 100)); + } await app.close(); await runFuture; await subscription.cancel(); diff --git a/test/test_support.dart b/test/test_support.dart index dd85f9a..fbbba2d 100644 --- a/test/test_support.dart +++ b/test/test_support.dart @@ -570,6 +570,65 @@ Future writeWebhookForwardGhScript({ await Process.run('chmod', ['+x', ghScript.path]); } +Future writeGosmeeForwardScript({ + required File gosmeeScript, + required String publicUrl, + required Map 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 initGitRepo(Directory directory) async { await GitDir.init(directory.path, allowContent: true); await runGit([