Playa/extensions/Volcanic.AgenticCore.Extensions.Scheduler/ScriptRunner.cs

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
                                       or ArgumentException or NotSupportedException) // e.g. a workspace path that can't be created
        {
            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 &lt;command&gt;</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 "&lt;command&gt;"</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();
        }
    }
}