MarioYang
/
code_work_spawner
forked from bkinnightskytw/code_work_spawner
1
0
Fork 0
Code Issues 1 Pull Requests Activity

Merge pull request #24 from existedinnettw/feat/21 

Refactor polling config for channel event sources
This commit is contained in:
existedinnettw 2026-04-08 22:33:39 +08:00 committed by GitHub
parent 479f44e4b4 5d0fbcc7f9
commit 4df29b1508
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
11 changed files with 639 additions and 77 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

35
examples/README.md
View File

@ -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

12
examples/simple-gitea.yaml
View File

@ -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:

11
examples/simple-github.yaml
View File

@ -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: |

240
lib/src/config.dart
View File

@ -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<String> 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<String, dynamic> root) {
final issueAssistant = defaults['issueAssistant'];
if (issueAssistant is Map<String, dynamic>) {
_assertCamelCaseMap(issueAssistant, scope: 'defaults.issueAssistant');
final eventSource = issueAssistant['eventSource'];
if (eventSource is Map<String, dynamic>) {
_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<String, dynamic> root) {
issueAssistant,
scope: 'projects.${entry.key}.issueAssistant',
);
final eventSource = issueAssistant['eventSource'];
if (eventSource is Map<String, dynamic>) {
_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<String, dynamic> root) {
}
}
void _normalizeLegacyEventSourceConfig(Map<String, dynamic> root) {
void normalizeIssueAssistant(
Map<String, dynamic> issueAssistant, {
required IssueAssistantEventSourceKind? defaultType,
}) {
final rawEventSource = issueAssistant['eventSource'];
Map<String, dynamic>? normalizedEventSource;
if (rawEventSource is String) {
normalizedEventSource = <String, dynamic>{'type': rawEventSource};
} else if (rawEventSource is Map<String, dynamic>) {
normalizedEventSource = Map<String, dynamic>.from(rawEventSource);
} else if (rawEventSource == null) {
if (issueAssistant.containsKey('pollInterval') ||
issueAssistant.containsKey('reconciliation') ||
issueAssistant.containsKey('reconcileInterval')) {
normalizedEventSource = <String, dynamic>{};
}
}
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<String, dynamic>) {
final issueAssistant = defaults['issueAssistant'];
if (issueAssistant is Map<String, dynamic>) {
normalizeIssueAssistant(
issueAssistant,
defaultType: IssueAssistantEventSourceKind.ghPolling,
);
}
}
final projects = root['projects'];
if (projects is Map<String, dynamic>) {
for (final entry in projects.entries) {
final project = entry.value;
if (project is! Map<String, dynamic>) {
continue;
}
final issueAssistant = project['issueAssistant'];
if (issueAssistant is! Map<String, dynamic>) {
continue;
}
normalizeIssueAssistant(
issueAssistant,
defaultType: project['provider'] == 'gitea'
? IssueAssistantEventSourceKind.teaPolling
: IssueAssistantEventSourceKind.ghPolling,
);
}
}
}
void _assertCamelCaseMap(Map<String, dynamic> 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
'''

39
lib/src/config_schema.dart
View File

@ -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<String>? mentionTriggers;
final String? prompt;
@ -118,6 +116,34 @@ class IssueAssistantConfigDocument {
Map<String, dynamic> 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<String, dynamic> json,
) => _$IssueAssistantEventSourceDocumentFromJson(json);
Map<String, dynamic> 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 }

50
lib/src/issue_event_source/base.dart
View File

@ -48,6 +48,16 @@ class RoutedIssueEventSource implements IssueEventSource {
required List<ProjectConfig> 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 =
<IssueAssistantEventSourceKind, List<ProjectConfig>>{
for (final eventSource in sources.keys)
@ -98,31 +108,38 @@ class RoutedIssueEventSource implements IssueEventSource {
Iterable<IssueAssistantEventSourceKind> _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,
},
);
}
}

139
test/app_config_test.dart
View File

@ -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<PollingIssueEventSourceConfig>(),
);
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<ChannelIssueEventSourceConfig>(),
);
// 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,
);
});
});
}

7
test/issue_assistant_app_run/gosmee_test.dart
View File

@ -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),

14
test/issue_assistant_app_run/webhook_forward_test.dart
View File

@ -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),

147
test/routed_issue_event_source_test.dart Normal file
View File

@ -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<String> runProjects = <String>[];
final List<String> runOnceProjects = <String>[];
@override
Future<void> run({
required List<ProjectConfig> projects,
required IssueEventBatchHandler onBatch,
}) async {
runProjects.addAll(projects.map((project) => project.key));
}
@override
Future<void> runOnce({
required List<ProjectConfig> projects,
required IssueEventBatchHandler onBatch,
}) async {
runOnceProjects.addAll(projects.map((project) => project.key));
}
@override
Future<void> 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 <CliResponderConfig>[],
capabilities: {AssistantCapability.planning},
notifications: const <NotificationConfig>[],
commentTag: 'code-work-spawner',
commentPrefixTemplate: defaultCommentPrefixTemplate,
commentFooterTemplate: defaultCommentFooterTemplate,
),
);
}

22
test/test_support.dart
View File

@ -576,10 +576,24 @@ Future<void> writeGitHubGosmeeGhScript({
required int webhookId,
required int issueNumber,
required List<Map<String, Object?>> comments,
List<Map<String, Object?>> issueListResponse = const [],
File? ghLog,
File? postedBodyFile,
String? viewerLogin,
}) async {
final searchItems = issueListResponse
.map(
(issue) => <String, Object?>{
...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<void> 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" == '