diff --git a/README.md b/README.md index fe647e6..ba50301 100644 --- a/README.md +++ b/README.md @@ -4,9 +4,7 @@ Repository issue thread assistant with an Agent Orchestrator style YAML config. It consumes issue tracker events for configured GitHub or Gitea repositories, reads issue threads, and uses CLI responders such as `codex`, `opencode`, or -`copilot-cli` to decide when to reply. Polling via `gh` and `tea` is the -currently bundled event source, with the runtime structured so future -channel-based sources can plug into the same processing pipeline. It supports +`copilot-cli` to decide when to reply. It supports multi-repo config, responder fallback chains, and optional Discord webhook and desktop notifications. diff --git a/bin/code_work_spawner.dart b/bin/code_work_spawner.dart index 0f4238a..5598baa 100644 --- a/bin/code_work_spawner.dart +++ b/bin/code_work_spawner.dart @@ -27,7 +27,7 @@ class _CodeWorkSpawnerCommandRunner extends CommandRunner { ..addFlag( 'once', abbr: '1', - help: 'Run a single poll cycle and exit.', + help: 'Run a single reconciliation cycle and exit.', negatable: false, ) ..addOption( diff --git a/examples/README.md b/examples/README.md index 9133e34..2a94853 100644 --- a/examples/README.md +++ b/examples/README.md @@ -3,7 +3,16 @@ The repo uses an Agent Orchestrator style YAML config with `defaults` and `projects`. Project entries can override `issueAssistant` fields when a repo needs a -different prompt, mention triggers, responder chain, or notification list. +different prompt, mention triggers, responder chain, notification list, or +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` 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. 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-github.yaml b/examples/simple-github.yaml index 8515dc5..e10164b 100644 --- a/examples/simple-github.yaml +++ b/examples/simple-github.yaml @@ -7,6 +7,8 @@ worktreeDir: ~/.worktrees defaults: issueAssistant: enabled: true + # Use gh/webhook-forward for long-running runs while keeping --once on polling. + # eventSource: gh/webhook-forward pollInterval: 30s maxConcurrentIssues: 3 mentionTriggers: ["@codex", "@helper"] diff --git a/lib/src/config.dart b/lib/src/config.dart index 6dbf3f2..04acdfb 100644 --- a/lib/src/config.dart +++ b/lib/src/config.dart @@ -10,6 +10,7 @@ import 'config_schema.dart'; export 'config_schema.dart' show AssistantCapability, + IssueAssistantEventSourceKind, IssueTrackerProvider, NotificationConfigDocument, NotificationEvent, @@ -132,6 +133,7 @@ class AppConfig { defaults: AppConfigDefaultsDocument( issueAssistant: IssueAssistantConfigDocument( enabled: true, + eventSource: IssueAssistantEventSourceKind.ghPolling, pollInterval: '30s', maxConcurrentIssues: 3, mentionTriggers: const ['@codex', '@helper'], @@ -188,15 +190,49 @@ class ProjectConfig { ); } + final provider = document.provider ?? IssueTrackerProvider.github; + var issueAssistant = defaultAssistant.merge(document.issueAssistant); + if (document.issueAssistant?.eventSource == null && + _isPollingEventSource(issueAssistant.eventSource)) { + issueAssistant = issueAssistant.copyWith( + eventSource: provider == IssueTrackerProvider.github + ? IssueAssistantEventSourceKind.ghPolling + : IssueAssistantEventSourceKind.teaPolling, + ); + } + switch (issueAssistant.eventSource) { + case IssueAssistantEventSourceKind.ghPolling: + case IssueAssistantEventSourceKind.ghWebhookForward: + if (provider != IssueTrackerProvider.github) { + throw FormatException( + 'projects.$key.issueAssistant.eventSource ' + '${_eventSourceConfigValue(issueAssistant.eventSource)} ' + 'requires provider: github.', + ); + } + case IssueAssistantEventSourceKind.teaPolling: + if (provider != IssueTrackerProvider.gitea) { + throw FormatException( + 'projects.$key.issueAssistant.eventSource tea/polling ' + 'requires provider: gitea.', + ); + } + case IssueAssistantEventSourceKind.teaGosmee: + throw FormatException( + 'projects.$key.issueAssistant.eventSource tea/gosmee ' + 'is not implemented yet.', + ); + } + return ProjectConfig( key: key, - provider: document.provider ?? IssueTrackerProvider.github, + provider: provider, repo: repo, repoSlug: _parseRepositorySlug(repo, field: 'projects.$key.repo'), path: _expandHome(path), defaultBranch: document.defaultBranch ?? 'main', sessionPrefix: document.sessionPrefix ?? _deriveSessionPrefix(key), - issueAssistant: defaultAssistant.merge(document.issueAssistant), + issueAssistant: issueAssistant, ); } @@ -217,6 +253,7 @@ class ProjectConfig { class IssueAssistantConfig { IssueAssistantConfig({ required this.enabled, + required this.eventSource, required this.pollInterval, required this.maxConcurrentIssues, required this.mentionTriggers, @@ -230,6 +267,7 @@ class IssueAssistantConfig { }); final bool enabled; + final IssueAssistantEventSourceKind eventSource; final Duration pollInterval; final int maxConcurrentIssues; final List mentionTriggers; @@ -255,6 +293,23 @@ class IssueAssistantConfig { overlay: overlay, ); } + + IssueAssistantConfig copyWith({IssueAssistantEventSourceKind? eventSource}) { + return IssueAssistantConfig( + enabled: enabled, + eventSource: eventSource ?? this.eventSource, + pollInterval: pollInterval, + maxConcurrentIssues: maxConcurrentIssues, + mentionTriggers: mentionTriggers, + prompt: prompt, + responders: responders, + capabilities: capabilities, + notifications: notifications, + commentTag: commentTag, + commentPrefixTemplate: commentPrefixTemplate, + commentFooterTemplate: commentFooterTemplate, + ); + } } class CliResponderConfig { @@ -349,6 +404,10 @@ class _IssueAssistantConfigResolver { final notifications = overlay?.notifications; return IssueAssistantConfig( enabled: overlay?.enabled ?? base?.enabled ?? true, + eventSource: + overlay?.eventSource ?? + base?.eventSource ?? + _defaultEventSourceForProvider(base), pollInterval: overlay?.pollInterval != null ? _parseDuration(overlay!.pollInterval!) : base?.pollInterval ?? const Duration(seconds: 30), @@ -389,6 +448,30 @@ class _IssueAssistantConfigResolver { defaultCommentFooterTemplate, ); } + + IssueAssistantEventSourceKind _defaultEventSourceForProvider( + IssueAssistantConfig? base, + ) { + return base?.eventSource ?? IssueAssistantEventSourceKind.ghPolling; + } +} + +String _eventSourceConfigValue(IssueAssistantEventSourceKind value) { + return switch (value) { + IssueAssistantEventSourceKind.ghPolling => 'gh/polling', + IssueAssistantEventSourceKind.teaPolling => 'tea/polling', + IssueAssistantEventSourceKind.ghWebhookForward => 'gh/webhook-forward', + IssueAssistantEventSourceKind.teaGosmee => 'tea/gosmee', + }; +} + +bool _isPollingEventSource(IssueAssistantEventSourceKind value) { + return switch (value) { + IssueAssistantEventSourceKind.ghPolling => true, + IssueAssistantEventSourceKind.teaPolling => true, + IssueAssistantEventSourceKind.ghWebhookForward => false, + IssueAssistantEventSourceKind.teaGosmee => false, + }; } String renderTemplate(String template, Map values) { @@ -610,6 +693,7 @@ ${_YamlEmitter().write({'defaults': defaultsSection})} # capabilities: ["planning", "bug_verification"] # commentTag: code-work-spawner + # eventSource: gh/polling # commentPrefixTemplate: | ${_commentBlock(defaultCommentPrefixTemplate, indent: ' # ')} # commentFooterTemplate: | @@ -638,6 +722,7 @@ ${_YamlEmitter().write({'projects': projectsSection})} # issueAssistant: # enabled: ${assistant?.enabled ?? true} + # eventSource: gh/webhook-forward # mentionTriggers: ["@helper"] # sessionPrefix: sample ''' diff --git a/lib/src/config_schema.dart b/lib/src/config_schema.dart index e695224..709eeb4 100644 --- a/lib/src/config_schema.dart +++ b/lib/src/config_schema.dart @@ -86,6 +86,7 @@ class ProjectConfigDocument { class IssueAssistantConfigDocument { const IssueAssistantConfigDocument({ this.enabled, + this.eventSource, this.pollInterval, this.maxConcurrentIssues, this.mentionTriggers, @@ -99,6 +100,7 @@ class IssueAssistantConfigDocument { }); final bool? enabled; + final IssueAssistantEventSourceKind? eventSource; final String? pollInterval; final int? maxConcurrentIssues; final List? mentionTriggers; @@ -179,6 +181,17 @@ class NotificationConfigDocument { @JsonEnum(fieldRename: FieldRename.snake) enum AssistantCapability { planning, bugVerification } +enum IssueAssistantEventSourceKind { + @JsonValue('gh/polling') + ghPolling, + @JsonValue('tea/polling') + teaPolling, + @JsonValue('gh/webhook-forward') + ghWebhookForward, + @JsonValue('tea/gosmee') + teaGosmee, +} + @JsonEnum(fieldRename: FieldRename.snake) enum IssueTrackerProvider { github, gitea } diff --git a/lib/src/issue_assistant_app.dart b/lib/src/issue_assistant_app.dart index 142a56f..c2af939 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_github_webhook_forward.dart'; import 'issue_tracker_client.dart'; import 'notifier.dart'; import 'workspace_manager.dart'; @@ -31,6 +32,7 @@ class IssueAssistantApp { final IssueEventSource issueEventSource; final CliResponderRunner responderRunner; final WorkspaceManager workspaceManager; + final Map> _issueProcessingFutures = {}; final NotifierRegistry Function(ProjectConfig project) notifierRegistryFactory; @@ -45,14 +47,25 @@ class IssueAssistantApp { ghCommand: ghCommand, teaCommand: teaCommand, ); + final pollingIssueEventSource = PollingIssueEventSource( + database: database, + issueTrackerClient: issueTrackerClient, + logger: _logger, + ); return IssueAssistantApp._( config: config, database: database, issueTrackerClient: issueTrackerClient, - issueEventSource: PollingIssueEventSource( - database: database, - issueTrackerClient: issueTrackerClient, - logger: _logger, + issueEventSource: RoutedIssueEventSource( + sources: { + IssueAssistantEventSourceKind.ghPolling: pollingIssueEventSource, + IssueAssistantEventSourceKind.teaPolling: pollingIssueEventSource, + IssueAssistantEventSourceKind.ghWebhookForward: + GitHubWebhookForwardIssueEventSource( + issueTrackerClient: issueTrackerClient, + logger: _logger, + ), + }, ), responderRunner: CliResponderRunner(), workspaceManager: WorkspaceManager(worktreeRoot: config.worktreeDir), @@ -133,7 +146,7 @@ class IssueAssistantApp { 'project=${project.key} process_issue issue=${issues[currentIndex].number} ' 'queue_index=${currentIndex + 1}/${issues.length} active_limit=$activeCount', ); - await _processIssue(project, issues[currentIndex]); + await _processIssueSerialized(project, issues[currentIndex]); } } @@ -428,6 +441,31 @@ class IssueAssistantApp { ); } + Future _processIssueSerialized( + ProjectConfig project, + GitHubIssueSummary issue, + ) async { + final issueKey = '${project.key}#${issue.number}'; + final previous = _issueProcessingFutures[issueKey]; + + late final Future current; + current = () async { + if (previous != null) { + await previous.catchError((_) {}); + } + await _processIssue(project, issue); + }(); + + _issueProcessingFutures[issueKey] = current; + try { + await current; + } finally { + if (identical(_issueProcessingFutures[issueKey], current)) { + _issueProcessingFutures.remove(issueKey); + } + } + } + String _buildPrompt({ required ProjectConfig project, required IssueThread thread, diff --git a/lib/src/issue_event_source.dart b/lib/src/issue_event_source.dart index c69ccba..7b3f177 100644 --- a/lib/src/issue_event_source.dart +++ b/lib/src/issue_event_source.dart @@ -38,6 +38,83 @@ abstract class IssueEventSource { Future close(); } +class RoutedIssueEventSource implements IssueEventSource { + RoutedIssueEventSource({required this.sources}); + + final Map sources; + + @override + Future run({ + required List projects, + required IssueEventBatchHandler onBatch, + }) async { + final projectsBySource = + >{ + for (final eventSource in sources.keys) + eventSource: [], + }; + for (final project in projects) { + for (final projectSource in _registeredSourcesFor(project)) { + final sourceProjects = projectsBySource[projectSource]; + if (sourceProjects == null) { + throw StateError( + 'No issue event source registered for ${projectSource.name}.', + ); + } + sourceProjects.add(project); + } + } + + await Future.wait( + projectsBySource.entries + .where((entry) => entry.value.isNotEmpty) + .map( + (entry) => sources[entry.key]!.run( + projects: entry.value, + onBatch: onBatch, + ), + ), + ); + } + + @override + Future runOnce({ + required List projects, + required IssueEventBatchHandler onBatch, + }) { + return sources[IssueAssistantEventSourceKind.ghPolling]!.runOnce( + projects: projects, + onBatch: onBatch, + ); + } + + @override + Future close() async { + for (final source in sources.values) { + await source.close(); + } + } + + Iterable _registeredSourcesFor( + ProjectConfig project, + ) sync* { + switch (project.issueAssistant.eventSource) { + case IssueAssistantEventSourceKind.ghPolling: + yield IssueAssistantEventSourceKind.ghPolling; + case IssueAssistantEventSourceKind.teaPolling: + yield IssueAssistantEventSourceKind.teaPolling; + case IssueAssistantEventSourceKind.ghWebhookForward: + // Keep reconciliation polling active for startup sync and missed + // webhook recovery while the forwarded channel handles low-latency + // issue/comment events. + yield IssueAssistantEventSourceKind.ghPolling; + yield IssueAssistantEventSourceKind.ghWebhookForward; + case IssueAssistantEventSourceKind.teaGosmee: + yield IssueAssistantEventSourceKind.teaGosmee; + } + } +} + class PollingIssueEventSource implements IssueEventSource { PollingIssueEventSource({ required this.database, diff --git a/lib/src/issue_event_source_github_webhook_forward.dart b/lib/src/issue_event_source_github_webhook_forward.dart new file mode 100644 index 0000000..ce48afa --- /dev/null +++ b/lib/src/issue_event_source_github_webhook_forward.dart @@ -0,0 +1,323 @@ +import 'dart:async'; +import 'dart:convert'; +import 'dart:io'; + +import 'package:logging/logging.dart'; + +import 'config.dart'; +import 'issue_event_source.dart'; +import 'issue_tracker_client.dart'; + +class GitHubWebhookForwardIssueEventSource implements IssueEventSource { + GitHubWebhookForwardIssueEventSource({ + required this.issueTrackerClient, + required Logger logger, + }) : _logger = logger; + + static const Duration _webhookCoalescingWindow = Duration(milliseconds: 500); + + final IssueTrackerClient issueTrackerClient; + final Logger _logger; + final Map _queuedIssues = {}; + final Set> _activeDispatches = >{}; + final Map _forwardProcesses = {}; + + 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 githubProjects = projects + .where((project) => project.provider == IssueTrackerProvider.github) + .toList(growable: false); + if (githubProjects.isEmpty) { + return; + } + + final completer = Completer(); + _runCompleter = completer; + final server = await HttpServer.bind(InternetAddress.loopbackIPv4, 0); + _server = server; + final repoToProject = { + for (final project in githubProjects) project.repo: project, + }; + server.listen( + (request) => unawaited( + _handleWebhookRequest( + request: request, + repoToProject: repoToProject, + onBatch: onBatch, + ), + ), + onError: (Object error, StackTrace stackTrace) { + _logError('github/webhook-forward server error=$error'); + if (!_isClosing && !completer.isCompleted) { + completer.completeError(error, stackTrace); + } + }, + ); + + final url = + 'http://${server.address.address}:${server.port}/github/webhooks'; + for (final project in githubProjects) { + final process = await Process.start(issueTrackerClient.ghCommand, [ + 'webhook', + 'forward', + '--events=issues,issue_comment', + '--repo=${project.repo}', + '--url=$url', + ]); + _forwardProcesses[project.key] = process; + _pipeProcessLogs( + project: project, + stream: process.stdout, + level: 'stdout', + ); + _pipeProcessLogs( + project: project, + stream: process.stderr, + level: 'stderr', + ); + unawaited( + process.exitCode.then((exitCode) { + _forwardProcesses.remove(project.key); + _log( + 'github/webhook-forward process project=${project.key} ' + 'exit_code=$exitCode', + ); + if (!_isClosing && !completer.isCompleted) { + completer.completeError( + StateError( + 'gh webhook forward 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( + 'github/webhook-forward 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 = _forwardProcesses.values.toList(growable: false); + _forwardProcesses.clear(); + for (final process in processes) { + process.kill(); + } + for (final process in processes) { + await process.exitCode.catchError((_) => 0); + } + + 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}'] = _QueuedWebhookIssue( + project: project, + issue: issue, + onBatch: onBatch, + ); + _scheduleFlush(); + } + } + + request.response.statusCode = HttpStatus.accepted; + await request.response.close(); + } catch (error) { + _logError('github/webhook-forward payload error=$error'); + request.response + ..statusCode = HttpStatus.badRequest + ..write('Invalid webhook payload.'); + await request.response.close(); + } + } + + 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: () => _QueuedWebhookBatch( + 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( + 'github/webhook-forward 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: 'github/webhook-forward', + ), + ); + } + }(); + + _activeDispatches.add(dispatchFuture); + try { + await dispatchFuture; + } finally { + _activeDispatches.remove(dispatchFuture); + } + } + + void _pipeProcessLogs({ + required ProjectConfig project, + required Stream> stream, + required String level, + }) { + stream.transform(utf8.decoder).transform(const LineSplitter()).listen(( + line, + ) { + _log('github/webhook-forward project=${project.key} $level=$line'); + }); + } + + void _log(String message) { + _logger.info(message); + } + + void _logError(String message) { + _logger.severe(message); + } +} + +class _QueuedWebhookIssue { + _QueuedWebhookIssue({ + required this.project, + required this.issue, + required this.onBatch, + }); + + final ProjectConfig project; + final GitHubIssueSummary issue; + final IssueEventBatchHandler onBatch; +} + +class _QueuedWebhookBatch { + _QueuedWebhookBatch({ + required this.project, + required this.onBatch, + required this.issues, + }); + + final ProjectConfig project; + final IssueEventBatchHandler onBatch; + final List issues; + + _QueuedWebhookBatch add(GitHubIssueSummary issue) { + issues.removeWhere((existing) => existing.number == issue.number); + issues.add(issue); + return this; + } +} diff --git a/test/app_config_test.dart b/test/app_config_test.dart index 80847ff..7d05134 100644 --- a/test/app_config_test.dart +++ b/test/app_config_test.dart @@ -350,6 +350,84 @@ projects: // And the repository slug still parses normally. expect(config.projects['sample']!.repoSlug.fullName, 'owner/sample'); + expect( + config.projects['sample']!.issueAssistant.eventSource, + IssueAssistantEventSourceKind.teaPolling, + ); + }); + + /// ```gherkin + /// Scenario: Load gh/webhook-forward as the issue event source + /// Given a GitHub project config that opts into gh/webhook-forward + /// When loading the app config from disk + /// Then the project keeps gh/webhook-forward as its event source + /// And the GitHub provider remains valid for that project + /// ``` + test('Load gh/webhook-forward as the issue event source', () async { + // Given a GitHub project config that opts into gh/webhook-forward. + final tempDir = await Directory.systemTemp.createTemp( + 'cws-webhook-forward-', + ); + 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: gh/webhook-forward +'''); + + // When loading the app config from disk. + final config = await AppConfig.load(configFile.path); + + // Then the project keeps gh/webhook-forward as its event source. + expect( + config.projects['sample']!.issueAssistant.eventSource, + IssueAssistantEventSourceKind.ghWebhookForward, + ); + + // And the GitHub provider remains valid for that project. + expect(config.projects['sample']!.provider, IssueTrackerProvider.github); + }); + + /// ```gherkin + /// Scenario: Reject gh/webhook-forward for non-GitHub providers + /// Given a Gitea project config that opts into gh/webhook-forward + /// When loading the app config from disk + /// Then loading fails with a clear provider compatibility error + /// ``` + test('Reject gh/webhook-forward for non-GitHub providers', () async { + // Given a Gitea project config that opts into gh/webhook-forward. + final tempDir = await Directory.systemTemp.createTemp( + 'cws-webhook-forward-gitea-', + ); + 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: gh/webhook-forward +'''); + + // 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('gh/webhook-forward requires provider: github'), + ), + ), + ); }); /// ```gherkin diff --git a/test/issue_assistant_app_test.dart b/test/issue_assistant_app_test.dart index a8b1d78..bf1a2d8 100644 --- a/test/issue_assistant_app_test.dart +++ b/test/issue_assistant_app_test.dart @@ -1595,6 +1595,322 @@ projects: }); }); + /// ```gherkin + /// Feature: GitHub webhook forward issue events + /// + /// As a maintainer running the long-lived assistant + /// I want GitHub webhook-forward events to trigger issue evaluation without polling + /// So that comment mentions can be processed through a channel-based source + /// ``` + group('IssueAssistantApp.run webhook-forward', () { + /// ```gherkin + /// Scenario: Reply to a mention received through gh/webhook-forward + /// Given a temporary project checkout that can be used as the local repo path + /// And a gh script that forwards one issue_comment webhook event + /// And a responder that emits a planning reply + /// And an orchestrator-style config that uses gh/webhook-forward + /// 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 gh webhook forward command is invoked for the GitHub project + /// ``` + test('Reply to a mention received through gh/webhook-forward', () async { + // Given a temporary project checkout that can be used as the local repo path. + final sandbox = await Directory.systemTemp.createTemp('cws-webhook-run-'); + final repoDir = await Directory(p.join(sandbox.path, 'repo')).create(); + await initGitRepo(repoDir); + + // And a gh script that forwards one issue_comment webhook event. + final ghScript = File(p.join(sandbox.path, 'gh')); + final ghLog = File(p.join(sandbox.path, 'gh-log.jsonl')); + final postedBody = File(p.join(sandbox.path, 'posted-body.md')); + final issue = { + 'number': 12, + 'title': 'Need channel-based trigger handling', + 'body': 'Please react to webhook-forward.', + 'state': 'open', + 'html_url': 'https://example.test/issues/12', + 'updated_at': '2026-04-08T04:00:00Z', + 'user': {'login': 'reporter'}, + 'labels': const >[], + }; + final comments = [ + { + 'id': 51, + 'body': '@helper can you handle forwarded comments?', + 'html_url': 'https://example.test/issues/12#issuecomment-51', + 'created_at': '2026-04-08T04:00:00Z', + 'updated_at': '2026-04-08T04:00:00Z', + 'user': {'login': 'reporter'}, + }, + ]; + await writeWebhookForwardGhScript( + ghScript: ghScript, + repo: 'owner/sample', + issue: issue, + comments: comments, + ghLog: ghLog, + postedBodyFile: postedBody, + viewerLogin: 'octocat', + ); + + // 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 forwarded issue_comment events', + 'summary': 'posted channel-based planning advice', + }); + + // And an orchestrator-style config that uses gh/webhook-forward. + final configFile = File(p.join(sandbox.path, 'agent-orchestrator.yaml')); + await configFile.writeAsString(''' +defaults: + issueAssistant: + enabled: true + mentionTriggers: ["@helper"] + responders: + - id: primary + command: ${responderScript.path} + timeout: 1m +projects: + sample: + provider: github + repo: owner/sample + path: ${repoDir.path} + issueAssistant: + eventSource: gh/webhook-forward +'''); + final app = await IssueAssistantApp.open( + config: await AppConfig.load(configFile.path), + databasePath: p.join(sandbox.path, 'state.sqlite3'), + ghCommand: ghScript.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 forwarded issue_comment events'), + ); + expect(postedComment, contains('@octocat')); + + // And the gh webhook forward command is invoked for the GitHub project. + final logLines = await ghLog.readAsLines(); + expect( + logLines.any( + (line) => line.contains( + 'webhook forward --events=issues,issue_comment --repo=owner/sample', + ), + ), + isTrue, + ); + }); + + /// ```gherkin + /// Scenario: Reconcile a missed GitHub webhook event through polling + /// Given a temporary project checkout that can be used as the local repo path + /// And a gh script that starts webhook forwarding without delivering an event + /// And the same gh script 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 gh/webhook-forward 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 gh webhook forward command is still invoked for the GitHub project + /// And the app performs more than one GitHub search request while running + /// ``` + test('Reconcile a missed GitHub webhook 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-webhook-reconcile-', + ); + final repoDir = await Directory(p.join(sandbox.path, 'repo')).create(); + await initGitRepo(repoDir); + + // And a gh script that starts webhook forwarding without delivering an event. + final ghScript = File(p.join(sandbox.path, 'gh')); + final ghLog = File(p.join(sandbox.path, 'gh-log.jsonl')); + final postedBody = File(p.join(sandbox.path, 'posted-body.md')); + final searchCountFile = File(p.join(sandbox.path, 'search-count')); + final issue = { + 'number': 18, + 'title': 'Need reconciliation after a missed webhook', + 'body': 'Please recover this mention from polling.', + 'state': 'open', + 'html_url': 'https://example.test/issues/18', + 'updated_at': '2026-04-08T05:00:00Z', + 'user': {'login': 'reporter'}, + 'labels': const >[], + }; + final searchItem = { + ...issue, + 'repository_url': 'https://api.github.com/repos/owner/sample', + }; + final comments = [ + { + 'id': 181, + 'body': '@helper please recover this even if the webhook was missed', + 'html_url': 'https://example.test/issues/18#issuecomment-181', + 'created_at': '2026-04-08T05:00:00Z', + 'updated_at': '2026-04-08T05:00:00Z', + 'user': {'login': 'reporter'}, + }, + ]; + await ghScript.writeAsString('''#!/usr/bin/env bash +set -euo pipefail +printf '%s\n' "\$*" >> "${ghLog.path}" + +if [[ "\$1" == "api" && "\$2" == "user" ]]; then + cat <<'JSON' +${jsonEncode({'login': 'octocat'})} +JSON + exit 0 +fi + +if [[ "\$1" == "api" && "\$4" == "search/issues" ]]; then + count=0 + if [[ -f "${searchCountFile.path}" ]]; then + count="\$(cat "${searchCountFile.path}")" + fi + count=\$((count + 1)) + printf '%s' "\$count" > "${searchCountFile.path}" + if [[ "\$count" -eq 1 ]]; then + cat <<'JSON' +{"total_count":0,"incomplete_results":false,"items":[]} +JSON + else + cat <<'JSON' +${jsonEncode({ + 'total_count': 1, + 'incomplete_results': false, + 'items': [searchItem], + })} +JSON + fi + exit 0 +fi + +if [[ "\$1" == "api" && "\$4" == "repos/owner/sample/issues/18/comments?per_page=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': 703, 'html_url': 'https://example.test/comment/703', 'body': 'posted'})} +JSON + exit 0 +fi + +if [[ "\$1" == "webhook" && "\$2" == "forward" ]]; then + while true; do + sleep 60 + done +fi + +echo "unexpected gh args: \$*" >&2 +exit 1 +'''); + await Process.run('chmod', ['+x', ghScript.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 webhook events through polling', + 'summary': 'posted reconciliation advice', + }); + + // And an orchestrator-style config that uses gh/webhook-forward 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: github + repo: owner/sample + path: ${repoDir.path} + issueAssistant: + eventSource: gh/webhook-forward +'''); + final app = await IssueAssistantApp.open( + config: await AppConfig.load(configFile.path), + databasePath: p.join(sandbox.path, 'state.sqlite3'), + ghCommand: ghScript.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 webhook events through polling'), + ); + + // And the gh webhook forward command is still invoked for the GitHub project. + final logLines = await ghLog.readAsLines(); + expect( + logLines.any( + (line) => line.contains( + 'webhook forward --events=issues,issue_comment --repo=owner/sample', + ), + ), + isTrue, + ); + + // And the app performs more than one GitHub search request while running. + expect( + 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, + greaterThanOrEqualTo(2), + ); + }); + }); + /// ```gherkin /// Feature: Continuous issue polling /// diff --git a/test/test_support.dart b/test/test_support.dart index e8284eb..dd85f9a 100644 --- a/test/test_support.dart +++ b/test/test_support.dart @@ -440,6 +440,136 @@ JSON await Process.run('chmod', ['+x', responderScript.path]); } +Future writeWebhookForwardGhScript({ + required File ghScript, + required String repo, + required Map issue, + required List> comments, + List> issueListResponse = const [], + File? ghLog, + File? postedBodyFile, + String? viewerLogin, +}) async { + final issueNumber = issue['number'] as int; + final payload = { + 'action': 'created', + 'issue': issue, + 'comment': comments.last, + 'repository': {'full_name': repo}, + }; + final searchItems = issueListResponse + .map( + (searchIssue) => { + ...searchIssue, + 'repository_url': 'https://api.github.com/repos/$repo', + }, + ) + .toList(growable: false); + final searchResponse = jsonEncode({ + 'total_count': searchItems.length, + 'incomplete_results': false, + 'items': searchItems, + }); + + final buffer = StringBuffer() + ..writeln('#!/usr/bin/env bash') + ..writeln('set -euo pipefail'); + + if (ghLog != null) { + buffer.writeln( + r'''printf '%s\n' "$*" >> ''' + '"${ghLog.path}"', + ); + } + + buffer.writeln(r'''if [[ "$1" == "api" && "$2" == "user" ]]; then'''); + buffer + ..writeln(" cat <<'JSON'") + ..writeln(jsonEncode({'login': viewerLogin ?? 'test-user'})) + ..writeln('JSON') + ..writeln(' exit 0') + ..writeln('fi'); + + buffer + ..writeln(r'''if [[ "$1" == "api" && "$4" == "search/issues" ]]; then''') + ..writeln(" cat <<'JSON'") + ..writeln(searchResponse) + ..writeln('JSON') + ..writeln(' exit 0') + ..writeln('fi'); + + buffer + ..writeln( + 'if [[ "\$1" == "api" && "\$4" == ' + '"repos/$repo/issues/$issueNumber/comments?per_page=100" ]]; then', + ) + ..writeln(" cat <<'JSON'") + ..writeln(jsonEncode(comments)) + ..writeln('JSON') + ..writeln(' exit 0') + ..writeln('fi'); + + buffer + ..writeln( + 'if [[ "\$1" == "api" && "\$4" == ' + '"repos/$repo/issues/$issueNumber/comments" ]]; then', + ) + ..writeln(r''' for arg in "$@"; do''') + ..writeln(r''' :'''); + if (postedBodyFile != null) { + buffer + ..writeln(r''' if [[ "$arg" == body=* ]]; then''') + ..writeln( + r''' printf '%s' "${arg#body=}" > ''' + '"${postedBodyFile.path}"', + ) + ..writeln(r''' fi'''); + } + buffer + ..writeln(r''' done''') + ..writeln(" cat <<'JSON'") + ..writeln( + jsonEncode({ + 'id': 702, + 'html_url': 'https://example.test/comment/702', + 'body': 'posted', + }), + ) + ..writeln('JSON') + ..writeln(' exit 0') + ..writeln('fi'); + + buffer + ..writeln(r'''if [[ "$1" == "webhook" && "$2" == "forward" ]]; then''') + ..writeln(r''' url=""''') + ..writeln(r''' for arg in "$@"; do''') + ..writeln(r''' if [[ "$arg" == --url=* ]]; then''') + ..writeln(r''' url="${arg#--url=}"''') + ..writeln(r''' fi''') + ..writeln(r''' done''') + ..writeln(r''' if [[ -z "$url" ]]; then''') + ..writeln(r''' echo "missing --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-GitHub-Event: issue_comment" \''') + ..writeln(r''' --data-binary @- "$url" <<'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 gh args: $*" >&2''') + ..writeln('exit 1'); + + await ghScript.writeAsString(buffer.toString()); + await Process.run('chmod', ['+x', ghScript.path]); +} + Future initGitRepo(Directory directory) async { await GitDir.init(directory.path, allowContent: true); await runGit([