forked from admin/VolcanicAgenticCore
feat(scheduler): ScriptRunner through the process spawn policy
Spec: on Windows the command goes through Arguments like the bash tool, since cmd.exe doesn't parse ArgumentList quoting; the workspace is created if missing, and a timeout vs cancellation outcome is spelled out.
This commit is contained in:
EugeneTes
@@ -0,0 +1,242 @@
|
||||
using System.ComponentModel;
|
||||
using System.Diagnostics;
|
||||
using System.Globalization;
|
||||
using System.Text;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using Volcanic.AgenticCore.Abstractions.Processes;
|
||||
|
||||
namespace Volcanic.AgenticCore.Extensions.Scheduler;
|
||||
|
||||
/// <summary>One watcher run. <see cref="TimeZoneId"/> becomes the script's <c>TZ</c> (the target's zone).</summary>
|
||||
public sealed record ScriptRequest(string Command, string Cwd, TimeSpan Timeout, string Target, string JobId, string? TimeZoneId);
|
||||
|
||||
/// <summary>What a watcher run means for the job.</summary>
|
||||
public abstract record ScriptOutcome
|
||||
{
|
||||
private ScriptOutcome() { }
|
||||
|
||||
/// <summary>Exit 0, empty stdout after trimming: the condition isn't met. Costs nothing.</summary>
|
||||
public sealed record Silent : ScriptOutcome;
|
||||
|
||||
/// <summary>Exit 0 with output: the condition is met; <see cref="Text"/> (trimmed) goes to the agent.</summary>
|
||||
public sealed record Output(string Text) : ScriptOutcome;
|
||||
|
||||
/// <summary>Non-zero exit, timeout, a hook that threw, or a shell that didn't start.</summary>
|
||||
public sealed record Failed(int? ExitCode, string StderrTail, bool TimedOut) : ScriptOutcome;
|
||||
}
|
||||
|
||||
/// <summary>Runs watcher scripts. Faked in scheduler tests.</summary>
|
||||
public interface IScriptRunner
|
||||
{
|
||||
/// <summary>Never throws for a script problem (that's <see cref="ScriptOutcome.Failed"/>); throws <see cref="OperationCanceledException"/> after killing the script when <paramref name="cancellationToken"/> is cancelled.</summary>
|
||||
Task<ScriptOutcome> RunAsync(ScriptRequest request, CancellationToken cancellationToken);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Runs a watcher with the <c>bash</c> tool's shell in the target's workspace, stdin closed, with the command, cwd and
|
||||
/// environment from the host's <see cref="IProcessSpawnPolicy"/> (<see cref="ProcessPurpose.ScheduledScript"/>): the same
|
||||
/// filtered environment as <c>bash</c>, <c>TZ</c> = the target's zone, host hooks applied. No session-level
|
||||
/// <see cref="ProcessSpawnEvent"/> runs: a job fires without a live session.
|
||||
/// </summary>
|
||||
public sealed class ScriptRunner(SchedulerOptions options, IProcessSpawnPolicy policy, ILogger<ScriptRunner> logger, TimeProvider? time = null) : IScriptRunner
|
||||
{
|
||||
public const int StderrTailLines = 20;
|
||||
private static readonly TimeSpan PipeGrace = TimeSpan.FromSeconds(5);
|
||||
private readonly TimeProvider _time = time ?? TimeProvider.System;
|
||||
|
||||
public async Task<ScriptOutcome> RunAsync(ScriptRequest request, CancellationToken cancellationToken)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(request);
|
||||
cancellationToken.ThrowIfCancellationRequested();
|
||||
|
||||
ProcessSpawnContext spawn;
|
||||
try
|
||||
{
|
||||
spawn = policy.Prepare(new ProcessSpawnRequest
|
||||
{
|
||||
Purpose = ProcessPurpose.ScheduledScript,
|
||||
Command = request.Command,
|
||||
Cwd = request.Cwd,
|
||||
Scope = request.Target,
|
||||
Origin = request.JobId,
|
||||
TimeZoneId = request.TimeZoneId,
|
||||
});
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
return new ScriptOutcome.Failed(null, $"process environment hook failed: {ex.Message}", false);
|
||||
}
|
||||
|
||||
var psi = StartInfo(spawn.Command);
|
||||
spawn.ApplyTo(psi); // cwd and exactly the prepared environment
|
||||
using var process = new Process { StartInfo = psi };
|
||||
try
|
||||
{
|
||||
Directory.CreateDirectory(spawn.Cwd);
|
||||
process.Start();
|
||||
}
|
||||
catch (Exception ex) when (ex is Win32Exception or InvalidOperationException or IOException or UnauthorizedAccessException)
|
||||
{
|
||||
return new ScriptOutcome.Failed(null, $"failed to start {psi.FileName}: {ex.Message}", false);
|
||||
}
|
||||
process.StandardInput.Close();
|
||||
var stdout = new OutputCapture();
|
||||
var stderr = new OutputCapture();
|
||||
var pumps = Task.WhenAll(PumpAsync(process.StandardOutput, stdout), PumpAsync(process.StandardError, stderr));
|
||||
|
||||
using var timeout = new CancellationTokenSource(request.Timeout, _time);
|
||||
using var linked = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, timeout.Token);
|
||||
var timedOut = false;
|
||||
try
|
||||
{
|
||||
await process.WaitForExitAsync(linked.Token).ConfigureAwait(false);
|
||||
}
|
||||
catch (OperationCanceledException)
|
||||
{
|
||||
Kill(process, request);
|
||||
timedOut = !cancellationToken.IsCancellationRequested;
|
||||
}
|
||||
await DrainAsync(pumps, request).ConfigureAwait(false);
|
||||
cancellationToken.ThrowIfCancellationRequested();
|
||||
|
||||
if (timedOut) return new ScriptOutcome.Failed(null, Tail(stderr.ToString()), true);
|
||||
if (process.ExitCode != 0) return new ScriptOutcome.Failed(process.ExitCode, Tail(stderr.ToString()), false);
|
||||
var text = stdout.ToString().Trim();
|
||||
return text.Length == 0 ? new ScriptOutcome.Silent() : new ScriptOutcome.Output(text);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// The <c>bash</c> tool's shell: <c>/bin/sh -c <command></c> through <see cref="ProcessStartInfo.ArgumentList"/>. On
|
||||
/// Windows <c>cmd.exe</c> doesn't parse the MSVC quoting <c>ArgumentList</c> produces, so the command goes through
|
||||
/// <see cref="ProcessStartInfo.Arguments"/> as <c>/d /s /c "<command>"</c>, exactly as <c>bash</c> does.
|
||||
/// A configured <see cref="SchedulerOptions.Shell"/> always uses <c>ArgumentList</c>.
|
||||
/// </summary>
|
||||
internal ProcessStartInfo StartInfo(string command)
|
||||
{
|
||||
var psi = new ProcessStartInfo
|
||||
{
|
||||
RedirectStandardInput = true,
|
||||
RedirectStandardOutput = true,
|
||||
RedirectStandardError = true,
|
||||
UseShellExecute = false,
|
||||
CreateNoWindow = true,
|
||||
StandardOutputEncoding = Encoding.UTF8,
|
||||
StandardErrorEncoding = Encoding.UTF8,
|
||||
};
|
||||
if (options.Shell.Count > 0)
|
||||
{
|
||||
psi.FileName = options.Shell[0];
|
||||
foreach (var argument in options.Shell.Skip(1)) psi.ArgumentList.Add(argument);
|
||||
psi.ArgumentList.Add(command);
|
||||
}
|
||||
else if (OperatingSystem.IsWindows())
|
||||
{
|
||||
psi.FileName = "cmd.exe";
|
||||
psi.Arguments = $"/d /s /c \"{command}\"";
|
||||
}
|
||||
else
|
||||
{
|
||||
psi.FileName = "/bin/sh";
|
||||
psi.ArgumentList.Add("-c");
|
||||
psi.ArgumentList.Add(command);
|
||||
}
|
||||
return psi;
|
||||
}
|
||||
|
||||
private void Kill(Process process, ScriptRequest request)
|
||||
{
|
||||
try
|
||||
{
|
||||
process.Kill(entireProcessTree: true);
|
||||
process.WaitForExit(PipeGrace);
|
||||
}
|
||||
catch (Exception ex) when (ex is InvalidOperationException or Win32Exception or NotSupportedException)
|
||||
{
|
||||
logger.LogDebug(ex, "Job {JobId}: killing the script failed (it may have exited)", request.JobId);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Waits for both output pipes to close, at most <see cref="PipeGrace"/> (a surviving grandchild may hold them).</summary>
|
||||
private async Task DrainAsync(Task pumps, ScriptRequest request)
|
||||
{
|
||||
try
|
||||
{
|
||||
await pumps.WaitAsync(PipeGrace, CancellationToken.None).ConfigureAwait(false);
|
||||
}
|
||||
catch (TimeoutException)
|
||||
{
|
||||
logger.LogWarning("Job {JobId}: the script's output pipes stayed open after it ended", request.JobId);
|
||||
}
|
||||
}
|
||||
|
||||
private static async Task PumpAsync(StreamReader reader, OutputCapture capture)
|
||||
{
|
||||
var buffer = new char[4096];
|
||||
try
|
||||
{
|
||||
int read;
|
||||
while ((read = await reader.ReadAsync(buffer.AsMemory(), CancellationToken.None).ConfigureAwait(false)) > 0)
|
||||
capture.Append(buffer.AsSpan(0, read));
|
||||
}
|
||||
catch (Exception ex) when (ex is IOException or ObjectDisposedException)
|
||||
{
|
||||
// The pipe broke or the process was disposed: keep what was read.
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>The last <see cref="StderrTailLines"/> lines.</summary>
|
||||
internal static string Tail(string text)
|
||||
{
|
||||
var lines = text.Replace("\r\n", "\n", StringComparison.Ordinal).TrimEnd('\n').Split('\n');
|
||||
return string.Join('\n', lines.TakeLast(StderrTailLines));
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Keeps the first <see cref="HeadChars"/> and the last <see cref="TailChars"/> characters of a stream.</summary>
|
||||
internal sealed class OutputCapture
|
||||
{
|
||||
public const int HeadChars = 48 * 1024;
|
||||
public const int TailChars = 16 * 1024;
|
||||
private readonly StringBuilder _head = new();
|
||||
private readonly char[] _tail = new char[TailChars];
|
||||
private readonly Lock _lock = new();
|
||||
private int _tailStart;
|
||||
private int _tailLength;
|
||||
private long _dropped;
|
||||
|
||||
public void Append(string text) => Append(text.AsSpan());
|
||||
|
||||
public void Append(ReadOnlySpan<char> text)
|
||||
{
|
||||
lock (_lock)
|
||||
{
|
||||
var toHead = Math.Min(text.Length, HeadChars - _head.Length);
|
||||
if (toHead > 0) _head.Append(text[..toHead]);
|
||||
foreach (var c in text[Math.Max(toHead, 0)..])
|
||||
{
|
||||
if (_tailLength == TailChars)
|
||||
{
|
||||
_tail[_tailStart] = c;
|
||||
_tailStart = (_tailStart + 1) % TailChars;
|
||||
_dropped++;
|
||||
}
|
||||
else
|
||||
{
|
||||
_tail[(_tailStart + _tailLength) % TailChars] = c;
|
||||
_tailLength++;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
public override string ToString()
|
||||
{
|
||||
lock (_lock)
|
||||
{
|
||||
var sb = new StringBuilder(_head.Length + _tailLength + 64).Append(_head);
|
||||
if (_dropped > 0) sb.Append(CultureInfo.InvariantCulture, $"\n[… {_dropped} characters omitted …]\n");
|
||||
for (var i = 0; i < _tailLength; i++) sb.Append(_tail[(_tailStart + i) % TailChars]);
|
||||
return sb.ToString();
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user