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

Merge pull request #49 from existedinnettw/feat/webhook_dtor 

fix: unregister gosmee webhooks on forced shutdown
This commit is contained in:
existedinnettw 2026-04-11 19:57:03 +08:00 committed by insleker
parent 3eed7772ef a1d0ffe8f3
commit e92b8b970c
12 changed files with 750 additions and 436 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
.github/workflows/release.yaml vendored
View File

@ -41,6 +41,9 @@ jobs:
- name: Install dependencies
run: dart pub get
- name: Generate code
run: dart run build_runner build --delete-conflicting-outputs
- name: Build executable
run: dart compile exe bin/code_work_spawner.dart -o dist/code_work_spawner${{ matrix.ext }}

92
lib/src/issue_tracker/issue_event_source/gitea_gosmee.dart
View File

@ -10,6 +10,7 @@ import '../client.dart';
import '../models.dart';
import '../../core/process_launcher.dart';
import 'base.dart';
import 'gosmee_shared.dart';
class GiteaGosmeeIssueEventSource implements IssueEventSource {
GiteaGosmeeIssueEventSource({
@ -206,12 +207,11 @@ class GiteaGosmeeIssueEventSource implements IssueEventSource {
final processes = _gosmeeProcesses.values.toList(growable: false);
_gosmeeProcesses.clear();
for (final process in processes) {
process.kill();
}
for (final process in processes) {
await process.exitCode.catchError((_) => 0);
}
await GosmeeShared.terminateProcesses(
processes: processes,
sourceLabel: 'gitea/gosmee',
logError: _logError,
);
final webhookIds = Map<String, int>.from(_webhookIdsByProject);
_webhookIdsByProject.clear();
@ -315,57 +315,14 @@ class GiteaGosmeeIssueEventSource implements IssueEventSource {
}
Future<String> _createGosmeeChannelUrl(String localUrl) async {
final result = await commandExecutor.run(gosmeeCommand, [
'--output',
'json',
'client',
'--new-url',
localUrl,
]);
_logDebug(
'gitea/gosmee channel_command_result local_url=$localUrl '
'exit_code=${result.exitCode} stdout=${_summarizeText(result.stdout.toString())} '
'stderr=${_summarizeText(result.stderr.toString())}',
return GosmeeShared.createChannelUrl(
commandExecutor: commandExecutor,
gosmeeCommand: gosmeeCommand,
localUrl: localUrl,
sourceLabel: 'gitea/gosmee',
log: _log,
logDebug: _logDebug,
);
if (result.exitCode != 0) {
throw ProcessException(
gosmeeCommand,
['--output', 'json', 'client', '--new-url', localUrl],
'${result.stdout}\n${result.stderr}',
result.exitCode,
);
}
final stdoutText = result.stdout.toString().trim();
if (stdoutText.isEmpty) {
throw const FormatException('gosmee did not print a public URL.');
}
_log(
'gitea/gosmee channel_raw_output local_url=$localUrl stdout=$stdoutText',
);
try {
final decodedJson = jsonDecode(stdoutText);
if (decodedJson is Map<String, dynamic>) {
final publicUrl = decodedJson['url']?.toString().trim();
if (publicUrl == null || publicUrl.isEmpty) {
throw const FormatException(
'gosmee JSON output did not contain a valid public URL.',
);
}
return publicUrl;
}
if (decodedJson is String) {
final publicUrl = decodedJson.trim();
if (publicUrl.isEmpty) {
throw const FormatException('gosmee did not print a public URL.');
}
return publicUrl;
}
} on FormatException {
if (Uri.tryParse(stdoutText)?.hasScheme ?? false) {
return stdoutText;
}
}
throw const FormatException('gosmee did not print a valid public URL.');
}
void _scheduleFlush() {
@ -429,14 +386,13 @@ class GiteaGosmeeIssueEventSource implements IssueEventSource {
required Stream<List<int>> stream,
required String level,
}) {
utf8.decoder.bind(stream).transform(const LineSplitter()).listen((line) {
final message = 'gitea/gosmee project=${project.key} $level=$line';
if (level == 'stderr') {
_logError(message);
return;
}
_log(message);
});
GosmeeShared.pipeProcessLogs(
logger: _logger,
sourceLabel: 'gitea/gosmee',
projectKey: project.key,
stream: stream,
level: level,
);
}
void _log(String message) {
@ -450,14 +406,6 @@ class GiteaGosmeeIssueEventSource implements IssueEventSource {
void _logError(String message) {
_logger.error(message);
}
String _summarizeText(String value) {
final singleLine = value.trim().replaceAll(RegExp(r'\s+'), ' ');
if (singleLine.length <= 240) {
return singleLine.isEmpty ? '(empty)' : singleLine;
}
return '${singleLine.substring(0, 240)}...';
}
}
class _QueuedGiteaIssue {

95
lib/src/issue_tracker/issue_event_source/github_gosmee.dart
View File

@ -10,6 +10,7 @@ import '../client.dart';
import '../models.dart';
import '../../core/process_launcher.dart';
import 'base.dart';
import 'gosmee_shared.dart';
class GitHubGosmeeIssueEventSource implements IssueEventSource {
GitHubGosmeeIssueEventSource({
@ -206,12 +207,11 @@ class GitHubGosmeeIssueEventSource implements IssueEventSource {
final processes = _gosmeeProcesses.values.toList(growable: false);
_gosmeeProcesses.clear();
for (final process in processes) {
process.kill();
}
for (final process in processes) {
await process.exitCode.catchError((_) => 0);
}
await GosmeeShared.terminateProcesses(
processes: processes,
sourceLabel: 'github/gosmee',
logError: _logError,
);
final webhookIds = Map<String, int>.from(_webhookIdsByProject);
_webhookIdsByProject.clear();
@ -315,60 +315,14 @@ class GitHubGosmeeIssueEventSource implements IssueEventSource {
}
Future<String> _createGosmeeChannelUrl(String localUrl) async {
final result = await commandExecutor.run(gosmeeCommand, [
'--output',
'json',
'client',
'--new-url',
localUrl,
]);
_logDebug(
'github/gosmee channel_command_result local_url=$localUrl '
'exit_code=${result.exitCode} stdout=${_summarizeText(result.stdout.toString())} '
'stderr=${_summarizeText(result.stderr.toString())}',
return GosmeeShared.createChannelUrl(
commandExecutor: commandExecutor,
gosmeeCommand: gosmeeCommand,
localUrl: localUrl,
sourceLabel: 'github/gosmee',
log: _log,
logDebug: _logDebug,
);
if (result.exitCode != 0) {
throw ProcessException(
gosmeeCommand,
['--output', 'json', 'client', '--new-url', localUrl],
'${result.stdout}\n${result.stderr}',
result.exitCode,
);
}
final stdoutText = result.stdout.toString().trim();
if (stdoutText.isEmpty) {
throw const FormatException('gosmee did not print a public URL.');
}
_log(
'github/gosmee channel_raw_output local_url=$localUrl stdout=$stdoutText',
);
try {
final decoded = jsonDecode(stdoutText);
if (decoded is Map<String, dynamic>) {
final url = decoded['url']?.toString().trim();
if (url == null || url.isEmpty) {
throw const FormatException(
'gosmee JSON output did not contain a valid public URL.',
);
}
return url;
}
if (decoded is String) {
final url = decoded.trim();
if (url.isEmpty) {
throw const FormatException('gosmee did not print a public URL.');
}
return url;
}
} on FormatException {
if (Uri.tryParse(stdoutText)?.hasScheme ?? false) {
return stdoutText;
}
}
throw const FormatException('gosmee did not print a valid public URL.');
}
void _scheduleFlush() {
@ -431,14 +385,13 @@ class GitHubGosmeeIssueEventSource implements IssueEventSource {
required Stream<List<int>> stream,
required String level,
}) {
utf8.decoder.bind(stream).transform(const LineSplitter()).listen((line) {
final message = 'github/gosmee project=${project.key} $level=$line';
if (level == 'stderr') {
_logger.error(message);
} else {
_logger.info(message);
}
});
GosmeeShared.pipeProcessLogs(
logger: _logger,
sourceLabel: 'github/gosmee',
projectKey: project.key,
stream: stream,
level: level,
);
}
void _log(String message) => _logger.info(message);
@ -446,14 +399,6 @@ class GitHubGosmeeIssueEventSource implements IssueEventSource {
void _logDebug(String message) => _logger.debug(message);
void _logError(String message) => _logger.error(message);
String _summarizeText(String value) {
final singleLine = value.trim().replaceAll(RegExp(r'\s+'), ' ');
if (singleLine.length <= 240) {
return singleLine.isEmpty ? '(empty)' : singleLine;
}
return '${singleLine.substring(0, 240)}...';
}
}
class _QueuedGitHubIssue {

95
lib/src/issue_tracker/issue_event_source/gitlab_gosmee.dart
View File

@ -8,6 +8,7 @@ import '../../core/process_launcher.dart';
import '../client.dart';
import '../models.dart';
import 'base.dart';
import 'gosmee_shared.dart';
class GitLabGosmeeIssueEventSource implements IssueEventSource {
GitLabGosmeeIssueEventSource({
@ -204,12 +205,11 @@ class GitLabGosmeeIssueEventSource implements IssueEventSource {
final processes = _gosmeeProcesses.values.toList(growable: false);
_gosmeeProcesses.clear();
for (final process in processes) {
process.kill();
}
for (final process in processes) {
await process.exitCode.catchError((_) => 0);
}
await GosmeeShared.terminateProcesses(
processes: processes,
sourceLabel: 'gitlab/gosmee',
logError: _logError,
);
final webhookIds = Map<String, int>.from(_webhookIdsByProject);
_webhookIdsByProject.clear();
@ -353,60 +353,14 @@ class GitLabGosmeeIssueEventSource implements IssueEventSource {
}
Future<String> _createGosmeeChannelUrl(String localUrl) async {
final result = await commandExecutor.run(gosmeeCommand, [
'--output',
'json',
'client',
'--new-url',
localUrl,
]);
_logDebug(
'gitlab/gosmee channel_command_result local_url=$localUrl '
'exit_code=${result.exitCode} stdout=${_summarizeText(result.stdout.toString())} '
'stderr=${_summarizeText(result.stderr.toString())}',
return GosmeeShared.createChannelUrl(
commandExecutor: commandExecutor,
gosmeeCommand: gosmeeCommand,
localUrl: localUrl,
sourceLabel: 'gitlab/gosmee',
log: _log,
logDebug: _logDebug,
);
if (result.exitCode != 0) {
throw ProcessException(
gosmeeCommand,
['--output', 'json', 'client', '--new-url', localUrl],
'${result.stdout}\n${result.stderr}',
result.exitCode,
);
}
final stdoutText = result.stdout.toString().trim();
if (stdoutText.isEmpty) {
throw const FormatException('gosmee did not print a public URL.');
}
_log(
'gitlab/gosmee channel_raw_output local_url=$localUrl stdout=$stdoutText',
);
try {
final decoded = jsonDecode(stdoutText);
if (decoded is Map<String, dynamic>) {
final url = decoded['url']?.toString().trim();
if (url == null || url.isEmpty) {
throw const FormatException(
'gosmee JSON output did not contain a valid public URL.',
);
}
return url;
}
if (decoded is String) {
final url = decoded.trim();
if (url.isEmpty) {
throw const FormatException('gosmee did not print a public URL.');
}
return url;
}
} on FormatException {
if (Uri.tryParse(stdoutText)?.hasScheme ?? false) {
return stdoutText;
}
}
throw const FormatException('gosmee did not print a valid public URL.');
}
void _scheduleFlush() {
@ -469,14 +423,13 @@ class GitLabGosmeeIssueEventSource implements IssueEventSource {
required Stream<List<int>> stream,
required String level,
}) {
utf8.decoder.bind(stream).transform(const LineSplitter()).listen((line) {
final message = 'gitlab/gosmee project=${project.key} $level=$line';
if (level == 'stderr') {
_logger.error(message);
} else {
_logger.info(message);
}
});
GosmeeShared.pipeProcessLogs(
logger: _logger,
sourceLabel: 'gitlab/gosmee',
projectKey: project.key,
stream: stream,
level: level,
);
}
void _log(String message) => _logger.info(message);
@ -484,14 +437,6 @@ class GitLabGosmeeIssueEventSource implements IssueEventSource {
void _logDebug(String message) => _logger.debug(message);
void _logError(String message) => _logger.error(message);
String _summarizeText(String value) {
final singleLine = value.trim().replaceAll(RegExp(r'\s+'), ' ');
if (singleLine.length <= 240) {
return singleLine.isEmpty ? '(empty)' : singleLine;
}
return '${singleLine.substring(0, 240)}...';
}
}
class _QueuedGitLabIssue {

140
lib/src/issue_tracker/issue_event_source/gosmee_shared.dart Normal file
View File

@ -0,0 +1,140 @@
import 'dart:async';
import 'dart:convert';
import 'dart:io';
import '../../config/app_logger.dart';
import '../../core/process_launcher.dart';
class GosmeeShared {
static const Duration processShutdownTimeout = Duration(seconds: 2);
static Future<String> createChannelUrl({
required ExternalCommandExecutor commandExecutor,
required String gosmeeCommand,
required String localUrl,
required String sourceLabel,
required void Function(String message) log,
required void Function(String message) logDebug,
}) async {
final result = await commandExecutor.run(gosmeeCommand, [
'--output',
'json',
'client',
'--new-url',
localUrl,
]);
logDebug(
'$sourceLabel channel_command_result local_url=$localUrl '
'exit_code=${result.exitCode} stdout=${_summarizeText(result.stdout.toString())} '
'stderr=${_summarizeText(result.stderr.toString())}',
);
if (result.exitCode != 0) {
throw ProcessException(
gosmeeCommand,
['--output', 'json', 'client', '--new-url', localUrl],
'${result.stdout}\n${result.stderr}',
result.exitCode,
);
}
final stdoutText = result.stdout.toString().trim();
if (stdoutText.isEmpty) {
throw const FormatException('gosmee did not print a public URL.');
}
log(
'$sourceLabel channel_raw_output local_url=$localUrl stdout=$stdoutText',
);
try {
final decoded = jsonDecode(stdoutText);
if (decoded is Map<String, dynamic>) {
final url = decoded['url']?.toString().trim();
if (url == null || url.isEmpty) {
throw const FormatException(
'gosmee JSON output did not contain a valid public URL.',
);
}
return url;
}
if (decoded is String) {
final url = decoded.trim();
if (url.isEmpty) {
throw const FormatException('gosmee did not print a public URL.');
}
return url;
}
} on FormatException {
if (Uri.tryParse(stdoutText)?.hasScheme ?? false) {
return stdoutText;
}
}
throw const FormatException('gosmee did not print a valid public URL.');
}
static Future<void> terminateProcesses({
required List<StartedExternalCommand> processes,
required String sourceLabel,
required void Function(String message) logError,
}) async {
for (final process in processes) {
process.kill();
}
for (final process in processes) {
await _awaitProcessExit(
process: process,
sourceLabel: sourceLabel,
logError: logError,
);
}
}
static void pipeProcessLogs({
required AppLogger logger,
required String sourceLabel,
required String projectKey,
required Stream<List<int>> stream,
required String level,
}) {
utf8.decoder.bind(stream).transform(const LineSplitter()).listen((line) {
final message = '$sourceLabel project=$projectKey $level=$line';
if (level == 'stderr') {
logger.error(message);
} else {
logger.info(message);
}
});
}
static Future<void> _awaitProcessExit({
required StartedExternalCommand process,
required String sourceLabel,
required void Function(String message) logError,
}) async {
try {
await process.exitCode.timeout(processShutdownTimeout);
return;
} on TimeoutException {
process.kill(ProcessSignal.sigkill);
} catch (_) {
return;
}
try {
await process.exitCode.timeout(processShutdownTimeout);
} on TimeoutException {
logError(
'$sourceLabel process did not exit after sigkill '
'timeout_ms=${processShutdownTimeout.inMilliseconds}',
);
} catch (_) {}
}
static String _summarizeText(String value) {
final singleLine = value.trim().replaceAll(RegExp(r'\s+'), ' ');
if (singleLine.length <= 240) {
return singleLine.isEmpty ? '(empty)' : singleLine;
}
return '${singleLine.substring(0, 240)}...';
}
}

160
test/issue_event_source/github_gosmee_test.dart
View File

@ -6,6 +6,7 @@ import 'package:path/path.dart' as p;
import 'package:test/test.dart';
import '../test_support.dart';
import 'test_helpers.dart';
void registerIssueAssistantAppRunGitHubGosmeeTests() {
/// ```gherkin
@ -29,11 +30,9 @@ void registerIssueAssistantAppRunGitHubGosmeeTests() {
/// ```
test('Reply to a mention received through gh/gosmee', () async {
// Given a temporary project checkout that can be used as the local repo path.
final sandbox = await Directory.systemTemp.createTemp(
'cws-gh-gosmee-run-',
);
final repoDir = await Directory(p.join(sandbox.path, 'repo')).create();
await initGitRepo(repoDir);
final sandboxRepo = await createSandboxRepo(prefix: 'cws-gh-gosmee-run-');
final sandbox = sandboxRepo.sandbox;
final repoDir = sandboxRepo.repoDir;
// And a gosmee script that forwards one GitHub issue_comment webhook event.
final gosmeeScript = File(p.join(sandbox.path, 'gosmee'));
@ -95,26 +94,14 @@ void registerIssueAssistantAppRunGitHubGosmeeTests() {
});
// And an orchestrator-style config that uses gh/gosmee.
final configFile = File(p.join(sandbox.path, 'agent-orchestrator.yaml'));
await configFile.writeAsString('''
defaults:
issueAssistant:
enabled: true
mentionTriggers: ["@helper"]
responders:
- id: primary
command: ${responderScript.path}
timeout: 1m
projects:
sample:
issueTracker:
provider: github
eventSource:
pollInterval: 5m
type: gh/gosmee
repo: owner/sample
path: ${repoDir.path}
''');
final configFile = await writeSingleProjectIssueAssistantConfig(
sandbox: sandbox,
responderPath: responderScript.path,
provider: 'github',
repo: 'owner/sample',
repoPath: repoDir.path,
eventSourceLines: const ['pollInterval: 5m', 'type: gh/gosmee'],
);
final app = await IssueAssistantApp.open(
config: await AppConfig.load(configFile.path),
databasePath: p.join(sandbox.path, 'state.sqlite3'),
@ -132,15 +119,7 @@ projects:
addTearDown(() => AppLogger.removeListener(captureLog));
// When the long-running app starts and receives the forwarded webhook.
final runFuture = app.run();
for (var attempt = 0; attempt < 200; attempt += 1) {
if (await postedBody.exists()) {
break;
}
await Future<void>.delayed(const Duration(milliseconds: 100));
}
await app.close();
await runFuture;
await runAppUntil(app: app, isReady: postedBody.exists);
// Then it reads issue comments and posts exactly one reply for that issue.
expect(await postedBody.exists(), isTrue);
@ -217,6 +196,119 @@ projects:
isTrue,
);
});
/// ```gherkin
/// Scenario: Delete webhook even when gosmee ignores SIGTERM during shutdown
/// Given a temporary project checkout that can be used as the local repo path
/// And a gosmee command that forwards one GitHub webhook event but ignores SIGTERM
/// And a gh script that can manage webhooks and post issue comments
/// And a responder that emits a planning reply
/// And an orchestrator-style config that uses gh/gosmee
/// When the long-running app starts and then closes
/// Then the app still deletes the GitHub webhook for the project
/// ```
test(
'Delete webhook even when gosmee ignores SIGTERM during shutdown',
() async {
// Given a temporary project checkout that can be used as the local repo path.
final sandboxRepo = await createSandboxRepo(
prefix: 'cws-gh-gosmee-shutdown-',
);
final sandbox = sandboxRepo.sandbox;
final repoDir = sandboxRepo.repoDir;
// And a gosmee command that forwards one GitHub webhook event but ignores SIGTERM.
final gosmeeScript = File(p.join(sandbox.path, 'gosmee'));
final issue = {
'number': 21,
'title': 'Shutdown should still unregister webhook',
'body': 'Please react before shutdown.',
'state': 'open',
'html_url': 'https://example.test/issues/21',
'updated_at': '2026-04-08T07:00:00Z',
'user': {'login': 'reporter'},
'labels': const <Map<String, Object?>>[],
};
final comments = [
{
'id': 151,
'body': '@helper verify webhook cleanup on shutdown',
'html_url': 'https://example.test/issues/21#issuecomment-151',
'created_at': '2026-04-08T07:00:00Z',
'updated_at': '2026-04-08T07:00:00Z',
'user': {'login': 'reporter'},
},
];
registerFakeGosmeeCommand(
command: gosmeeScript.path,
publicUrl: 'https://gosmee.example.test/github-shutdown',
payload: {
'action': 'created',
'issue': issue,
'comment': comments.last,
'repository': {'full_name': 'owner/sample'},
},
webhookHeader: 'X-GitHub-Event',
exitOnSigterm: false,
);
// And a gh script that can manage webhooks and post issue comments.
final ghScript = File(p.join(sandbox.path, 'gh'));
final ghLog = File(p.join(sandbox.path, 'gh-log.jsonl'));
final postedBody = File(p.join(sandbox.path, 'posted-body.md'));
await writeGitHubGosmeeGhScript(
ghScript: ghScript,
repo: 'owner/sample',
webhookId: 81,
issueNumber: 21,
comments: comments,
ghLog: ghLog,
postedBodyFile: postedBody,
viewerLogin: 'octocat',
);
// And a responder that emits a planning reply.
final responderScript = File(p.join(sandbox.path, 'responder.sh'));
await writeResponderScript(responderScript, {
'decision': 'reply',
'mode': 'planning',
'markdown':
'Plan:\n- ensure webhook cleanup when shutdown races forwarding',
'summary': 'posted shutdown cleanup advice',
});
// And an orchestrator-style config that uses gh/gosmee.
final configFile = await writeSingleProjectIssueAssistantConfig(
sandbox: sandbox,
responderPath: responderScript.path,
provider: 'github',
repo: 'owner/sample',
repoPath: repoDir.path,
eventSourceLines: const ['pollInterval: 5m', 'type: gh/gosmee'],
);
final app = await IssueAssistantApp.open(
config: await AppConfig.load(configFile.path),
databasePath: p.join(sandbox.path, 'state.sqlite3'),
ghCommand: ghScript.path,
gosmeeCommand: gosmeeScript.path,
responderRunner: fakeResponderRunner(),
commandExecutor: fakeExternalCommandExecutor(),
);
// When the long-running app starts and then closes.
await runAppUntil(app: app, isReady: postedBody.exists);
// Then the app still deletes the GitHub webhook for the project.
final ghLogLines = await ghLog.readAsLines();
expect(
ghLogLines.any(
(line) =>
line.contains('api -X DELETE repos/owner/sample/hooks/81'),
),
isTrue,
);
},
);
});
}

175
test/issue_event_source/gitlab_gosmee_test.dart
View File

@ -6,6 +6,7 @@ import 'package:path/path.dart' as p;
import 'package:test/test.dart';
import '../test_support.dart';
import 'test_helpers.dart';
void registerIssueAssistantAppRunGitLabGosmeeTests() {
/// ```gherkin
@ -29,11 +30,11 @@ void registerIssueAssistantAppRunGitLabGosmeeTests() {
/// ```
test('Reply to a mention received through glab/gosmee', () async {
// Given a temporary project checkout that can be used as the local repo path.
final sandbox = await Directory.systemTemp.createTemp(
'cws-glab-gosmee-run-',
final sandboxRepo = await createSandboxRepo(
prefix: 'cws-glab-gosmee-run-',
);
final repoDir = await Directory(p.join(sandbox.path, 'repo')).create();
await initGitRepo(repoDir);
final sandbox = sandboxRepo.sandbox;
final repoDir = sandboxRepo.repoDir;
// And a gosmee script that forwards one GitLab note webhook event.
final gosmeeScript = File(p.join(sandbox.path, 'gosmee'));
@ -109,26 +110,14 @@ void registerIssueAssistantAppRunGitLabGosmeeTests() {
});
// And an orchestrator-style config that uses glab/gosmee.
final configFile = File(p.join(sandbox.path, 'agent-orchestrator.yaml'));
await configFile.writeAsString('''
defaults:
issueAssistant:
enabled: true
mentionTriggers: ["@helper"]
responders:
- id: primary
command: ${responderScript.path}
timeout: 1m
projects:
sample:
issueTracker:
provider: gitlab
eventSource:
pollInterval: 5m
type: glab/gosmee
repo: group/subgroup/sample
path: ${repoDir.path}
''');
final configFile = await writeSingleProjectIssueAssistantConfig(
sandbox: sandbox,
responderPath: responderScript.path,
provider: 'gitlab',
repo: 'group/subgroup/sample',
repoPath: repoDir.path,
eventSourceLines: const ['pollInterval: 5m', 'type: glab/gosmee'],
);
final app = await IssueAssistantApp.open(
config: await AppConfig.load(configFile.path),
databasePath: p.join(sandbox.path, 'state.sqlite3'),
@ -146,15 +135,7 @@ projects:
addTearDown(() => AppLogger.removeListener(captureLog));
// When the long-running app starts and receives the forwarded webhook.
final runFuture = app.run();
for (var attempt = 0; attempt < 200; attempt += 1) {
if (await postedBody.exists()) {
break;
}
await Future<void>.delayed(const Duration(milliseconds: 100));
}
await app.close();
await runFuture;
await runAppUntil(app: app, isReady: postedBody.exists);
// Then it reads issue comments and posts exactly one reply for that issue.
expect(await postedBody.exists(), isTrue);
@ -236,6 +217,134 @@ projects:
isTrue,
);
});
/// ```gherkin
/// Scenario: Delete webhook even when gosmee ignores SIGTERM during shutdown
/// Given a temporary project checkout that can be used as the local repo path
/// And a gosmee command that forwards one GitLab webhook event but ignores SIGTERM
/// And a glab script that can manage webhooks and post issue comments
/// And a responder that emits a planning reply
/// And an orchestrator-style config that uses glab/gosmee
/// When the long-running app starts and then closes
/// Then the app still deletes the GitLab webhook for the project
/// ```
test(
'Delete webhook even when gosmee ignores SIGTERM during shutdown',
() async {
// Given a temporary project checkout that can be used as the local repo path.
final sandboxRepo = await createSandboxRepo(
prefix: 'cws-glab-gosmee-shutdown-',
);
final sandbox = sandboxRepo.sandbox;
final repoDir = sandboxRepo.repoDir;
// And a gosmee command that forwards one GitLab webhook event but ignores SIGTERM.
final gosmeeScript = File(p.join(sandbox.path, 'gosmee'));
final issue = <String, Object?>{
'iid': 21,
'title': 'Shutdown should still unregister webhook',
'description': 'Please react before shutdown.',
'state': 'opened',
'web_url':
'https://gitlab.example.test/group/subgroup/sample/-/issues/21',
'updated_at': '2026-04-08T07:00:00Z',
'author': {'username': 'reporter'},
'labels': const <String>[],
};
final comments = <Map<String, Object?>>[
{
'id': 151,
'body': '@helper verify webhook cleanup on shutdown',
'created_at': '2026-04-08T07:00:00Z',
'updated_at': '2026-04-08T07:00:00Z',
'author': {'username': 'reporter'},
'noteable_url':
'https://gitlab.example.test/group/subgroup/sample/-/issues/21',
},
];
registerFakeGosmeeCommand(
command: gosmeeScript.path,
publicUrl: 'https://gosmee.example.test/gitlab-shutdown',
payload: {
'object_kind': 'note',
'project': {'path_with_namespace': 'group/subgroup/sample'},
'issue': issue,
'object_attributes': {
'action': 'create',
'note': comments.last['body'],
'noteable_type': 'Issue',
'url': comments.last['noteable_url'],
},
'user': {'username': 'reporter'},
},
webhookHeader: 'X-Gitlab-Event',
exitOnSigterm: false,
);
// And a glab script that can manage webhooks and post issue comments.
final glabScript = File(p.join(sandbox.path, 'glab'));
final glabLog = File(p.join(sandbox.path, 'glab-log.jsonl'));
final postedBody = File(p.join(sandbox.path, 'posted-body.md'));
await writeFakeGlabScript(
glabScript: glabScript,
issueListResponsesByRepo: {
'group/subgroup/sample': const <Map<String, Object?>>[],
},
commentResponsesByRepo: {
'group/subgroup/sample': {21: comments},
},
postCommentIssueNumbersByRepo: {
'group/subgroup/sample': {21},
},
glabLog: glabLog,
postedBodyFile: postedBody,
viewerLogin: 'glab-user',
webhookIdsByRepo: {'group/subgroup/sample': 81},
);
// And a responder that emits a planning reply.
final responderScript = File(p.join(sandbox.path, 'responder.sh'));
await writeResponderScript(responderScript, {
'decision': 'reply',
'mode': 'planning',
'markdown':
'Plan:\n- ensure webhook cleanup when shutdown races forwarding',
'summary': 'posted shutdown cleanup advice',
});
// And an orchestrator-style config that uses glab/gosmee.
final configFile = await writeSingleProjectIssueAssistantConfig(
sandbox: sandbox,
responderPath: responderScript.path,
provider: 'gitlab',
repo: 'group/subgroup/sample',
repoPath: repoDir.path,
eventSourceLines: const ['pollInterval: 5m', 'type: glab/gosmee'],
);
final app = await IssueAssistantApp.open(
config: await AppConfig.load(configFile.path),
databasePath: p.join(sandbox.path, 'state.sqlite3'),
glabCommand: glabScript.path,
gosmeeCommand: gosmeeScript.path,
responderRunner: fakeResponderRunner(),
commandExecutor: fakeExternalCommandExecutor(),
);
// When the long-running app starts and then closes.
await runAppUntil(app: app, isReady: postedBody.exists);
// Then the app still deletes the GitLab webhook for the project.
final glabLogLines = await glabLog.readAsLines();
expect(
glabLogLines.any(
(line) => line.contains(
'api -X DELETE projects/group%2Fsubgroup%2Fsample/hooks/81',
),
),
isTrue,
);
},
);
});
}

217
test/issue_event_source/gosmee_test.dart
View File

@ -7,6 +7,7 @@ import 'package:path/path.dart' as p;
import 'package:test/test.dart';
import '../test_support.dart';
import 'test_helpers.dart';
void registerIssueAssistantAppRunGosmeeTests() {
/// ```gherkin
@ -30,9 +31,9 @@ void registerIssueAssistantAppRunGosmeeTests() {
/// ```
test('Reply to a mention received through tea/gosmee', () async {
// Given a temporary project checkout that can be used as the local repo path.
final sandbox = await Directory.systemTemp.createTemp('cws-gosmee-run-');
final repoDir = await Directory(p.join(sandbox.path, 'repo')).create();
await initGitRepo(repoDir);
final sandboxRepo = await createSandboxRepo(prefix: 'cws-gosmee-run-');
final sandbox = sandboxRepo.sandbox;
final repoDir = sandboxRepo.repoDir;
// And a gosmee script that forwards one Gitea issue_comment webhook event.
final gosmeeScript = File(p.join(sandbox.path, 'gosmee'));
@ -101,26 +102,14 @@ void registerIssueAssistantAppRunGosmeeTests() {
});
// And an orchestrator-style config that uses tea/gosmee.
final configFile = File(p.join(sandbox.path, 'agent-orchestrator.yaml'));
await configFile.writeAsString('''
defaults:
issueAssistant:
enabled: true
mentionTriggers: ["@helper"]
responders:
- id: primary
command: ${responderScript.path}
timeout: 1m
projects:
sample:
issueTracker:
provider: gitea
eventSource:
pollInterval: 5m
type: tea/gosmee
repo: owner/sample
path: ${repoDir.path}
''');
final configFile = await writeSingleProjectIssueAssistantConfig(
sandbox: sandbox,
responderPath: responderScript.path,
provider: 'gitea',
repo: 'owner/sample',
repoPath: repoDir.path,
eventSourceLines: const ['pollInterval: 5m', 'type: tea/gosmee'],
);
await File(p.join(sandbox.path, '.env')).writeAsString('''
CWS_GITEA_HOST=${giteaServer.baseUrl}
CWS_GITEA_TOKEN=test-token
@ -141,15 +130,10 @@ CWS_GITEA_TOKEN=test-token
addTearDown(() => AppLogger.removeListener(captureLog));
// When the long-running app starts and receives the forwarded webhook.
final runFuture = app.run();
for (var attempt = 0; attempt < 200; attempt += 1) {
if (giteaServer.postedBodies.isNotEmpty) {
break;
}
await Future<void>.delayed(const Duration(milliseconds: 100));
}
await app.close();
await runFuture;
await runAppUntil(
app: app,
isReady: () async => giteaServer.postedBodies.isNotEmpty,
);
// Then it reads issue comments and posts exactly one reply for that issue.
expect(giteaServer.postedBodies, hasLength(1));
@ -225,11 +209,11 @@ CWS_GITEA_TOKEN=test-token
/// ```
test('Reconcile a missed Gitea gosmee event through polling', () async {
// Given a temporary project checkout that can be used as the local repo path.
final sandbox = await Directory.systemTemp.createTemp(
'cws-gosmee-reconcile-',
final sandboxRepo = await createSandboxRepo(
prefix: 'cws-gosmee-reconcile-',
);
final repoDir = await Directory(p.join(sandbox.path, 'repo')).create();
await initGitRepo(repoDir);
final sandbox = sandboxRepo.sandbox;
final repoDir = sandboxRepo.repoDir;
// And a gosmee script that starts forwarding without delivering an event.
final gosmeeScript = File(p.join(sandbox.path, 'gosmee'));
@ -296,27 +280,18 @@ CWS_GITEA_TOKEN=test-token
});
// 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
mentionTriggers: ["@helper"]
responders:
- id: primary
command: ${responderScript.path}
timeout: 1m
projects:
sample:
issueTracker:
provider: gitea
eventSource:
type: tea/gosmee
reconciliation: continuous
reconcileInterval: 1s
repo: owner/sample
path: ${repoDir.path}
''');
final configFile = await writeSingleProjectIssueAssistantConfig(
sandbox: sandbox,
responderPath: responderScript.path,
provider: 'gitea',
repo: 'owner/sample',
repoPath: repoDir.path,
eventSourceLines: const [
'type: tea/gosmee',
'reconciliation: continuous',
'reconcileInterval: 1s',
],
);
await File(p.join(sandbox.path, '.env')).writeAsString('''
CWS_GITEA_HOST=${giteaServer.baseUrl}
CWS_GITEA_TOKEN=test-token
@ -330,15 +305,10 @@ CWS_GITEA_TOKEN=test-token
);
// When the long-running app starts and enough time passes for reconciliation polling.
final runFuture = app.run();
for (var attempt = 0; attempt < 200; attempt += 1) {
if (giteaServer.postedBodies.isNotEmpty) {
break;
}
await Future<void>.delayed(const Duration(milliseconds: 100));
}
await app.close();
await runFuture;
await runAppUntil(
app: app,
isReady: () async => giteaServer.postedBodies.isNotEmpty,
);
// Then it posts exactly one reply for the issue discovered by polling.
expect(giteaServer.postedBodies, hasLength(1));
@ -365,6 +335,121 @@ CWS_GITEA_TOKEN=test-token
isTrue,
);
});
/// ```gherkin
/// Scenario: Delete webhook even when gosmee ignores SIGTERM during shutdown
/// Given a temporary project checkout that can be used as the local repo path
/// And a gosmee command that forwards one Gitea webhook event but ignores SIGTERM
/// And a Gitea API fixture that can manage webhooks and post issue comments
/// And a responder that emits a planning reply
/// And an orchestrator-style config that uses tea/gosmee
/// When the long-running app starts and then closes
/// Then the app still deletes the Gitea webhook for the project
/// ```
test('Delete webhook even when gosmee ignores SIGTERM during shutdown', () async {
// Given a temporary project checkout that can be used as the local repo path.
final sandboxRepo = await createSandboxRepo(
prefix: 'cws-gosmee-shutdown-',
);
final sandbox = sandboxRepo.sandbox;
final repoDir = sandboxRepo.repoDir;
// And a gosmee command that forwards one Gitea webhook event but ignores SIGTERM.
final gosmeeScript = File(p.join(sandbox.path, 'gosmee'));
final issue = {
'index': 21,
'title': 'Shutdown should still unregister webhook',
'body': 'Please react before shutdown.',
'state': 'open',
'url': 'https://gitea.example.test/owner/sample/issues/21',
'updated_at': '2026-04-08T07:00:00Z',
'poster': {'login': 'reporter'},
'labels': const <Map<String, Object?>>[],
};
final comments = [
{
'id': 151,
'body': '@helper verify webhook cleanup on shutdown',
'url':
'https://gitea.example.test/owner/sample/issues/21#issuecomment-151',
'created_at': '2026-04-08T07:00:00Z',
'updated_at': '2026-04-08T07:00:00Z',
'poster': {'login': 'reporter'},
},
];
registerFakeGosmeeCommand(
command: gosmeeScript.path,
publicUrl: 'https://gosmee.example.test/shutdown',
payload: {
'action': 'created',
'issue': issue,
'comment': comments.last,
'repository': {'full_name': 'owner/sample'},
},
exitOnSigterm: false,
);
// And a Gitea API fixture that can manage webhooks and post issue comments.
final giteaServer = await FakeGiteaApiServer.start(
issueListResponsesByRepo: {
'owner/sample': const <Map<String, Object?>>[],
},
commentResponsesByRepo: {
'owner/sample': {21: comments},
},
postCommentResponsesByRepo: {
'owner/sample': {
21: {
'id': 803,
'body': 'posted',
'html_url': 'https://gitea.example.test/comment/803',
},
},
},
viewerLogin: 'tea-octocat',
);
addTearDown(giteaServer.close);
// And a responder that emits a planning reply.
final responderScript = File(p.join(sandbox.path, 'responder.sh'));
await writeResponderScript(responderScript, {
'decision': 'reply',
'mode': 'planning',
'markdown':
'Plan:\n- ensure webhook cleanup when shutdown races forwarding',
'summary': 'posted shutdown cleanup advice',
});
// And an orchestrator-style config that uses tea/gosmee.
final configFile = await writeSingleProjectIssueAssistantConfig(
sandbox: sandbox,
responderPath: responderScript.path,
provider: 'gitea',
repo: 'owner/sample',
repoPath: repoDir.path,
eventSourceLines: const ['pollInterval: 5m', 'type: tea/gosmee'],
);
await File(p.join(sandbox.path, '.env')).writeAsString('''
CWS_GITEA_HOST=${giteaServer.baseUrl}
CWS_GITEA_TOKEN=test-token
''');
final app = await IssueAssistantApp.open(
config: await AppConfig.load(configFile.path),
databasePath: p.join(sandbox.path, 'state.sqlite3'),
gosmeeCommand: gosmeeScript.path,
responderRunner: fakeResponderRunner(),
commandExecutor: fakeExternalCommandExecutor(),
);
// When the long-running app starts and then closes.
await runAppUntil(
app: app,
isReady: () async => giteaServer.postedBodies.isNotEmpty,
);
// Then the app still deletes the Gitea webhook for the project.
expect(giteaServer.deletedWebhookIds, contains(81));
});
});
}

68
test/issue_event_source/test_helpers.dart Normal file
View File

@ -0,0 +1,68 @@
import 'dart:io';
import 'package:code_work_spawner/code_work_spawner.dart';
import 'package:path/path.dart' as p;
import '../test_support.dart';
Future<({Directory sandbox, Directory repoDir})> createSandboxRepo({
required String prefix,
}) async {
final sandbox = await Directory.systemTemp.createTemp(prefix);
final repoDir = await Directory(p.join(sandbox.path, 'repo')).create();
await initGitRepo(repoDir);
return (sandbox: sandbox, repoDir: repoDir);
}
Future<File> writeSingleProjectIssueAssistantConfig({
required Directory sandbox,
required String responderPath,
required String provider,
required String repo,
required String repoPath,
required List<String> eventSourceLines,
}) async {
final configFile = File(p.join(sandbox.path, 'agent-orchestrator.yaml'));
final eventSourceYaml = eventSourceLines
.map((line) => ' $line')
.join('\n');
await configFile.writeAsString('''
defaults:
issueAssistant:
enabled: true
mentionTriggers: ["@helper"]
responders:
- id: primary
command: $responderPath
timeout: 1m
projects:
sample:
issueTracker:
provider: $provider
eventSource:
$eventSourceYaml
repo: $repo
path: $repoPath
''');
return configFile;
}
Future<void> runAppUntil({
required IssueAssistantApp app,
required Future<bool> Function() isReady,
int maxAttempts = 200,
Duration pollInterval = const Duration(milliseconds: 100),
}) async {
final runFuture = app.run();
try {
for (var attempt = 0; attempt < maxAttempts; attempt += 1) {
if (await isReady()) {
break;
}
await Future<void>.delayed(pollInterval);
}
} finally {
await app.close();
}
await runFuture;
}

118
test/issue_event_source/webhook_forward_test.dart
View File

@ -7,6 +7,7 @@ import 'package:path/path.dart' as p;
import 'package:test/test.dart';
import '../test_support.dart';
import 'test_helpers.dart';
void registerIssueAssistantAppRunWebhookForwardTests() {
/// ```gherkin
@ -29,9 +30,9 @@ void registerIssueAssistantAppRunWebhookForwardTests() {
/// ```
test('Reply to a mention received through gh/webhook-forward', () async {
// Given a temporary project checkout that can be used as the local repo path.
final sandbox = await Directory.systemTemp.createTemp('cws-webhook-run-');
final repoDir = await Directory(p.join(sandbox.path, 'repo')).create();
await initGitRepo(repoDir);
final sandboxRepo = await createSandboxRepo(prefix: 'cws-webhook-run-');
final sandbox = sandboxRepo.sandbox;
final repoDir = sandboxRepo.repoDir;
// And a gh script that forwards one issue_comment webhook event.
final ghScript = File(p.join(sandbox.path, 'gh'));
@ -77,25 +78,14 @@ void registerIssueAssistantAppRunWebhookForwardTests() {
});
// And an orchestrator-style config that uses gh/webhook-forward.
final configFile = File(p.join(sandbox.path, 'agent-orchestrator.yaml'));
await configFile.writeAsString('''
defaults:
issueAssistant:
enabled: true
mentionTriggers: ["@helper"]
responders:
- id: primary
command: ${responderScript.path}
timeout: 1m
projects:
sample:
issueTracker:
provider: github
eventSource:
type: gh/webhook-forward
repo: owner/sample
path: ${repoDir.path}
''');
final configFile = await writeSingleProjectIssueAssistantConfig(
sandbox: sandbox,
responderPath: responderScript.path,
provider: 'github',
repo: 'owner/sample',
repoPath: repoDir.path,
eventSourceLines: const ['type: gh/webhook-forward'],
);
final app = await IssueAssistantApp.open(
config: await AppConfig.load(configFile.path),
databasePath: p.join(sandbox.path, 'state.sqlite3'),
@ -105,15 +95,7 @@ projects:
);
// When the long-running app starts and receives the forwarded webhook.
final runFuture = app.run();
for (var attempt = 0; attempt < 200; attempt += 1) {
if (await postedBody.exists()) {
break;
}
await Future<void>.delayed(const Duration(milliseconds: 100));
}
await app.close();
await runFuture;
await runAppUntil(app: app, isReady: postedBody.exists);
// Then it reads issue comments and posts exactly one reply for that issue.
expect(await postedBody.exists(), isTrue);
@ -146,11 +128,11 @@ projects:
/// ```
test('Log webhook forward stderr at error level', () async {
// Given a temporary project checkout that can be used as the local repo path.
final sandbox = await Directory.systemTemp.createTemp(
'cws-webhook-stderr-',
final sandboxRepo = await createSandboxRepo(
prefix: 'cws-webhook-stderr-',
);
final repoDir = await Directory(p.join(sandbox.path, 'repo')).create();
await initGitRepo(repoDir);
final sandbox = sandboxRepo.sandbox;
final repoDir = sandboxRepo.repoDir;
// And a gh script that writes a webhook creation failure to stderr while forwarding.
final ghScript = File(p.join(sandbox.path, 'gh'));
@ -215,17 +197,12 @@ projects:
AppLogger.addListener(captureLog);
// When the long-running app starts and webhook-forward emits that stderr line.
final runFuture = app.run();
for (var attempt = 0; attempt < 200; attempt += 1) {
if (records.any(
await runAppUntil(
app: app,
isReady: () async => records.any(
(record) => record.message.contains('error creating webhook'),
)) {
break;
}
await Future<void>.delayed(const Duration(milliseconds: 100));
}
await app.close();
await runFuture;
),
);
AppLogger.removeListener(captureLog);
// Then the app logs the webhook-forward stderr message at error level.
@ -255,11 +232,11 @@ projects:
/// ```
test('Reconcile a missed GitHub webhook event through polling', () async {
// Given a temporary project checkout that can be used as the local repo path.
final sandbox = await Directory.systemTemp.createTemp(
'cws-webhook-reconcile-',
final sandboxRepo = await createSandboxRepo(
prefix: 'cws-webhook-reconcile-',
);
final repoDir = await Directory(p.join(sandbox.path, 'repo')).create();
await initGitRepo(repoDir);
final sandbox = sandboxRepo.sandbox;
final repoDir = sandboxRepo.repoDir;
// And a gh script that starts webhook forwarding without delivering an event.
final ghScript = File(p.join(sandbox.path, 'gh'));
@ -380,27 +357,18 @@ projects:
});
// 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
mentionTriggers: ["@helper"]
responders:
- id: primary
command: ${responderScript.path}
timeout: 1m
projects:
sample:
issueTracker:
provider: github
eventSource:
type: gh/webhook-forward
reconciliation: continuous
reconcileInterval: 1s
repo: owner/sample
path: ${repoDir.path}
''');
final configFile = await writeSingleProjectIssueAssistantConfig(
sandbox: sandbox,
responderPath: responderScript.path,
provider: 'github',
repo: 'owner/sample',
repoPath: repoDir.path,
eventSourceLines: const [
'type: gh/webhook-forward',
'reconciliation: continuous',
'reconcileInterval: 1s',
],
);
final app = await IssueAssistantApp.open(
config: await AppConfig.load(configFile.path),
databasePath: p.join(sandbox.path, 'state.sqlite3'),
@ -410,15 +378,7 @@ projects:
);
// When the long-running app starts and enough time passes for reconciliation polling.
final runFuture = app.run();
for (var attempt = 0; attempt < 200; attempt += 1) {
if (await postedBody.exists()) {
break;
}
await Future<void>.delayed(const Duration(milliseconds: 100));
}
await app.close();
await runFuture;
await runAppUntil(app: app, isReady: postedBody.exists);
// Then it posts exactly one reply for the issue discovered by polling.
expect(await postedBody.exists(), isTrue);

3
test/support/fake_gosmee.dart
View File

@ -24,6 +24,7 @@ void registerFakeGosmeeCommand({
Map<String, Object?>? payload,
File? gosmeeLog,
String webhookHeader = 'X-Gitea-Event',
bool exitOnSigterm = true,
}) {
fakeCommandHandlers[command] = (arguments, {timeout}) async {
if (gosmeeLog != null) {
@ -44,7 +45,7 @@ void registerFakeGosmeeCommand({
if (gosmeeLog != null) {
await _appendLine(gosmeeLog, '${arguments.join(' ')}\n');
}
final process = FakeStartedExternalCommand();
final process = FakeStartedExternalCommand(exitOnSigterm: exitOnSigterm);
if (arguments.length >= 3 &&
arguments[0] == 'client' &&
arguments[1] == publicUrl) {

18
test/support/script_utils.dart
View File

@ -61,6 +61,15 @@ ExternalCommandExecutor fakeExternalCommandExecutor() {
}
class FakeStartedExternalCommand implements StartedExternalCommand {
FakeStartedExternalCommand({
this.exitOnSigterm = true,
this.exitOnSigkill = true,
this.sigkillExitCode = 137,
});
final bool exitOnSigterm;
final bool exitOnSigkill;
final int sigkillExitCode;
final StreamController<List<int>> _stdoutController = StreamController();
final StreamController<List<int>> _stderrController = StreamController();
final Completer<int> _exitCodeCompleter = Completer<int>();
@ -96,9 +105,18 @@ class FakeStartedExternalCommand implements StartedExternalCommand {
@override
bool kill([ProcessSignal signal = ProcessSignal.sigterm]) {
if (signal == ProcessSignal.sigkill) {
if (exitOnSigkill) {
complete(sigkillExitCode);
}
return true;
}
if (signal == ProcessSignal.sigterm && exitOnSigterm) {
complete();
return true;
}
return true;
}
}
bool _fallbackValuesRegistered = false;