diff --git a/examples/README.md b/examples/README.md index 7b33ca9..fdf2cae 100644 --- a/examples/README.md +++ b/examples/README.md @@ -11,15 +11,32 @@ GitHub projects default to `gh/polling`, and Gitea projects default to `gh/gosmee` or `gh/webhook-forward`, and Gitea projects can switch to `tea/gosmee`. `gh/gosmee` creates a temporary GitHub webhook that targets a generated gosmee -URL, then keeps the same polling reconciliation path active for startup sync -and missed-event recovery. -`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. -`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. +URL. `gh/webhook-forward` starts `gh webhook forward` for `issues` and +`issue_comment` events. `tea/gosmee` creates a temporary Gitea webhook that +targets a generated gosmee URL. Channel-based event sources always run one +startup reconciliation poll. Set `reconciliation: continuous` plus +`reconcileInterval` to keep periodic reconciliation active for missed-event +recovery, or use `reconciliation: startup-only` to skip background polling +after startup. `--once` still uses a single polling reconciliation cycle so +existing backlog handling keeps working. + +Use a nested `issueAssistant.eventSource` object so source-specific fields stay +grouped with the selected source type. For example: + +```yaml +issueAssistant: + eventSource: + type: gh/webhook-forward + reconciliation: continuous + reconcileInterval: 300s +``` + +```yaml +issueAssistant: + eventSource: + type: gh/polling + pollInterval: 30s +``` 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 47fce2f..3b7890a 100644 --- a/examples/simple-gitea.yaml +++ b/examples/simple-gitea.yaml @@ -7,10 +7,14 @@ 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 + # Use tea/gosmee for long-running runs. + eventSource: + type: tea/polling + pollInterval: 30s + # eventSource: + # type: tea/gosmee + # reconciliation: startup-only + # reconcileInterval: 30s maxConcurrentIssues: 3 mentionTriggers: ["@codex", "@helper"] responders: diff --git a/examples/simple-github.yaml b/examples/simple-github.yaml index 7cc89cb..fd0f65c 100644 --- a/examples/simple-github.yaml +++ b/examples/simple-github.yaml @@ -7,9 +7,14 @@ worktreeDir: ~/.worktrees defaults: issueAssistant: enabled: true - # Use gh/gosmee for long-running runs while keeping --once on polling. - # eventSource: gh/gosmee - pollInterval: 30s + # Use gh/gosmee or gh/webhook-forward for long-running runs. + eventSource: + type: gh/polling + pollInterval: 30s + # eventSource: + # type: gh/gosmee + # reconciliation: startup-only + # reconcileInterval: 30s maxConcurrentIssues: 3 mentionTriggers: ["@codex", "@helper"] commentPrefixTemplate: | diff --git a/lib/src/config.dart b/lib/src/config.dart index 3a9eb20..9fb2eac 100644 --- a/lib/src/config.dart +++ b/lib/src/config.dart @@ -11,6 +11,7 @@ export 'config_schema.dart' show AssistantCapability, IssueAssistantEventSourceKind, + IssueAssistantReconciliationKind, IssueTrackerProvider, NotificationConfigDocument, NotificationEvent, @@ -83,6 +84,7 @@ class AppConfig { } _assertCamelCaseConfigKeys(root); + _normalizeLegacyEventSourceConfig(root); final document = _parseDocument(root); return _fromDocument(path, document); } @@ -133,8 +135,10 @@ class AppConfig { defaults: AppConfigDefaultsDocument( issueAssistant: IssueAssistantConfigDocument( enabled: true, - eventSource: IssueAssistantEventSourceKind.ghPolling, - pollInterval: '30s', + eventSource: IssueAssistantEventSourceDocument( + type: IssueAssistantEventSourceKind.ghPolling, + pollInterval: '30s', + ), maxConcurrentIssues: 3, mentionTriggers: const ['@codex', '@helper'], ), @@ -192,22 +196,26 @@ class ProjectConfig { final provider = document.provider ?? IssueTrackerProvider.github; var issueAssistant = defaultAssistant.merge(document.issueAssistant); - if (document.issueAssistant?.eventSource == null && - _isPollingEventSource(issueAssistant.eventSource)) { + if (document.issueAssistant?.eventSource?.type == null && + issueAssistant.eventSource is PollingIssueEventSourceConfig) { issueAssistant = issueAssistant.copyWith( - eventSource: provider == IssueTrackerProvider.github - ? IssueAssistantEventSourceKind.ghPolling - : IssueAssistantEventSourceKind.teaPolling, + eventSource: + (issueAssistant.eventSource as PollingIssueEventSourceConfig) + .copyWith( + type: provider == IssueTrackerProvider.github + ? IssueAssistantEventSourceKind.ghPolling + : IssueAssistantEventSourceKind.teaPolling, + ), ); } - switch (issueAssistant.eventSource) { + switch (issueAssistant.eventSource.type) { case IssueAssistantEventSourceKind.ghPolling: case IssueAssistantEventSourceKind.ghGosmee: case IssueAssistantEventSourceKind.ghWebhookForward: if (provider != IssueTrackerProvider.github) { throw FormatException( 'projects.$key.issueAssistant.eventSource ' - '${_eventSourceConfigValue(issueAssistant.eventSource)} ' + '${_eventSourceConfigValue(issueAssistant.eventSource.type)} ' 'requires provider: github.', ); } @@ -257,7 +265,6 @@ class IssueAssistantConfig { IssueAssistantConfig({ required this.enabled, required this.eventSource, - required this.pollInterval, required this.maxConcurrentIssues, required this.mentionTriggers, required this.prompt, @@ -270,8 +277,7 @@ class IssueAssistantConfig { }); final bool enabled; - final IssueAssistantEventSourceKind eventSource; - final Duration pollInterval; + final IssueEventSourceConfig eventSource; final int maxConcurrentIssues; final List mentionTriggers; final String prompt; @@ -297,11 +303,10 @@ class IssueAssistantConfig { ); } - IssueAssistantConfig copyWith({IssueAssistantEventSourceKind? eventSource}) { + IssueAssistantConfig copyWith({IssueEventSourceConfig? eventSource}) { return IssueAssistantConfig( enabled: enabled, eventSource: eventSource ?? this.eventSource, - pollInterval: pollInterval, maxConcurrentIssues: maxConcurrentIssues, mentionTriggers: mentionTriggers, prompt: prompt, @@ -315,6 +320,56 @@ class IssueAssistantConfig { } } +sealed class IssueEventSourceConfig { + const IssueEventSourceConfig({required this.type}); + + final IssueAssistantEventSourceKind type; + + bool get usesChannelEventSource => !_isPollingEventSource(type); +} + +class PollingIssueEventSourceConfig extends IssueEventSourceConfig { + const PollingIssueEventSourceConfig({ + required super.type, + required this.pollInterval, + }); + + final Duration pollInterval; + + PollingIssueEventSourceConfig copyWith({ + IssueAssistantEventSourceKind? type, + Duration? pollInterval, + }) { + return PollingIssueEventSourceConfig( + type: type ?? this.type, + pollInterval: pollInterval ?? this.pollInterval, + ); + } +} + +class ChannelIssueEventSourceConfig extends IssueEventSourceConfig { + const ChannelIssueEventSourceConfig({ + required super.type, + required this.reconciliation, + required this.reconcileInterval, + }); + + final IssueAssistantReconciliationKind reconciliation; + final Duration reconcileInterval; + + ChannelIssueEventSourceConfig copyWith({ + IssueAssistantEventSourceKind? type, + IssueAssistantReconciliationKind? reconciliation, + Duration? reconcileInterval, + }) { + return ChannelIssueEventSourceConfig( + type: type ?? this.type, + reconciliation: reconciliation ?? this.reconciliation, + reconcileInterval: reconcileInterval ?? this.reconcileInterval, + ); + } +} + class CliResponderConfig { CliResponderConfig({ required this.id, @@ -407,13 +462,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), + eventSource: _resolveEventSource( + base: base?.eventSource, + overlay: overlay?.eventSource, + ), maxConcurrentIssues: _resolvePositiveInt( value: overlay?.maxConcurrentIssues, fallback: base?.maxConcurrentIssues ?? 3, @@ -452,10 +504,54 @@ class _IssueAssistantConfigResolver { ); } - IssueAssistantEventSourceKind _defaultEventSourceForProvider( - IssueAssistantConfig? base, - ) { - return base?.eventSource ?? IssueAssistantEventSourceKind.ghPolling; + IssueEventSourceConfig _resolveEventSource({ + required IssueEventSourceConfig? base, + required IssueAssistantEventSourceDocument? overlay, + }) { + final type = + overlay?.type ?? base?.type ?? IssueAssistantEventSourceKind.ghPolling; + if (_isPollingEventSource(type)) { + if (overlay?.reconciliation != null || + overlay?.reconcileInterval != null) { + throw FormatException( + 'Polling event sources do not support reconciliation settings.', + ); + } + final fallbackPollInterval = switch (base) { + PollingIssueEventSourceConfig(:final pollInterval) => pollInterval, + ChannelIssueEventSourceConfig(:final reconcileInterval) => + reconcileInterval, + null => const Duration(seconds: 30), + }; + return PollingIssueEventSourceConfig( + type: type, + pollInterval: overlay?.pollInterval != null + ? _parseDuration(overlay!.pollInterval!) + : fallbackPollInterval, + ); + } + + final fallbackReconcileInterval = switch (base) { + ChannelIssueEventSourceConfig(:final reconcileInterval) => + reconcileInterval, + PollingIssueEventSourceConfig(:final pollInterval) => pollInterval, + null => const Duration(seconds: 30), + }; + return ChannelIssueEventSourceConfig( + type: type, + reconciliation: + overlay?.reconciliation ?? + switch (base) { + ChannelIssueEventSourceConfig(:final reconciliation) => + reconciliation, + _ => IssueAssistantReconciliationKind.continuous, + }, + reconcileInterval: overlay?.reconcileInterval != null + ? _parseDuration(overlay!.reconcileInterval!) + : overlay?.pollInterval != null + ? _parseDuration(overlay!.pollInterval!) + : fallbackReconcileInterval, + ); } } @@ -582,6 +678,13 @@ void _assertCamelCaseConfigKeys(Map root) { final issueAssistant = defaults['issueAssistant']; if (issueAssistant is Map) { _assertCamelCaseMap(issueAssistant, scope: 'defaults.issueAssistant'); + final eventSource = issueAssistant['eventSource']; + if (eventSource is Map) { + _assertCamelCaseMap( + eventSource, + scope: 'defaults.issueAssistant.eventSource', + ); + } final responders = issueAssistant['responders']; if (responders is List) { for (final responder in responders) { @@ -621,6 +724,13 @@ void _assertCamelCaseConfigKeys(Map root) { issueAssistant, scope: 'projects.${entry.key}.issueAssistant', ); + final eventSource = issueAssistant['eventSource']; + if (eventSource is Map) { + _assertCamelCaseMap( + eventSource, + scope: 'projects.${entry.key}.issueAssistant.eventSource', + ); + } final responders = issueAssistant['responders']; if (responders is List) { for (final responder in responders) { @@ -648,6 +758,80 @@ void _assertCamelCaseConfigKeys(Map root) { } } +void _normalizeLegacyEventSourceConfig(Map root) { + void normalizeIssueAssistant( + Map issueAssistant, { + required IssueAssistantEventSourceKind? defaultType, + }) { + final rawEventSource = issueAssistant['eventSource']; + Map? normalizedEventSource; + + if (rawEventSource is String) { + normalizedEventSource = {'type': rawEventSource}; + } else if (rawEventSource is Map) { + normalizedEventSource = Map.from(rawEventSource); + } else if (rawEventSource == null) { + if (issueAssistant.containsKey('pollInterval') || + issueAssistant.containsKey('reconciliation') || + issueAssistant.containsKey('reconcileInterval')) { + normalizedEventSource = {}; + } + } + + if (normalizedEventSource == null) { + return; + } + + if (defaultType != null && !normalizedEventSource.containsKey('type')) { + normalizedEventSource['type'] = _eventSourceConfigValue(defaultType); + } + + for (final field in [ + 'pollInterval', + 'reconciliation', + 'reconcileInterval', + ]) { + if (issueAssistant.containsKey(field) && + !normalizedEventSource.containsKey(field)) { + normalizedEventSource[field] = issueAssistant.remove(field); + } + } + + issueAssistant['eventSource'] = normalizedEventSource; + } + + final defaults = root['defaults']; + if (defaults is Map) { + final issueAssistant = defaults['issueAssistant']; + if (issueAssistant is Map) { + normalizeIssueAssistant( + issueAssistant, + defaultType: IssueAssistantEventSourceKind.ghPolling, + ); + } + } + + final projects = root['projects']; + if (projects is Map) { + for (final entry in projects.entries) { + final project = entry.value; + if (project is! Map) { + continue; + } + final issueAssistant = project['issueAssistant']; + if (issueAssistant is! Map) { + continue; + } + normalizeIssueAssistant( + issueAssistant, + defaultType: project['provider'] == 'gitea' + ? IssueAssistantEventSourceKind.teaPolling + : IssueAssistantEventSourceKind.ghPolling, + ); + } + } +} + void _assertCamelCaseMap(Map map, {required String scope}) { for (final key in map.keys) { if (key.contains('_')) { @@ -698,7 +882,6 @@ ${_YamlEmitter().write({'defaults': defaultsSection})} # capabilities: ["planning", "bug_verification"] # commentTag: code-work-spawner - # eventSource: gh/polling # commentPrefixTemplate: | ${_commentBlock(defaultCommentPrefixTemplate, indent: ' # ')} # commentFooterTemplate: | @@ -727,7 +910,10 @@ ${_YamlEmitter().write({'projects': projectsSection})} # issueAssistant: # enabled: ${assistant?.enabled ?? true} - # eventSource: gh/gosmee + # eventSource: + # type: gh/gosmee + # reconciliation: startup-only + # reconcileInterval: 30s # mentionTriggers: ["@helper"] # sessionPrefix: sample ''' diff --git a/lib/src/config_schema.dart b/lib/src/config_schema.dart index de5ebac..8e82bd4 100644 --- a/lib/src/config_schema.dart +++ b/lib/src/config_schema.dart @@ -87,7 +87,6 @@ class IssueAssistantConfigDocument { const IssueAssistantConfigDocument({ this.enabled, this.eventSource, - this.pollInterval, this.maxConcurrentIssues, this.mentionTriggers, this.prompt, @@ -100,8 +99,7 @@ class IssueAssistantConfigDocument { }); final bool? enabled; - final IssueAssistantEventSourceKind? eventSource; - final String? pollInterval; + final IssueAssistantEventSourceDocument? eventSource; final int? maxConcurrentIssues; final List? mentionTriggers; final String? prompt; @@ -118,6 +116,34 @@ class IssueAssistantConfigDocument { Map toJson() => _$IssueAssistantConfigDocumentToJson(this); } +@JsonSerializable( + checked: true, + createToJson: true, + disallowUnrecognizedKeys: true, + explicitToJson: true, + includeIfNull: false, +) +class IssueAssistantEventSourceDocument { + const IssueAssistantEventSourceDocument({ + this.type, + this.pollInterval, + this.reconciliation, + this.reconcileInterval, + }); + + final IssueAssistantEventSourceKind? type; + final String? pollInterval; + final IssueAssistantReconciliationKind? reconciliation; + final String? reconcileInterval; + + factory IssueAssistantEventSourceDocument.fromJson( + Map json, + ) => _$IssueAssistantEventSourceDocumentFromJson(json); + + Map toJson() => + _$IssueAssistantEventSourceDocumentToJson(this); +} + @JsonSerializable( checked: true, createToJson: true, @@ -194,6 +220,13 @@ enum IssueAssistantEventSourceKind { teaGosmee, } +enum IssueAssistantReconciliationKind { + @JsonValue('startup-only') + startupOnly, + @JsonValue('continuous') + continuous, +} + @JsonEnum(fieldRename: FieldRename.snake) enum IssueTrackerProvider { github, gitea } diff --git a/lib/src/issue_event_source/base.dart b/lib/src/issue_event_source/base.dart index e1cdc78..6b7a72f 100644 --- a/lib/src/issue_event_source/base.dart +++ b/lib/src/issue_event_source/base.dart @@ -48,6 +48,16 @@ class RoutedIssueEventSource implements IssueEventSource { required List projects, required IssueEventBatchHandler onBatch, }) async { + final startupReconcileProjects = projects + .where(_shouldRunStartupReconciliation) + .toList(growable: false); + if (startupReconcileProjects.isNotEmpty) { + await sources[IssueAssistantEventSourceKind.ghPolling]!.runOnce( + projects: startupReconcileProjects, + onBatch: onBatch, + ); + } + final projectsBySource = >{ for (final eventSource in sources.keys) @@ -98,31 +108,38 @@ class RoutedIssueEventSource implements IssueEventSource { Iterable _registeredSourcesFor( ProjectConfig project, ) sync* { - switch (project.issueAssistant.eventSource) { + switch (project.issueAssistant.eventSource.type) { case IssueAssistantEventSourceKind.ghPolling: yield IssueAssistantEventSourceKind.ghPolling; case IssueAssistantEventSourceKind.teaPolling: yield IssueAssistantEventSourceKind.teaPolling; case IssueAssistantEventSourceKind.ghGosmee: - // Keep reconciliation polling active for startup sync and missed - // webhook recovery while gosmee handles low-latency issue/comment - // events for GitHub projects. - yield IssueAssistantEventSourceKind.ghPolling; + if (_shouldRunContinuousReconciliation(project)) { + yield IssueAssistantEventSourceKind.ghPolling; + } yield IssueAssistantEventSourceKind.ghGosmee; 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; + if (_shouldRunContinuousReconciliation(project)) { + 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; + if (_shouldRunContinuousReconciliation(project)) { + yield IssueAssistantEventSourceKind.teaPolling; + } yield IssueAssistantEventSourceKind.teaGosmee; } } + + bool _shouldRunStartupReconciliation(ProjectConfig project) => + project.issueAssistant.eventSource.usesChannelEventSource; + + bool _shouldRunContinuousReconciliation(ProjectConfig project) => + project.issueAssistant.eventSource.usesChannelEventSource && + project.issueAssistant.eventSource is ChannelIssueEventSourceConfig && + (project.issueAssistant.eventSource as ChannelIssueEventSourceConfig) + .reconciliation == + IssueAssistantReconciliationKind.continuous; } class PollingIssueEventSource implements IssueEventSource { @@ -242,7 +259,12 @@ class PollingIssueEventSource implements IssueEventSource { for (final state in dueStates) { while (!state.nextDueAt.isAfter(batchCompletedAt)) { state.nextDueAt = state.nextDueAt.add( - state.project.issueAssistant.pollInterval, + switch (state.project.issueAssistant.eventSource) { + PollingIssueEventSourceConfig(:final pollInterval) => + pollInterval, + ChannelIssueEventSourceConfig(:final reconcileInterval) => + reconcileInterval, + }, ); } } diff --git a/test/app_config_test.dart b/test/app_config_test.dart index aecb5c3..34f4171 100644 --- a/test/app_config_test.dart +++ b/test/app_config_test.dart @@ -103,9 +103,15 @@ projects: defaultCommentFooterTemplate, ); expect( - config.projects['sample']!.issueAssistant.pollInterval, + (config.projects['sample']!.issueAssistant.eventSource + as PollingIssueEventSourceConfig) + .pollInterval, const Duration(seconds: 30), ); + expect( + config.projects['sample']!.issueAssistant.eventSource, + isA(), + ); expect(config.projects['sample']!.issueAssistant.maxConcurrentIssues, 3); }); @@ -351,7 +357,7 @@ projects: // And the repository slug still parses normally. expect(config.projects['sample']!.repoSlug.fullName, 'owner/sample'); expect( - config.projects['sample']!.issueAssistant.eventSource, + config.projects['sample']!.issueAssistant.eventSource.type, IssueAssistantEventSourceKind.teaPolling, ); }); @@ -374,7 +380,8 @@ projects: repo: owner/sample path: ${tempDir.path} issueAssistant: - eventSource: gh/gosmee + eventSource: + type: gh/gosmee '''); // When loading the app config from disk. @@ -382,7 +389,7 @@ projects: // Then the project keeps gh/gosmee as its event source. expect( - config.projects['sample']!.issueAssistant.eventSource, + config.projects['sample']!.issueAssistant.eventSource.type, IssueAssistantEventSourceKind.ghGosmee, ); @@ -409,7 +416,8 @@ projects: repo: owner/sample path: ${tempDir.path} issueAssistant: - eventSource: gh/gosmee + eventSource: + type: gh/gosmee '''); // When loading the app config from disk. @@ -448,7 +456,8 @@ projects: repo: owner/sample path: ${tempDir.path} issueAssistant: - eventSource: gh/webhook-forward + eventSource: + type: gh/webhook-forward '''); // When loading the app config from disk. @@ -456,7 +465,7 @@ projects: // Then the project keeps gh/webhook-forward as its event source. expect( - config.projects['sample']!.issueAssistant.eventSource, + config.projects['sample']!.issueAssistant.eventSource.type, IssueAssistantEventSourceKind.ghWebhookForward, ); @@ -464,6 +473,95 @@ projects: expect(config.projects['sample']!.provider, IssueTrackerProvider.github); }); + /// ```gherkin + /// Scenario: Load channel reconciliation settings + /// Given a GitHub project config that uses gh/webhook-forward reconciliation settings + /// When loading the app config from disk + /// Then the project keeps the configured reconciliation mode + /// And the configured reconcile interval is parsed for channel recovery + /// ``` + test('Load channel reconciliation settings', () async { + // Given a GitHub project config that uses gh/webhook-forward reconciliation settings. + final tempDir = await Directory.systemTemp.createTemp( + 'cws-channel-reconcile-', + ); + 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: + type: gh/webhook-forward + reconciliation: startup-only + reconcileInterval: 5m +'''); + + // When loading the app config from disk. + final config = await AppConfig.load(configFile.path); + + // Then the project keeps the configured reconciliation mode. + expect( + (config.projects['sample']!.issueAssistant.eventSource + as ChannelIssueEventSourceConfig) + .reconciliation, + IssueAssistantReconciliationKind.startupOnly, + ); + + // And the configured reconcile interval is parsed for channel recovery. + expect( + (config.projects['sample']!.issueAssistant.eventSource + as ChannelIssueEventSourceConfig) + .reconcileInterval, + const Duration(minutes: 5), + ); + }); + + /// ```gherkin + /// Scenario: Reuse pollInterval as the fallback reconcile interval + /// Given a legacy channel-based project config that still sets pollInterval + /// When loading the app config from disk + /// Then the project keeps the parsed poll interval for backward compatibility + /// And the same value is reused as the reconcile interval + /// ``` + test('Reuse pollInterval as the fallback reconcile interval', () async { + // Given a legacy channel-based project config that still sets pollInterval. + final tempDir = await Directory.systemTemp.createTemp( + 'cws-legacy-channel-poll-', + ); + 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: + type: gh/webhook-forward + pollInterval: 45s +'''); + + // When loading the app config from disk. + final config = await AppConfig.load(configFile.path); + + // Then the project keeps the parsed poll interval for backward compatibility. + expect( + config.projects['sample']!.issueAssistant.eventSource, + isA(), + ); + + // And the same value is reused as the reconcile interval. + expect( + (config.projects['sample']!.issueAssistant.eventSource + as ChannelIssueEventSourceConfig) + .reconcileInterval, + const Duration(seconds: 45), + ); + }); + /// ```gherkin /// Scenario: Reject gh/webhook-forward for non-GitHub providers /// Given a Gitea project config that opts into gh/webhook-forward @@ -483,7 +581,8 @@ projects: repo: owner/sample path: ${tempDir.path} issueAssistant: - eventSource: gh/webhook-forward + eventSource: + type: gh/webhook-forward '''); // When loading the app config from disk. @@ -520,7 +619,8 @@ projects: repo: owner/sample path: ${tempDir.path} issueAssistant: - eventSource: tea/gosmee + eventSource: + type: tea/gosmee '''); // When loading the app config from disk. @@ -528,7 +628,7 @@ projects: // Then the project keeps tea/gosmee as its event source. expect( - config.projects['sample']!.issueAssistant.eventSource, + config.projects['sample']!.issueAssistant.eventSource.type, IssueAssistantEventSourceKind.teaGosmee, ); @@ -555,7 +655,8 @@ projects: repo: owner/sample path: ${tempDir.path} issueAssistant: - eventSource: tea/gosmee + eventSource: + type: tea/gosmee '''); // When loading the app config from disk. @@ -719,6 +820,16 @@ projects: .replaceFirst(' # timeout: 10m', ' timeout: 10m') .replaceFirst(' # issueAssistant:', ' issueAssistant:') .replaceFirst(' # enabled: true', ' enabled: true') + .replaceFirst(' # eventSource:', ' eventSource:') + .replaceFirst(' # type: gh/gosmee', ' type: gh/gosmee') + .replaceFirst( + ' # reconciliation: startup-only', + ' reconciliation: startup-only', + ) + .replaceFirst( + ' # reconcileInterval: 30s', + ' reconcileInterval: 30s', + ) .replaceFirst( ' # mentionTriggers: ["@helper"]', ' mentionTriggers: ["@helper"]', @@ -749,6 +860,12 @@ projects: expect(config.projects['sample']!.issueAssistant.mentionTriggers, [ '@helper', ]); + expect( + (config.projects['sample']!.issueAssistant.eventSource + as ChannelIssueEventSourceConfig) + .reconciliation, + IssueAssistantReconciliationKind.startupOnly, + ); }); }); } diff --git a/test/issue_assistant_app_run/gosmee_test.dart b/test/issue_assistant_app_run/gosmee_test.dart index 4498bfb..7a2d4fe 100644 --- a/test/issue_assistant_app_run/gosmee_test.dart +++ b/test/issue_assistant_app_run/gosmee_test.dart @@ -208,7 +208,7 @@ projects: /// 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 + /// And an orchestrator-style config that uses tea/gosmee with continuous reconciliation /// 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 @@ -344,13 +344,12 @@ exit 1 'summary': 'posted reconciliation advice', }); - // And an orchestrator-style config that uses tea/gosmee with a short poll interval. + // And an orchestrator-style config that uses tea/gosmee with continuous reconciliation. final configFile = File(p.join(sandbox.path, 'agent-orchestrator.yaml')); await configFile.writeAsString(''' defaults: issueAssistant: enabled: true - pollInterval: 1s mentionTriggers: ["@helper"] responders: - id: primary @@ -363,6 +362,8 @@ projects: path: ${repoDir.path} issueAssistant: eventSource: tea/gosmee + reconciliation: continuous + reconcileInterval: 1s '''); final app = await IssueAssistantApp.open( config: await AppConfig.load(configFile.path), diff --git a/test/issue_assistant_app_run/webhook_forward_test.dart b/test/issue_assistant_app_run/webhook_forward_test.dart index 2e9c85b..5af5825 100644 --- a/test/issue_assistant_app_run/webhook_forward_test.dart +++ b/test/issue_assistant_app_run/webhook_forward_test.dart @@ -168,6 +168,13 @@ JSON exit 0 fi +if [[ "\$1" == "api" && "\$4" == "search/issues" ]]; then + cat <<'JSON' +{"total_count":0,"incomplete_results":false,"items":[]} +JSON + exit 0 +fi + echo "unexpected gh args: \$*" >&2 exit 1 '''); @@ -232,7 +239,7 @@ projects: /// 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 + /// And an orchestrator-style config that uses gh/webhook-forward with continuous reconciliation /// 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 @@ -348,13 +355,12 @@ exit 1 'summary': 'posted reconciliation advice', }); - // And an orchestrator-style config that uses gh/webhook-forward with a short poll interval. + // And an orchestrator-style config that uses gh/webhook-forward with continuous reconciliation. final configFile = File(p.join(sandbox.path, 'agent-orchestrator.yaml')); await configFile.writeAsString(''' defaults: issueAssistant: enabled: true - pollInterval: 1s mentionTriggers: ["@helper"] responders: - id: primary @@ -367,6 +373,8 @@ projects: path: ${repoDir.path} issueAssistant: eventSource: gh/webhook-forward + reconciliation: continuous + reconcileInterval: 1s '''); final app = await IssueAssistantApp.open( config: await AppConfig.load(configFile.path), diff --git a/test/routed_issue_event_source_test.dart b/test/routed_issue_event_source_test.dart new file mode 100644 index 0000000..c558fb3 --- /dev/null +++ b/test/routed_issue_event_source_test.dart @@ -0,0 +1,147 @@ +import 'package:code_work_spawner/code_work_spawner.dart'; +import 'package:github/github.dart'; +import 'package:test/test.dart'; + +void main() { + /// ```gherkin + /// Feature: Routed issue event sources + /// + /// As a maintainer configuring channel-based event sources + /// I want startup reconciliation and continuous reconciliation to route differently + /// So that channel sources can opt into either startup-only or background recovery polling + /// ``` + group('RoutedIssueEventSource', () { + /// ```gherkin + /// Scenario: Run startup-only reconciliation for channel sources + /// Given a routed event source with fake polling and channel handlers + /// And a project that uses gh/webhook-forward with startup-only reconciliation + /// When the routed event source starts in long-running mode + /// Then the polling source runs once for startup reconciliation + /// And only the channel source stays registered for long-running execution + /// ``` + test('Run startup-only reconciliation for channel sources', () async { + // Given a routed event source with fake polling and channel handlers. + final polling = _RecordingIssueEventSource(); + final webhookForward = _RecordingIssueEventSource(); + final routed = RoutedIssueEventSource( + sources: { + IssueAssistantEventSourceKind.ghPolling: polling, + IssueAssistantEventSourceKind.teaPolling: polling, + IssueAssistantEventSourceKind.ghWebhookForward: webhookForward, + }, + ); + + // And a project that uses gh/webhook-forward with startup-only reconciliation. + final project = _project( + eventSource: IssueAssistantEventSourceKind.ghWebhookForward, + reconciliation: IssueAssistantReconciliationKind.startupOnly, + ); + + // When the routed event source starts in long-running mode. + await routed.run(projects: [project], onBatch: (_) async {}); + + // Then the polling source runs once for startup reconciliation. + expect(polling.runOnceProjects, ['sample']); + + // And only the channel source stays registered for long-running execution. + expect(polling.runProjects, isEmpty); + expect(webhookForward.runProjects, ['sample']); + }); + + /// ```gherkin + /// Scenario: Keep continuous reconciliation polling active for channel sources + /// Given a routed event source with fake polling and channel handlers + /// And a project that uses gh/webhook-forward with continuous reconciliation + /// When the routed event source starts in long-running mode + /// Then the polling source runs once for startup reconciliation + /// And the polling source also stays registered for background recovery + /// ``` + test( + 'Keep continuous reconciliation polling active for channel sources', + () async { + // Given a routed event source with fake polling and channel handlers. + final polling = _RecordingIssueEventSource(); + final webhookForward = _RecordingIssueEventSource(); + final routed = RoutedIssueEventSource( + sources: { + IssueAssistantEventSourceKind.ghPolling: polling, + IssueAssistantEventSourceKind.teaPolling: polling, + IssueAssistantEventSourceKind.ghWebhookForward: webhookForward, + }, + ); + + // And a project that uses gh/webhook-forward with continuous reconciliation. + final project = _project( + eventSource: IssueAssistantEventSourceKind.ghWebhookForward, + reconciliation: IssueAssistantReconciliationKind.continuous, + ); + + // When the routed event source starts in long-running mode. + await routed.run(projects: [project], onBatch: (_) async {}); + + // Then the polling source runs once for startup reconciliation. + expect(polling.runOnceProjects, ['sample']); + + // And the polling source also stays registered for background recovery. + expect(polling.runProjects, ['sample']); + expect(webhookForward.runProjects, ['sample']); + }, + ); + }); +} + +class _RecordingIssueEventSource implements IssueEventSource { + final List runProjects = []; + final List runOnceProjects = []; + + @override + Future run({ + required List projects, + required IssueEventBatchHandler onBatch, + }) async { + runProjects.addAll(projects.map((project) => project.key)); + } + + @override + Future runOnce({ + required List projects, + required IssueEventBatchHandler onBatch, + }) async { + runOnceProjects.addAll(projects.map((project) => project.key)); + } + + @override + Future close() async {} +} + +ProjectConfig _project({ + required IssueAssistantEventSourceKind eventSource, + required IssueAssistantReconciliationKind reconciliation, +}) { + return ProjectConfig( + key: 'sample', + provider: IssueTrackerProvider.github, + repo: 'owner/sample', + repoSlug: RepositorySlug('owner', 'sample'), + path: '/tmp/sample', + defaultBranch: 'main', + sessionPrefix: 'sam', + issueAssistant: IssueAssistantConfig( + enabled: true, + eventSource: ChannelIssueEventSourceConfig( + type: eventSource, + reconciliation: reconciliation, + reconcileInterval: const Duration(seconds: 30), + ), + maxConcurrentIssues: 3, + mentionTriggers: const ['@helper'], + prompt: defaultIssueAssistantPrompt, + responders: const [], + capabilities: {AssistantCapability.planning}, + notifications: const [], + commentTag: 'code-work-spawner', + commentPrefixTemplate: defaultCommentPrefixTemplate, + commentFooterTemplate: defaultCommentFooterTemplate, + ), + ); +} diff --git a/test/test_support.dart b/test/test_support.dart index 5956e25..d3ce6c7 100644 --- a/test/test_support.dart +++ b/test/test_support.dart @@ -576,10 +576,24 @@ Future writeGitHubGosmeeGhScript({ required int webhookId, required int issueNumber, required List> comments, + List> issueListResponse = const [], File? ghLog, File? postedBodyFile, String? viewerLogin, }) async { + final searchItems = issueListResponse + .map( + (issue) => { + ...issue, + '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'); @@ -599,6 +613,14 @@ Future writeGitHubGosmeeGhScript({ ..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" == '