From 536c0a945026ea1e296f913b6a8719779340e90e Mon Sep 17 00:00:00 2001 From: existedinnettw Date: Wed, 8 Apr 2026 12:51:24 +0800 Subject: [PATCH] refactor: extract polling issue event source (#11) * refactor: extract polling issue event source (#10) Decouple tracker event discovery from issue processing so channel-based sources can plug into the runtime later without rewriting reply handling. --- README.md | 18 +- lib/code_work_spawner.dart | 1 + lib/src/issue_assistant_app.dart | 306 +++-------------------- lib/src/issue_event_source.dart | 380 +++++++++++++++++++++++++++++ test/issue_assistant_app_test.dart | 28 ++- test/test_support.dart | 10 +- 6 files changed, 448 insertions(+), 295 deletions(-) create mode 100644 lib/src/issue_event_source.dart diff --git a/README.md b/README.md index ca09f2d..fe647e6 100644 --- a/README.md +++ b/README.md @@ -2,15 +2,17 @@ Repository issue thread assistant with an Agent Orchestrator style YAML config. -It polls configured GitHub or Gitea repositories with `gh` or `tea`, reads -issue threads, and uses CLI responders such as `codex`, `opencode`, or -`copilot-cli` to decide when to reply. It supports multi-repo config, -responder fallback chains, and optional Discord webhook and desktop -notifications. +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 +multi-repo config, responder fallback chains, and optional Discord webhook and +desktop notifications. ## Highlights -- Poll GitHub and Gitea issues across multiple repositories +- Consume GitHub and Gitea issue events across multiple repositories - Trigger replies from mentions or meaningful thread updates - Run planning and bug-verification flows in isolated worktrees - Send optional Discord or desktop notifications for assistant events @@ -44,13 +46,13 @@ dart run bin/code_work_spawner.dart init-config --output agent-orchestrator.yaml dart run build_runner build --delete-conflicting-outputs ``` -Single poll cycle: +Single event-source reconciliation cycle: ```bash dart run bin/code_work_spawner.dart --once ``` -Long-running poller: +Long-running tracker runtime: ```bash dart run bin/code_work_spawner.dart diff --git a/lib/code_work_spawner.dart b/lib/code_work_spawner.dart index d2d92a7..9c44c5a 100644 --- a/lib/code_work_spawner.dart +++ b/lib/code_work_spawner.dart @@ -2,6 +2,7 @@ export 'src/comment_templates.dart'; export 'src/config.dart'; export 'src/database.dart'; export 'src/issue_tracker_client.dart'; +export 'src/issue_event_source.dart'; export 'src/issue_assistant_app.dart'; export 'src/notifier.dart'; export 'src/workspace_manager.dart'; diff --git a/lib/src/issue_assistant_app.dart b/lib/src/issue_assistant_app.dart index e3f9d08..142a56f 100644 --- a/lib/src/issue_assistant_app.dart +++ b/lib/src/issue_assistant_app.dart @@ -1,26 +1,25 @@ import 'dart:async'; import 'dart:convert'; -import 'dart:io'; import 'package:crypto/crypto.dart'; -import 'package:github/github.dart'; import 'package:logging/logging.dart'; import 'cli_responder.dart'; import 'config.dart'; import 'database.dart'; +import 'issue_event_source.dart'; import 'issue_tracker_client.dart'; import 'notifier.dart'; import 'workspace_manager.dart'; class IssueAssistantApp { static final Logger _logger = Logger('code_work_spawner.app'); - static const Duration _pollCoalescingWindow = Duration(seconds: 2); IssueAssistantApp._({ required this.config, required this.database, required this.issueTrackerClient, + required this.issueEventSource, required this.responderRunner, required this.workspaceManager, required this.notifierRegistryFactory, @@ -29,16 +28,11 @@ class IssueAssistantApp { final AppConfig config; final AppDatabase database; final IssueTrackerClient issueTrackerClient; + final IssueEventSource issueEventSource; final CliResponderRunner responderRunner; final WorkspaceManager workspaceManager; final NotifierRegistry Function(ProjectConfig project) notifierRegistryFactory; - final Map _projectPollStates = {}; - final Set> _activePolls = >{}; - Timer? _schedulerTimer; - Completer? _runCompleter; - bool _isClosing = false; - bool _isClosed = false; static Future open({ required AppConfig config, @@ -47,12 +41,18 @@ class IssueAssistantApp { String teaCommand = 'tea', }) async { final database = AppDatabase.open(databasePath); + final issueTrackerClient = IssueTrackerClient( + ghCommand: ghCommand, + teaCommand: teaCommand, + ); return IssueAssistantApp._( config: config, database: database, - issueTrackerClient: IssueTrackerClient( - ghCommand: ghCommand, - teaCommand: teaCommand, + issueTrackerClient: issueTrackerClient, + issueEventSource: PollingIssueEventSource( + database: database, + issueTrackerClient: issueTrackerClient, + logger: _logger, ), responderRunner: CliResponderRunner(), workspaceManager: WorkspaceManager(worktreeRoot: config.worktreeDir), @@ -64,13 +64,6 @@ class IssueAssistantApp { } Future run() async { - if (_isClosed) { - throw StateError('IssueAssistantApp is already closed.'); - } - if (_runCompleter != null) { - return _runCompleter!.future; - } - final enabledProjects = config.projects.values .where((project) => project.issueAssistant.enabled) .toList(growable: false); @@ -78,28 +71,13 @@ class IssueAssistantApp { _log('no enabled projects configured'); return; } - - final completer = Completer(); - _runCompleter = completer; - - final now = DateTime.now(); - for (final project in enabledProjects) { - _projectPollStates.putIfAbsent( - project.key, - () => _ProjectPollState(project: project, nextDueAt: now), - ); - } - unawaited(_triggerDueProjectPolls(source: 'startup')); - - try { - await completer.future; - } finally { - _cancelScheduler(); - } + await issueEventSource.run( + projects: enabledProjects, + onBatch: _processBatch, + ); } Future runOnce() async { - _log('starting poll cycle projects=${config.projects.length}'); final enabledProjects = []; for (final project in config.projects.values) { if (!project.issueAssistant.enabled) { @@ -112,232 +90,23 @@ class IssueAssistantApp { _log('no enabled projects configured'); return; } - await _pollProjects(enabledProjects); - _log('poll cycle completed'); - } - - Future close() async { - if (_isClosed || _isClosing) { - return; - } - - _isClosing = true; - _cancelScheduler(); - _completeRunIfPending(); - await Future.wait(_activePolls.toList(growable: false)); - await database.close(); - _isClosed = true; - _isClosing = false; - } - - Future _triggerDueProjectPolls({required String source}) async { - if (_isClosing || _isClosed) { - return; - } - - final now = DateTime.now(); - final dueStates = _projectPollStates.values - .where((state) => !state.isRunning && !state.nextDueAt.isAfter(now)) - .toList(growable: false); - if (dueStates.isEmpty) { - _scheduleNextPoll(); - return; - } - - late final Future pollFuture; - pollFuture = () async { - for (final state in dueStates) { - state.isRunning = true; - } - final stopwatch = Stopwatch()..start(); - final projects = dueStates - .map((state) => state.project) - .toList(growable: false); - _log( - 'poll tick source=$source projects=${projects.map((p) => p.key).join(",")}', - ); - try { - await _pollProjects(projects); - stopwatch.stop(); - final batchCompletedAt = DateTime.now(); - for (final state in dueStates) { - while (!state.nextDueAt.isAfter(batchCompletedAt)) { - state.nextDueAt = state.nextDueAt.add( - state.project.issueAssistant.pollInterval, - ); - } - } - _log( - 'poll tick source=$source projects=${projects.map((p) => p.key).join(",")} ' - 'duration_ms=${stopwatch.elapsedMilliseconds}', - ); - } catch (error, stackTrace) { - stopwatch.stop(); - _logError( - 'poll tick source=$source projects=${projects.map((p) => p.key).join(",")} ' - 'duration_ms=${stopwatch.elapsedMilliseconds} error=$error', - ); - if (_runCompleter != null) { - _logError( - 'poll tick source=$source will_retry=true ' - 'next_run=scheduled stack_trace=$stackTrace', - ); - } - } finally { - for (final state in dueStates) { - state.isRunning = false; - } - _activePolls.remove(pollFuture); - _scheduleNextPoll(); - } - }(); - - _activePolls.add(pollFuture); - await pollFuture; - } - - void _cancelScheduler() { - _schedulerTimer?.cancel(); - _schedulerTimer = null; - } - - void _scheduleNextPoll() { - if (_isClosing || _isClosed || _runCompleter == null) { - return; - } - _cancelScheduler(); - - DateTime? nextDueAt; - for (final state in _projectPollStates.values) { - if (state.isRunning) { - continue; - } - if (nextDueAt == null || state.nextDueAt.isBefore(nextDueAt)) { - nextDueAt = state.nextDueAt; - } - } - if (nextDueAt == null) { - return; - } - - final scheduledAt = nextDueAt.add(_pollCoalescingWindow); - final delay = scheduledAt.difference(DateTime.now()); - _schedulerTimer = Timer( - delay.isNegative ? Duration.zero : delay, - () => unawaited(_triggerDueProjectPolls(source: 'interval')), + await issueEventSource.runOnce( + projects: enabledProjects, + onBatch: _processBatch, ); } - void _completeRunIfPending() { - final completer = _runCompleter; - if (completer != null && !completer.isCompleted) { - completer.complete(); - } + Future close() async { + await issueEventSource.close(); + await database.close(); } - Future _pollProjects(List projects) async { - final pollRunIds = {}; - final projectsByRepo = { - for (final project in projects) _projectRepoKey(project): project, - }; - final sinceByProject = {}; - final latestSeenByProject = {}; - - for (final project in projects) { - pollRunIds[project.key] = await database.startPollRun(project.key); - final repoDirectory = Directory(project.path); - final repoExists = await repoDirectory.exists(); - final projectState = await database.findProjectState(project.key); - final since = projectState?.lastSeenUpdatedAt == null - ? null - : DateTime.parse(projectState!.lastSeenUpdatedAt!); - sinceByProject[project.key] = since; - latestSeenByProject[project.key] = since; - _log( - 'poll project=${project.key} repo=${project.repo} ' - 'path=${project.path} path_exists=$repoExists ' - 'since=${since?.toUtc().toIso8601String() ?? "initial"}', - ); - } - - try { - final issuesByProject = >{ - for (final project in projects) project.key: [], - }; - for (final provider in IssueTrackerProvider.values) { - final providerProjects = projects - .where((project) => project.provider == provider) - .toList(growable: false); - if (providerProjects.isEmpty) { - continue; - } - final providerSince = providerProjects - .map((project) => sinceByProject[project.key]) - .whereType() - .fold(null, (current, next) { - if (current == null || next.isBefore(current)) { - return next; - } - return current; - }); - - final issues = await issueTrackerClient.fetchUpdatedIssuesForRepos( - provider: provider, - repos: providerProjects.map((project) => project.repoSlug), - since: providerSince, - ); - for (final issue in issues) { - final project = - projectsByRepo[_providerRepoKey(provider, issue.repoSlug)]; - if (project == null) { - continue; - } - final since = sinceByProject[project.key]; - if (since != null && !issue.updatedAt.isAfter(since)) { - continue; - } - issuesByProject[project.key]!.add(issue); - final latestSeen = latestSeenByProject[project.key]; - if (latestSeen == null || issue.updatedAt.isAfter(latestSeen)) { - latestSeenByProject[project.key] = issue.updatedAt; - } - } - } - - for (final project in projects) { - final projectIssues = issuesByProject[project.key]!; - _log('project=${project.key} fetched_issues=${projectIssues.length}'); - _log( - 'project=${project.key} issue_queue=${projectIssues.length} ' - 'max_concurrent=${project.issueAssistant.maxConcurrentIssues}', - ); - await _processIssuesConcurrently(project, projectIssues); - - await database.upsertProjectState( - projectKey: project.key, - repo: project.repo, - lastSeenUpdatedAt: latestSeenByProject[project.key], - ); - await database.finishPollRun( - pollRunId: pollRunIds.remove(project.key)!, - success: true, - ); - _log( - 'poll project=${project.key} completed latest_seen=' - '${latestSeenByProject[project.key]?.toUtc().toIso8601String() ?? "unchanged"}', - ); - } - } catch (error) { - for (final entry in pollRunIds.entries) { - await database.finishPollRun( - pollRunId: entry.value, - success: false, - error: error.toString(), - ); - _log('poll project=${entry.key} failed error=$error'); - } - rethrow; - } + Future _processBatch(IssueEventBatch batch) { + _log( + 'project=${batch.project.key} tracker_source=${batch.sourceKind} ' + 'fetched_issues=${batch.issues.length}', + ); + return _processIssuesConcurrently(batch.project, batch.issues); } Future _processIssuesConcurrently( @@ -796,23 +565,4 @@ class IssueAssistantApp { _log('responder=$responderId issue=$issueKey stream=stderr\n$stderr'); } } - - String _projectRepoKey(ProjectConfig project) { - return _providerRepoKey(project.provider, project.repoSlug); - } - - String _providerRepoKey( - IssueTrackerProvider provider, - RepositorySlug repoSlug, - ) { - return '${provider.name}:${repoSlug.fullName}'; - } -} - -class _ProjectPollState { - _ProjectPollState({required this.project, required this.nextDueAt}); - - final ProjectConfig project; - DateTime nextDueAt; - bool isRunning = false; } diff --git a/lib/src/issue_event_source.dart b/lib/src/issue_event_source.dart new file mode 100644 index 0000000..c69ccba --- /dev/null +++ b/lib/src/issue_event_source.dart @@ -0,0 +1,380 @@ +import 'dart:async'; +import 'dart:io'; + +import 'package:github/github.dart'; +import 'package:logging/logging.dart'; + +import 'config.dart'; +import 'database.dart'; +import 'issue_tracker_client.dart'; + +typedef IssueEventBatchHandler = Future Function(IssueEventBatch batch); + +class IssueEventBatch { + IssueEventBatch({ + required this.project, + required this.issues, + required this.latestSeenUpdatedAt, + required this.sourceKind, + }); + + final ProjectConfig project; + final List issues; + final DateTime? latestSeenUpdatedAt; + final String sourceKind; +} + +abstract class IssueEventSource { + Future run({ + required List projects, + required IssueEventBatchHandler onBatch, + }); + + Future runOnce({ + required List projects, + required IssueEventBatchHandler onBatch, + }); + + Future close(); +} + +class PollingIssueEventSource implements IssueEventSource { + PollingIssueEventSource({ + required this.database, + required this.issueTrackerClient, + required Logger logger, + }) : _logger = logger; + + static const Duration _pollCoalescingWindow = Duration(seconds: 2); + + final AppDatabase database; + final IssueTrackerClient issueTrackerClient; + final Logger _logger; + final Map _projectPollStates = {}; + final Set> _activePolls = >{}; + Timer? _schedulerTimer; + 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 completer = Completer(); + _runCompleter = completer; + + final now = DateTime.now(); + for (final project in projects) { + _projectPollStates.putIfAbsent( + project.key, + () => _ProjectPollState(project: project, nextDueAt: now), + ); + } + unawaited( + _triggerDueProjectPolls( + source: 'startup', + projects: projects, + onBatch: onBatch, + ), + ); + + try { + await completer.future; + } finally { + _cancelScheduler(); + } + } + + @override + Future runOnce({ + required List projects, + required IssueEventBatchHandler onBatch, + }) async { + _log('starting poll cycle projects=${projects.length}'); + await _pollProjects(projects, onBatch: onBatch); + _log('poll cycle completed'); + } + + @override + Future close() async { + if (_isClosed || _isClosing) { + return; + } + + _isClosing = true; + _cancelScheduler(); + _completeRunIfPending(); + await Future.wait(_activePolls.toList(growable: false)); + _isClosed = true; + _isClosing = false; + } + + Future _triggerDueProjectPolls({ + required String source, + required List projects, + required IssueEventBatchHandler onBatch, + }) async { + if (_isClosing || _isClosed) { + return; + } + + final now = DateTime.now(); + final dueStates = _projectPollStates.values + .where((state) => !state.isRunning && !state.nextDueAt.isAfter(now)) + .toList(growable: false); + if (dueStates.isEmpty) { + _scheduleNextPoll(projects: projects, onBatch: onBatch); + return; + } + + late final Future pollFuture; + pollFuture = () async { + for (final state in dueStates) { + state.isRunning = true; + } + final stopwatch = Stopwatch()..start(); + final dueProjects = dueStates + .map((state) => state.project) + .toList(growable: false); + _log( + 'poll tick source=$source projects=${dueProjects.map((p) => p.key).join(",")}', + ); + try { + await _pollProjects(dueProjects, onBatch: onBatch); + stopwatch.stop(); + final batchCompletedAt = DateTime.now(); + for (final state in dueStates) { + while (!state.nextDueAt.isAfter(batchCompletedAt)) { + state.nextDueAt = state.nextDueAt.add( + state.project.issueAssistant.pollInterval, + ); + } + } + _log( + 'poll tick source=$source projects=${dueProjects.map((p) => p.key).join(",")} ' + 'duration_ms=${stopwatch.elapsedMilliseconds}', + ); + } catch (error, stackTrace) { + stopwatch.stop(); + _logError( + 'poll tick source=$source projects=${dueProjects.map((p) => p.key).join(",")} ' + 'duration_ms=${stopwatch.elapsedMilliseconds} error=$error', + ); + if (_runCompleter != null) { + _logError( + 'poll tick source=$source will_retry=true ' + 'next_run=scheduled stack_trace=$stackTrace', + ); + } + } finally { + for (final state in dueStates) { + state.isRunning = false; + } + _activePolls.remove(pollFuture); + _scheduleNextPoll(projects: projects, onBatch: onBatch); + } + }(); + + _activePolls.add(pollFuture); + await pollFuture; + } + + Future _pollProjects( + List projects, { + required IssueEventBatchHandler onBatch, + }) async { + final pollRunIds = {}; + final projectsByRepo = { + for (final project in projects) _projectRepoKey(project): project, + }; + final sinceByProject = {}; + final latestSeenByProject = {}; + + for (final project in projects) { + pollRunIds[project.key] = await database.startPollRun(project.key); + final repoDirectory = Directory(project.path); + final repoExists = await repoDirectory.exists(); + final projectState = await database.findProjectState(project.key); + final since = projectState?.lastSeenUpdatedAt == null + ? null + : DateTime.parse(projectState!.lastSeenUpdatedAt!); + sinceByProject[project.key] = since; + latestSeenByProject[project.key] = since; + _log( + 'poll project=${project.key} repo=${project.repo} ' + 'path=${project.path} path_exists=$repoExists ' + 'since=${since?.toUtc().toIso8601String() ?? "initial"}', + ); + } + + try { + final issuesByProject = >{ + for (final project in projects) project.key: [], + }; + for (final provider in IssueTrackerProvider.values) { + final providerProjects = projects + .where((project) => project.provider == provider) + .toList(growable: false); + if (providerProjects.isEmpty) { + continue; + } + final providerSince = providerProjects + .map((project) => sinceByProject[project.key]) + .whereType() + .fold(null, (current, next) { + if (current == null || next.isBefore(current)) { + return next; + } + return current; + }); + + final issues = await issueTrackerClient.fetchUpdatedIssuesForRepos( + provider: provider, + repos: providerProjects.map((project) => project.repoSlug), + since: providerSince, + ); + for (final issue in issues) { + final project = + projectsByRepo[_providerRepoKey(provider, issue.repoSlug)]; + if (project == null) { + continue; + } + final since = sinceByProject[project.key]; + if (since != null && !issue.updatedAt.isAfter(since)) { + continue; + } + issuesByProject[project.key]!.add(issue); + final latestSeen = latestSeenByProject[project.key]; + if (latestSeen == null || issue.updatedAt.isAfter(latestSeen)) { + latestSeenByProject[project.key] = issue.updatedAt; + } + } + } + + for (final project in projects) { + final projectIssues = issuesByProject[project.key]!; + _log('project=${project.key} fetched_issues=${projectIssues.length}'); + _log( + 'project=${project.key} issue_queue=${projectIssues.length} ' + 'max_concurrent=${project.issueAssistant.maxConcurrentIssues}', + ); + await onBatch( + IssueEventBatch( + project: project, + issues: projectIssues, + latestSeenUpdatedAt: latestSeenByProject[project.key], + sourceKind: 'polling', + ), + ); + + await database.upsertProjectState( + projectKey: project.key, + repo: project.repo, + lastSeenUpdatedAt: latestSeenByProject[project.key], + ); + await database.finishPollRun( + pollRunId: pollRunIds.remove(project.key)!, + success: true, + ); + _log( + 'poll project=${project.key} completed latest_seen=' + '${latestSeenByProject[project.key]?.toUtc().toIso8601String() ?? "unchanged"}', + ); + } + } catch (error) { + for (final entry in pollRunIds.entries) { + await database.finishPollRun( + pollRunId: entry.value, + success: false, + error: error.toString(), + ); + _log('poll project=${entry.key} failed error=$error'); + } + rethrow; + } + } + + void _cancelScheduler() { + _schedulerTimer?.cancel(); + _schedulerTimer = null; + } + + void _scheduleNextPoll({ + required List projects, + required IssueEventBatchHandler onBatch, + }) { + if (_isClosing || _isClosed || _runCompleter == null) { + return; + } + _cancelScheduler(); + + DateTime? nextDueAt; + for (final state in _projectPollStates.values) { + if (state.isRunning) { + continue; + } + if (nextDueAt == null || state.nextDueAt.isBefore(nextDueAt)) { + nextDueAt = state.nextDueAt; + } + } + if (nextDueAt == null) { + return; + } + + final scheduledAt = nextDueAt.add(_pollCoalescingWindow); + final delay = scheduledAt.difference(DateTime.now()); + _schedulerTimer = Timer( + delay.isNegative ? Duration.zero : delay, + () => unawaited( + _triggerDueProjectPolls( + source: 'interval', + projects: projects, + onBatch: onBatch, + ), + ), + ); + } + + void _completeRunIfPending() { + final completer = _runCompleter; + if (completer != null && !completer.isCompleted) { + completer.complete(); + } + } + + String _projectRepoKey(ProjectConfig project) { + return _providerRepoKey(project.provider, project.repoSlug); + } + + String _providerRepoKey( + IssueTrackerProvider provider, + RepositorySlug repoSlug, + ) { + return '${provider.name}:${repoSlug.fullName}'; + } + + void _log(String message) { + _logger.info(message); + } + + void _logError(String message) { + _logger.severe(message); + } +} + +class _ProjectPollState { + _ProjectPollState({required this.project, required this.nextDueAt}); + + final ProjectConfig project; + DateTime nextDueAt; + bool isRunning = false; +} diff --git a/test/issue_assistant_app_test.dart b/test/issue_assistant_app_test.dart index a72cdd6..a8b1d78 100644 --- a/test/issue_assistant_app_test.dart +++ b/test/issue_assistant_app_test.dart @@ -914,7 +914,7 @@ projects: /// And a responder that sleeps for one second while logging each issue number /// And an orchestrator-style config with maxConcurrentIssues set to three /// When the app processes one polling cycle - /// Then the cycle completes in under three seconds + /// Then all responder executions start before the first responder finishes /// And all three issues still receive replies /// ``` test('Process multiple issues concurrently up to the configured limit', () async { @@ -1038,16 +1038,30 @@ projects: ); // When the app processes one polling cycle. - final stopwatch = Stopwatch()..start(); await app.runOnce(); - stopwatch.stop(); await app.close(); - // Then the cycle completes in under three seconds. - expect(stopwatch.elapsed, lessThan(const Duration(seconds: 3))); + // Then all responder executions start before the first responder finishes. final eventLines = await eventLog.readAsLines(); - expect(eventLines.where((line) => line.startsWith('start:')).length, 3); - expect(eventLines.where((line) => line.startsWith('end:')).length, 3); + final startLines = eventLines + .where((line) => line.startsWith('start:')) + .toList(growable: false); + final endLines = eventLines + .where((line) => line.startsWith('end:')) + .toList(growable: false); + final startTimes = + startLines + .map((line) => int.parse(line.split(':').last)) + .toList(growable: false) + ..sort(); + final endTimes = + endLines + .map((line) => int.parse(line.split(':').last)) + .toList(growable: false) + ..sort(); + expect(startLines, hasLength(3)); + expect(endLines, hasLength(3)); + expect(endTimes.first, greaterThanOrEqualTo(startTimes.last)); // And all three issues still receive replies. final logLines = await ghLog.readAsLines(); diff --git a/test/test_support.dart b/test/test_support.dart index 826f490..e8284eb 100644 --- a/test/test_support.dart +++ b/test/test_support.dart @@ -423,10 +423,16 @@ Future writeDelayedLoggingResponderScript( }) async { await responderScript.writeAsString('''#!/usr/bin/env bash set -euo pipefail -printf 'start:%s:%s\n' "\$CWS_ISSUE_NUMBER" "\$(date +%s)" >> "${eventLog.path}" +timestamp_ms() { + python3 - <<'PY' +import time +print(int(time.time() * 1000)) +PY +} +printf 'start:%s:%s\n' "\$CWS_ISSUE_NUMBER" "\$(timestamp_ms)" >> "${eventLog.path}" cat >/dev/null sleep ${delay.inSeconds} -printf 'end:%s:%s\n' "\$CWS_ISSUE_NUMBER" "\$(date +%s)" >> "${eventLog.path}" +printf 'end:%s:%s\n' "\$CWS_ISSUE_NUMBER" "\$(timestamp_ms)" >> "${eventLog.path}" cat <<'JSON' ${jsonEncode(response)} JSON