Skip to content

refactor: centralize bounded subprocess supervision (PR 1) #38

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ElbertePlinio merged 4 commits into from
Jul 20, 2026
Merged

refactor: centralize bounded subprocess supervision (PR 1) #38

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
fa17193
refactor: centralize bounded subprocess supervision
ElbertePlinio Jul 20, 2026
969d4aa
fix: harden bounded subprocess supervision
ElbertePlinio Jul 20, 2026
cb903f1
fix: propagate setsid exec failures
ElbertePlinio Jul 20, 2026
910782c
fix: simplify Linux subprocess launcher
ElbertePlinio Jul 20, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions .github/workflows/ci.yml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -89,6 +89,7 @@ jobs:
test/storage/settings_readme_test.dart \
test/storage/settings_test.dart
dart test --concurrency=1 \
test/runner/bounded_subprocess_test.dart \
test/runner/codegen_task_executor_blocking_test.dart \
test/runner/corpus_task_qa_test.dart \
test/runner/evaluator_resource_limits_test.dart \
Expand Down
200 changes: 17 additions & 183 deletions app/lib/agent/droid_agent_harness.dart
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@ import 'dart:io';
import 'package:dart_arena/agent/agent_harness.dart';
import 'package:dart_arena/agent/agent_run_result.dart';
import 'package:dart_arena/providers/factory_custom_model_environment.dart';
import 'package:dart_arena/runner/bounded_subprocess.dart';
import 'package:dart_arena/runner/generated_code_sandbox.dart';
import 'package:dart_arena/runner/subprocess_environment.dart';
import 'package:path/path.dart' as p;
Expand Down Expand Up @@ -175,7 +176,6 @@ $instruction
if (!Platform.isWindows) {
environment['PWD'] = workingDirectory.path;
}
late final Process process;
try {
final processStart = generatedCodeSandbox == null
? SandboxedProcessStart(
Expand All @@ -193,95 +193,38 @@ $instruction
resourceLimits: null,
extraReadOnlyPaths: extraReadOnlyPaths,
);
process = await Process.start(
processStart.executable,
processStart.arguments,
final processResult = await runBoundedSubprocess(
executable: processStart.executable,
arguments: processStart.arguments,
workingDirectory: processStart.workingDirectory,
runInShell: false,
environment: processStart.environment,
includeParentEnvironment: false,
maxOutputBytes: maxEncodedOutputBytes(maxProcessOutputChars),
maxOutputCharacters: maxProcessOutputChars,
timeout: timeout,
capture: BoundedSubprocessCapture.trailing,
);
} on Object {
await cwdProxy?.dispose();
rethrow;
}
try {
final stdoutBuffer = _BoundedTailTextCollector(maxProcessOutputChars);
final stderrBuffer = _BoundedTailTextCollector(maxProcessOutputChars);
final outputLimitExceeded = Completer<void>();
void markOutputLimitExceeded() {
if (!outputLimitExceeded.isCompleted) {
outputLimitExceeded.complete();
}
}

final stdoutDone = process.stdout
.transform(systemEncoding.decoder)
.listen((chunk) {
stdoutBuffer.write(chunk);
if (stdoutBuffer.exceeded) markOutputLimitExceeded();
})
.asFuture<void>();
final stderrDone = process.stderr
.transform(systemEncoding.decoder)
.listen((chunk) {
stderrBuffer.write(chunk);
if (stderrBuffer.exceeded) markOutputLimitExceeded();
})
.asFuture<void>();

var timedOut = false;
var outputLimitHit = false;
int exitCode;
final timeoutExceeded = Completer<void>();
final timeoutTimer = Timer(timeout, () {
if (!timeoutExceeded.isCompleted) timeoutExceeded.complete();
});
try {
final signal = await Future.any<Object>([
process.exitCode,
timeoutExceeded.future.then(
(_) => const _AgentProcessTimeoutExceeded(),
),
outputLimitExceeded.future.then(
(_) => const _AgentProcessOutputLimitExceeded(),
),
]);
if (signal is int) {
exitCode = signal;
} else {
timedOut = signal is _AgentProcessTimeoutExceeded;
outputLimitHit = signal is _AgentProcessOutputLimitExceeded;
await terminateProcessTree(process.pid, ProcessSignal.sigterm);
exitCode = await process.exitCode.timeout(
const Duration(seconds: 2),
onTimeout: () async {
await terminateProcessTree(process.pid, ProcessSignal.sigkill);
return -1;
},
);
}
} finally {
timeoutTimer.cancel();
}
await Future.wait([stdoutDone, stderrDone]);
await cwdProxy?.syncBack();
sw.stop();
final stdoutPreview = _trimPreview(stdoutBuffer.text, 16 * 1024);
final rawStderrPreview = _trimPreview(stderrBuffer.text, 16 * 1024);
final timedOut =
processResult.termination == BoundedSubprocessTermination.timedOut;
final outputLimitHit = processResult.outputLimitExceeded;
final stdoutPreview = _trimPreview(processResult.stdout, 16 * 1024);
final rawStderrPreview = _trimPreview(processResult.stderr, 16 * 1024);
final stderrPreview = outputLimitHit && rawStderrPreview.isEmpty
? 'agent harness output exceeded $maxProcessOutputChars characters'
: rawStderrPreview;

return AgentRunResult(
status: timedOut
? AgentRunStatus.timeout
: exitCode == 0 && !outputLimitHit
: processResult.exitCode == 0 && !outputLimitHit
? AgentRunStatus.success
: AgentRunStatus.failure,
stdoutPreview: stdoutPreview,
stderrPreview: stderrPreview,
exitCode: timedOut && exitCode == -1 ? null : exitCode,
exitCode: timedOut && processResult.exitCode == -1
? null
: processResult.exitCode,
latency: sw.elapsed,
metadata: {
'executable': exe,
Expand Down Expand Up @@ -363,85 +306,6 @@ $instruction
if (value.length <= maxChars) return value;
return value.substring(value.length - maxChars);
}

static Future<void> terminateProcessTree(
int pid,
ProcessSignal signal,
) async {
if (Platform.isWindows) {
final args = ['/PID', '$pid', '/T'];
if (signal == ProcessSignal.sigkill) args.add('/F');
await _tryRunProcess('taskkill', args);
return;
}

final descendants = await _descendantPids(pid);
for (final childPid in descendants.reversed) {
_killPid(childPid, signal);
}
_killPid(pid, signal);
}

static Future<List<int>> _descendantPids(int pid) async {
final descendants = <int>[];
for (final childPid in await _childPids(pid)) {
descendants.add(childPid);
descendants.addAll(await _descendantPids(childPid));
}
return descendants;
}

static Future<List<int>> _childPids(int pid) async {
final pgrep = await _tryRunProcess('pgrep', ['-P', '$pid']);
if (pgrep?.exitCode == 0) {
return _parsePids(pgrep!.stdout.toString());
}

final ps = await _tryRunProcess('ps', ['-o', 'pid=', '--ppid', '$pid']);
if (ps?.exitCode == 0) {
return _parsePids(ps!.stdout.toString());
}
return const [];
}

static Future<ProcessResult?> _tryRunProcess(
String executable,
List<String> arguments,
) async {
try {
return await Process.run(
executable,
arguments,
runInShell: false,
environment: benchmarkSubprocessEnvironment(),
includeParentEnvironment: false,
);
} on Object {
return null;
}
}

static List<int> _parsePids(String output) => output
.split(RegExp(r'\s+'))
.map((s) => int.tryParse(s.trim()))
.whereType<int>()
.toList(growable: false);

static void _killPid(int pid, ProcessSignal signal) {
if (!_tryKillPid(pid, signal)) {
_tryKillPid(pid, null);
}
}

static bool _tryKillPid(int pid, ProcessSignal? signal) {
try {
return signal == null
? Process.killPid(pid)
: Process.killPid(pid, signal);
} on Object {
return false;
}
}
}

class _WorkingDirectoryProxy {
Expand All @@ -463,33 +327,3 @@ class _WorkingDirectoryProxy {
if (await root.exists()) await root.delete(recursive: true);
}
}

class _AgentProcessTimeoutExceeded {
const _AgentProcessTimeoutExceeded();
}

class _AgentProcessOutputLimitExceeded {
const _AgentProcessOutputLimitExceeded();
}

class _BoundedTailTextCollector {
_BoundedTailTextCollector(this.maxChars);

final int maxChars;
final _buffer = StringBuffer();
var exceeded = false;

void write(String chunk) {
if (chunk.isEmpty) return;
_buffer.write(chunk);
if (_buffer.length > maxChars) {
exceeded = true;
final value = _buffer.toString();
_buffer
..clear()
..write(value.substring(value.length - maxChars));
}
}

String get text => _buffer.toString();
}
92 changes: 16 additions & 76 deletions app/lib/agent/minimal_agent_harness.dart
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,9 +3,9 @@ import 'dart:io';

import 'package:dart_arena/agent/agent_harness.dart';
import 'package:dart_arena/agent/agent_run_result.dart';
import 'package:dart_arena/agent/droid_agent_harness.dart';
import 'package:dart_arena/providers/model_provider.dart';
import 'package:dart_arena/providers/model_stream_event.dart';
import 'package:dart_arena/runner/bounded_subprocess.dart';
import 'package:dart_arena/runner/generated_code_sandbox.dart';
import 'package:dart_arena/runner/subprocess_environment.dart';

Expand Down Expand Up @@ -229,79 +229,27 @@ class MinimalAgentHarness implements AgentHarness, AgentHarnessProvenance {
environment: environment,
allowInternet: allowInternet,
);
final process = await Process.start(
processStart.executable,
processStart.arguments,
final result = await runBoundedSubprocess(
executable: processStart.executable,
arguments: processStart.arguments,
workingDirectory: processStart.workingDirectory,
runInShell: false,
environment: processStart.environment,
includeParentEnvironment: false,
maxOutputBytes: maxEncodedOutputBytes(maxOutputChars),
maxOutputCharacters: maxOutputChars,
timeout: timeout,
capture: BoundedSubprocessCapture.trailing,
);
final stdout = _BoundedTailText(maxOutputChars);
final stderr = _BoundedTailText(maxOutputChars);
final outputLimitExceeded = Completer<void>();
void markOutputLimitExceeded() {
if (!outputLimitExceeded.isCompleted) outputLimitExceeded.complete();
if (result.termination == BoundedSubprocessTermination.timedOut) {
throw TimeoutException('bash command timed out');
}

final stdoutDone = process.stdout.transform(systemEncoding.decoder).listen((
chunk,
) {
stdout.write(chunk);
if (stdout.exceeded) markOutputLimitExceeded();
}).asFuture<void>();
final stderrDone = process.stderr.transform(systemEncoding.decoder).listen((
chunk,
) {
stderr.write(chunk);
if (stderr.exceeded) markOutputLimitExceeded();
}).asFuture<void>();
final timeoutExceeded = Completer<void>();
final timeoutTimer = Timer(timeout, () {
if (!timeoutExceeded.isCompleted) timeoutExceeded.complete();
});
var outputLimitHit = false;
var timedOut = false;
int exitCode;
try {
final signal = await Future.any<Object>([
process.exitCode,
timeoutExceeded.future.then((_) => const _CommandTimedOut()),
outputLimitExceeded.future.then((_) => const _CommandOutputLimited()),
]);
if (signal is int) {
exitCode = signal;
} else {
timedOut = signal is _CommandTimedOut;
outputLimitHit = signal is _CommandOutputLimited;
await DroidAgentHarness.terminateProcessTree(
process.pid,
ProcessSignal.sigterm,
);
exitCode = await process.exitCode.timeout(
const Duration(seconds: 2),
onTimeout: () async {
await DroidAgentHarness.terminateProcessTree(
process.pid,
ProcessSignal.sigkill,
);
return -1;
},
);
}
} finally {
timeoutTimer.cancel();
}
await Future.wait([stdoutDone, stderrDone]);
if (timedOut) throw TimeoutException('bash command timed out');
if (outputLimitHit) {
throw _CommandOutputLimitExceeded(stdout.text, stderr.text);
if (result.outputLimitExceeded) {
throw _CommandOutputLimitExceeded(result.stdout, result.stderr);
}
return _CommandResult(
exitCode: exitCode,
stdout: stdout.text,
stderr: stderr.text,
outputTruncated: stdout.exceeded || stderr.exceeded,
exitCode: result.exitCode,
stdout: result.stdout,
stderr: result.stderr,
outputTruncated: result.stdoutLimitExceeded || result.stderrLimitExceeded,
);
}

Expand Down Expand Up @@ -399,14 +347,6 @@ class _CommandResult {
final bool outputTruncated;
}

class _CommandTimedOut {
const _CommandTimedOut();
}

class _CommandOutputLimited {
const _CommandOutputLimited();
}

class _CommandOutputLimitExceeded implements Exception {
const _CommandOutputLimitExceeded(this.stdout, this.stderr);

Expand Down
Loading
Loading