using System.Globalization; using System.Text; using System.Text.Json; using ClawdDotNet.Core.Audit; using ClawdDotNet.Core.Storage; using ClawdDotNet.Core.Tasks; using Microsoft.Data.Sqlite; using Microsoft.Extensions.Logging; namespace ClawdDotNet.Core.Automation; /// What the replaced tasks cost over the measured days — taken from the receipts, not estimated. public sealed record MeasuredSavings(int Days, int Runs, decimal CostUsd, long Tokens) { public decimal CostPerMonthUsd => Days == 0 ? 0 : Math.Round(CostUsd * 30m / Days, 2); public int RunsPerMonth => Days == 0 ? 0 : (int)Math.Round(Runs * 30.0 / Days); } /// #726: An agent's proposal, as the human decides it. public sealed record AutomationProposal( long Id, string AgentId, string Purpose, string Trigger, IReadOnlyList TaskIds, IReadOnlyList Services, string Risks, string TestGuide, string WorkflowJson, string? Justification, string Summary, IReadOnlyList Nodes, IReadOnlyList Warnings, MeasuredSavings Savings, string Status, string? Reason, DateTime CreatedAt, DateTime? DecidedAt, string? TaskId, long? Answers); /// What an agent hands in. public sealed record AutomationRequest( string AgentId, string Purpose, string Trigger, IReadOnlyList TaskIds, IReadOnlyList Services, string Risks, string TestGuide, string WorkflowJson, string? Justification, long? Answers, string? Argument); public sealed record AutomationOutcome(bool Ok, string Message, long? Id = null); public interface IAutomationProposals { Task> FindAsync(string subject, CancellationToken ct); Task ProposeAsync(AutomationRequest request, CancellationToken ct); } /// /// #726: Agents propose n8n workflows with reasons; a human decides and imports by hand. /// Agents never create workflows themselves — a workflow is a program that runs outside every /// check here, with all of n8n's credentials. Importing by hand is the point: it makes the /// human look at it instead of clicking "looks fine". /// /// Measured, not guessed. The saving is what the replaced tasks' runs cost over the /// last days. No flood: at most /// open proposals per agent, and a proposal for the same purpose as an open one is refused. A /// rejected proposal keeps its reason — a new one for the same purpose must answer it. /// /// Looked for, not only waited for. spots recurring /// tasks whose runs keep doing the same thing and puts them to their agent. /// public sealed class AutomationProposals( SqliteStorage storage, IAuditRepository audit, ITaskRepository tasks, TaskboardService? board, ILogger logger, TimeProvider? clock = null) : IAutomationProposals, IAsyncDisposable { public const int MaxOpenPerAgent = 3; public const int MeasureDays = 28; /// Runs in days from which a recurring task is looked at. public const int CandidateMinRuns = 8; public const int CandidateDays = 14; /// Share of runs with the same sequence of calls that makes a task a candidate. public const double CandidateSameShare = 0.75; private static readonly JsonSerializerOptions Json = new(); private readonly TimeProvider _clock = clock ?? TimeProvider.System; private readonly CancellationTokenSource _cts = new(); private Task? _loop; private DateTime Now => _clock.GetUtcNow().UtcDateTime; public async Task> FindAsync(string subject, CancellationToken ct) => (await ListAsync(null, 500, ct)) .Where(p => TaskSimilarity.Compare(p.Purpose, subject) != TaskTitleMatch.None || p.Purpose.Contains(subject.Trim(), StringComparison.OrdinalIgnoreCase)) .Take(10) .ToList(); public async Task ProposeAsync(AutomationRequest r, CancellationToken ct) { if (string.IsNullOrWhiteSpace(r.Purpose) || string.IsNullOrWhiteSpace(r.Trigger)) return new(false, "'purpose' and 'trigger' are required — what it is for, and what starts it."); if (r.TaskIds.Count == 0) return new(false, "'tasks' is required — the tasks it would replace. The saving is measured from their runs."); if (string.IsNullOrWhiteSpace(r.Risks)) return new(false, "'risks' is required — what can go wrong, and what the workflow sends outside."); if (string.IsNullOrWhiteSpace(r.TestGuide)) return new(false, "'test_guide' is required — how the human checks it after importing."); foreach (var id in r.TaskIds) if (await tasks.GetAsync(id, ct) is null) return new(false, $"There is no task '{id}'."); var all = await ListAsync(null, 1000, ct); var mine = all.Where(p => p.Status == "proposed" && string.Equals(p.AgentId, r.AgentId, StringComparison.OrdinalIgnoreCase)).ToList(); if (mine.Count >= MaxOpenPerAgent) return new(false, $"You have {mine.Count} proposals waiting for a human ({string.Join(", ", mine.Select(m => "#" + m.Id))}). Wait for a decision first."); if (all.FirstOrDefault(p => p.Status is "proposed" or "accepted" && Same(p, r)) is { } open) return new(false, $"Proposal #{open.Id} ({open.Status}) is about the same: \"{open.Purpose}\". Do not propose it again."); if (all.FirstOrDefault(p => p.Status == "rejected" && Same(p, r)) is { } rejected && (r.Answers != rejected.Id || string.IsNullOrWhiteSpace(r.Argument))) return new(false, $"Proposal #{rejected.Id} for the same purpose was rejected: \"{rejected.Reason}\". " + $"A new one must answer that: give 'answers' {rejected.Id} and an 'argument' that is not already there."); var review = WorkflowReview.Review(r.WorkflowJson, r.Justification); if (!review.Ok) return new(false, "The workflow does not pass the check:\n- " + string.Join("\n- ", review.Problems)); var savings = await MeasureAsync(r.TaskIds, ct); var id2 = await storage.WriteAsync(async conn => { using var cmd = conn.CreateCommand(); cmd.CommandText = """ INSERT INTO AutomationProposals (AgentId, Purpose, Trigger, TaskIds, Services, Risks, TestGuide, WorkflowJson, Justification, Summary, Nodes, Warnings, Savings, Status, CreatedAt, Answers, Argument) VALUES (@agent, @purpose, @trigger, @tasks, @services, @risks, @guide, @workflow, @justification, @summary, @nodes, @warnings, @savings, 'proposed', @at, @answers, @argument) RETURNING Id """; cmd.Parameters.AddWithValue("@agent", r.AgentId); cmd.Parameters.AddWithValue("@purpose", r.Purpose.Trim()); cmd.Parameters.AddWithValue("@trigger", r.Trigger.Trim()); cmd.Parameters.AddWithValue("@tasks", JsonSerializer.Serialize(r.TaskIds)); cmd.Parameters.AddWithValue("@services", JsonSerializer.Serialize(r.Services)); cmd.Parameters.AddWithValue("@risks", r.Risks.Trim()); cmd.Parameters.AddWithValue("@guide", r.TestGuide.Trim()); cmd.Parameters.AddWithValue("@workflow", r.WorkflowJson); cmd.Parameters.AddWithValue("@justification", (object?)r.Justification?.Trim() ?? DBNull.Value); cmd.Parameters.AddWithValue("@summary", review.Summary); cmd.Parameters.AddWithValue("@nodes", JsonSerializer.Serialize(review.Nodes, Json)); cmd.Parameters.AddWithValue("@warnings", JsonSerializer.Serialize(review.Warnings, Json)); cmd.Parameters.AddWithValue("@savings", JsonSerializer.Serialize(savings, Json)); cmd.Parameters.AddWithValue("@at", Now.ToString("O")); cmd.Parameters.AddWithValue("@answers", (object?)r.Answers ?? DBNull.Value); cmd.Parameters.AddWithValue("@argument", (object?)r.Argument?.Trim() ?? DBNull.Value); return Convert.ToInt64(await cmd.ExecuteScalarAsync(ct), CultureInfo.InvariantCulture); }, ct); if (board is not null) { var task = await board.CreateAsync(new TaskItem { Title = $"Automation proposal #{id2}: {r.Purpose.Trim()}", Body = $"{r.AgentId} proposes an n8n workflow. Decide it in the approvals view; if you accept, import the JSON " + $"into n8n by hand, check it with the test guide, and release it as an MCP tool.\n\n" + $"Measured over {savings.Days} days: {savings.Runs} run(s), {savings.CostUsd.ToString("0.00", CultureInfo.InvariantCulture)} USD " + $"(about {savings.CostPerMonthUsd.ToString("0.00", CultureInfo.InvariantCulture)} USD a month).\n" + $"What it does: {review.Summary}", Assignee = TaskAssignee.Human, Priority = 2 }, ct); await storage.WriteAsync(async conn => { using var cmd = conn.CreateCommand(); cmd.CommandText = "UPDATE AutomationProposals SET TaskId = @task WHERE Id = @id"; cmd.Parameters.AddWithValue("@task", task.Id); cmd.Parameters.AddWithValue("@id", id2); return await cmd.ExecuteNonQueryAsync(ct); }, ct); } logger.LogInformation("Automation proposal #{Id} by {Agent}: {Purpose}", id2, r.AgentId, r.Purpose); return new(true, $"Proposal #{id2} is with a human. Measured: {savings.RunsPerMonth} run(s) and about " + $"{savings.CostPerMonthUsd.ToString("0.00", CultureInfo.InvariantCulture)} USD a month. " + (review.Warnings.Count > 0 ? "Noted for the human: " + string.Join(" ", review.Warnings) : ""), id2); } /// The human's decision. Accepting records it; importing stays with the human, in n8n. public async Task DecideAsync(long id, bool accept, string reason, CancellationToken ct) { if ((await ListAsync("Id = @p", 1, ct, id.ToString(CultureInfo.InvariantCulture))).FirstOrDefault() is not { } proposal) return new(false, $"There is no proposal #{id}."); if (proposal.Status != "proposed") return new(false, $"Proposal #{id} is decided already ({proposal.Status})."); if (!accept && string.IsNullOrWhiteSpace(reason)) return new(false, "A rejection needs its reason — it stays with the proposal, so the same one does not come back."); await storage.WriteAsync(async conn => { using var cmd = conn.CreateCommand(); cmd.CommandText = "UPDATE AutomationProposals SET Status = @s, Reason = @r, DecidedAt = @at WHERE Id = @id AND Status = 'proposed'"; cmd.Parameters.AddWithValue("@s", accept ? "accepted" : "rejected"); cmd.Parameters.AddWithValue("@r", (reason ?? "").Trim()); cmd.Parameters.AddWithValue("@at", Now.ToString("O")); cmd.Parameters.AddWithValue("@id", id); return await cmd.ExecuteNonQueryAsync(ct); }, ct); if (proposal.TaskId is { } taskId && board is not null && await board.GetAsync(taskId, ct) is { Status: not (TaskItemStatus.Done or TaskItemStatus.Canceled) }) { await board.AddCommentAsync(taskId, "human", accept ? "Accepted." + (string.IsNullOrWhiteSpace(reason) ? "" : " " + reason.Trim()) : "Rejected: " + reason.Trim(), ct); await board.UpdateAsync(taskId, t => t with { Status = accept ? TaskItemStatus.Done : TaskItemStatus.Canceled }, ct); } return new(true, accept ? $"Proposal #{id} accepted. Import the workflow into n8n by hand, test it, release it as an MCP tool — then switch the replaced tasks off." : $"Proposal #{id} rejected; the reason stays with it."); } public Task> ListAsync(string? status, int limit, CancellationToken ct) => ListAsync(status is null ? null : "Status = @p", limit, ct, status); // ─── measuring ─── public async Task MeasureAsync(IReadOnlyList taskIds, CancellationToken ct) { var since = Now.AddDays(-MeasureDays); var runs = 0; decimal cost = 0; long tokens = 0; foreach (var id in taskIds.Distinct(StringComparer.OrdinalIgnoreCase)) foreach (var receipt in await audit.ListReceiptsForTaskAsync(id, ct)) if (receipt.OccurredAt >= since) { runs++; cost += receipt.CostIsKnown ? receipt.CostUsd : 0; tokens += receipt.PromptTokens + receipt.CompletionTokens; } return new MeasuredSavings(MeasureDays, runs, cost, tokens); } // ─── candidates ─── /// /// Recurring tasks whose runs keep doing the same thing — the same sequence of calls in most /// of them — are put to their agent once a month: maybe a workflow does it without a model. /// Returns the tasks noted in this pass. /// public async Task> FindCandidatesAsync(CancellationToken ct) { if (board is null) return []; var since = Now.AddDays(-CandidateDays); var proposals = await ListAsync(null, 1000, ct); var noted = new List(); var perAgent = new Dictionary(StringComparer.OrdinalIgnoreCase); foreach (var task in await tasks.ListAsync(new TaskQuery(), ct)) { if (task.When is not { Kind: TaskWhenKind.Every or TaskWhenKind.Cron } || task.Status is TaskItemStatus.Canceled or TaskItemStatus.Archived) continue; if (TaskAssignee.AgentId(task.Assignee) is not { Length: > 0 } agent) continue; if (proposals.Any(p => p.Status is "proposed" or "accepted" && p.TaskIds.Contains(task.Id, StringComparer.OrdinalIgnoreCase))) continue; if (await NotedSinceAsync(task.Id, Now.AddDays(-30), ct) || perAgent.GetValueOrDefault(agent) >= 2) continue; var receipts = (await audit.ListReceiptsForTaskAsync(task.Id, ct)).Where(r => r.OccurredAt >= since).ToList(); if (receipts.Count < CandidateMinRuns) continue; var signatures = new List(); foreach (var receipt in receipts) signatures.Add(Signature(await audit.ListForRunAsync(receipt.RunId, ct))); var common = signatures.GroupBy(s => s).OrderByDescending(g => g.Count()).First(); if (common.Key.Length == 0 || (double)common.Count() / signatures.Count < CandidateSameShare) continue; var cost = receipts.Where(r => r.CostIsKnown).Sum(r => r.CostUsd); await board.CreateAsync(new TaskItem { Title = $"Automation candidate: {task.Title}", Body = $"Task {task.Id} ran {receipts.Count} times in {CandidateDays} days for {cost.ToString("0.00", CultureInfo.InvariantCulture)} USD, " + $"and {common.Count()} of the runs made the same calls: {common.Key}.\n\n" + "Consider whether an n8n workflow could do it without a model. If yes: Escalation automation_find (is there a proposal already?), " + "then automation_propose with the workflow JSON, the tasks it replaces, the risks and a test guide. " + "If no: close this task and say why — judgement, changing inputs, or decisions a workflow cannot make.", Assignee = "@" + agent, Priority = 2 }, ct); await NoteAsync(task.Id, ct); perAgent[agent] = perAgent.GetValueOrDefault(agent) + 1; noted.Add(task.Id); } if (noted.Count > 0) logger.LogInformation("Automation candidates put to their agents: {Tasks}", string.Join(", ", noted)); return noted; } /// The calls of a run in order, repeats folded — "FileRW.read > DirectAPI.get > FileRW.write". internal static string Signature(IReadOnlyList entries) { var steps = new List(); foreach (var e in entries.Where(e => e.Status != AuditStatus.Started && e.Tool is not ("Taskboard" or "Escalation"))) { var step = Inspection.RunInspector.OperationOf(e.Arguments) is { } op ? $"{e.Tool}.{op}" : e.Tool; if (steps.Count == 0 || steps[^1] != step) steps.Add(step); } return string.Join(" > ", steps); } private static bool Same(AutomationProposal existing, AutomationRequest request) => TaskSimilarity.Compare(existing.Purpose, request.Purpose) == TaskTitleMatch.Duplicate || (existing.TaskIds.Count > 0 && existing.TaskIds.OrderBy(t => t).SequenceEqual(request.TaskIds.OrderBy(t => t), StringComparer.OrdinalIgnoreCase)); // ─── storage ─── private async Task NotedSinceAsync(string taskId, DateTime since, CancellationToken ct) { await using var conn = await storage.OpenConnectionAsync(ct); using var cmd = conn.CreateCommand(); cmd.CommandText = "SELECT COUNT(*) FROM AutomationCandidates WHERE TaskId = @t AND NotedAt >= @since"; cmd.Parameters.AddWithValue("@t", taskId); cmd.Parameters.AddWithValue("@since", since.ToString("O")); return Convert.ToInt32(await cmd.ExecuteScalarAsync(ct), CultureInfo.InvariantCulture) > 0; } private Task NoteAsync(string taskId, CancellationToken ct) => storage.WriteAsync(async conn => { using var cmd = conn.CreateCommand(); cmd.CommandText = "INSERT INTO AutomationCandidates (TaskId, NotedAt) VALUES (@t, @at)"; cmd.Parameters.AddWithValue("@t", taskId); cmd.Parameters.AddWithValue("@at", Now.ToString("O")); return await cmd.ExecuteNonQueryAsync(ct); }, ct); private async Task> ListAsync(string? where, int limit, CancellationToken ct, string? value = null) { await using var conn = await storage.OpenConnectionAsync(ct); using var cmd = conn.CreateCommand(); cmd.CommandText = """ SELECT Id, AgentId, Purpose, Trigger, TaskIds, Services, Risks, TestGuide, WorkflowJson, Justification, Summary, Nodes, Warnings, Savings, Status, Reason, CreatedAt, DecidedAt, TaskId, Answers FROM AutomationProposals """ + (where is null ? "" : " WHERE " + where) + " ORDER BY Id DESC LIMIT @limit"; if (value is not null) cmd.Parameters.AddWithValue("@p", value); cmd.Parameters.AddWithValue("@limit", limit); var result = new List(); await using var r = await cmd.ExecuteReaderAsync(ct); while (await r.ReadAsync(ct)) result.Add(new AutomationProposal( r.GetInt64(0), r.GetString(1), r.GetString(2), r.GetString(3), JsonSerializer.Deserialize>(r.GetString(4)) ?? [], JsonSerializer.Deserialize>(r.GetString(5)) ?? [], r.GetString(6), r.GetString(7), r.GetString(8), Text(r, 9), r.GetString(10), JsonSerializer.Deserialize>(r.GetString(11), Json) ?? [], JsonSerializer.Deserialize>(r.GetString(12)) ?? [], JsonSerializer.Deserialize(r.GetString(13), Json) ?? new MeasuredSavings(0, 0, 0, 0), r.GetString(14), Text(r, 15), Time(r, 16)!.Value, Time(r, 17), Text(r, 18), r.IsDBNull(19) ? null : r.GetInt64(19))); return result; } private static string? Text(SqliteDataReader r, int i) => r.IsDBNull(i) ? null : r.GetString(i); private static DateTime? Time(SqliteDataReader r, int i) => r.IsDBNull(i) ? null : DateTime.Parse(r.GetString(i), CultureInfo.InvariantCulture, DateTimeStyles.RoundtripKind); // ─── loop ─── public void Start() => _loop ??= RunAsync(_cts.Token); private async Task RunAsync(CancellationToken ct) { try { await Task.Delay(TimeSpan.FromMinutes(10), ct); } catch (OperationCanceledException) { return; } while (!ct.IsCancellationRequested) { try { await FindCandidatesAsync(ct); } catch (Exception ex) when (ex is not OperationCanceledException) { logger.LogWarning(ex, "Looking for automation candidates failed"); } try { await Task.Delay(TimeSpan.FromHours(24), ct); } catch (OperationCanceledException) { return; } } } public async ValueTask DisposeAsync() { await _cts.CancelAsync(); if (_loop is not null) { try { await _loop; } catch (OperationCanceledException) { } } _cts.Dispose(); } }