Skip to content
Merged
53 changes: 53 additions & 0 deletions src/Netclaw.Actors.Tests/Jobs/BackgroundJobDefinitionStoreTests.cs
Original file line number Diff line number Diff line change
Expand Up @@ -97,6 +97,59 @@ public void Current_job_with_trust_fields_roundtrips_exact_values()
Assert.Equal("C0ABC/1712000000.000001", loaded.SessionId.Value);
}

/// <summary>
/// Terminal-job cleanup: <see cref="BackgroundJobDefinitionStore.DeleteJobArtifacts"/>
/// removes BOTH the definition file and the job's output-log directory, so the
/// store cannot grow without bound after a job's retention window elapses.
/// </summary>
[Fact]
public void DeleteJobArtifacts_removes_definition_and_output_directory()
{
var store = new BackgroundJobDefinitionStore(_paths);
var jobId = new BackgroundJobId("cleanup-job-001");

store.Save(new BackgroundJobDefinition
{
Id = jobId,
Command = "dotnet test",
SessionId = new Netclaw.Actors.Protocol.SessionId("C0ABC/1712000000.000001"),
Rationale = "Run the test suite.",
Status = BackgroundJobStatus.Completed,
TimeoutSeconds = 300,
Audience = TrustAudience.Team,
Boundary = TrustBoundary.Team,
OriginChannelType = Netclaw.Actors.Channels.ChannelType.Slack
});

// Simulate a real job's output log on disk.
var outputLogPath = store.GetOutputLogPath(jobId);
File.WriteAllText(outputLogPath, "build output");

Assert.NotNull(store.Get(jobId));
Assert.True(File.Exists(outputLogPath));

var removed = store.DeleteJobArtifacts(jobId);

Assert.True(removed);
Assert.Null(store.Get(jobId));
Assert.False(File.Exists(outputLogPath));
Assert.False(Directory.Exists(Path.GetDirectoryName(outputLogPath)));
}

/// <summary>
/// Idempotent cleanup: deleting artifacts for an already-removed job reports
/// false and does not throw.
/// </summary>
[Fact]
public void DeleteJobArtifacts_missing_job_returns_false_without_throwing()
{
var store = new BackgroundJobDefinitionStore(_paths);

var removed = store.DeleteJobArtifacts(new BackgroundJobId("never-existed"));

Assert.False(removed);
}

/// <summary>
/// Byte-equality gate for issue #994 Pass 7b. Wrapping <c>BackgroundJobDefinition.Id</c>
/// in <see cref="BackgroundJobId"/> and <c>SessionId</c> in
Expand Down
104 changes: 102 additions & 2 deletions src/Netclaw.Actors.Tests/Jobs/BackgroundJobManagerActorTests.cs
Original file line number Diff line number Diff line change
Expand Up @@ -49,6 +49,21 @@ protected override async Task AfterAllAsync()

private IActorRef GetManager() => ActorRegistry.For(Sys).Get<BackgroundJobManagerActorKey>();

private BackgroundJobDefinition MakeTerminalDefinition(string jobId, BackgroundJobStatus status, long completedAtMs) => new()
{
Id = new BackgroundJobId(jobId),
Command = "echo hello",
SessionId = new SessionId("test/thread"),
Rationale = "test run",
Status = status,
StartedAtMs = completedAtMs - 60_000,
CompletedAtMs = completedAtMs,
TimeoutSeconds = 60,
Audience = TrustAudience.Personal,
Boundary = TrustBoundary.Personal,
OriginChannelType = ChannelType.Tui
};

private StartBackgroundJob MakeStartCommand(string command = "echo hello") => new()
{
Command = command,
Expand Down Expand Up @@ -309,8 +324,7 @@ await manager.Ask<BackgroundJobManagerHealthResponse>(

[Fact]
public async Task StartupReconciliation_EmitsAlert_ForLegacyJobMissingTrustFields()
{
var paths = new NetclawPaths(_dir.Path);
{ var paths = new NetclawPaths(_dir.Path);
paths.EnsureDirectoriesExist();

const string jobId = "legacy-job-alert";
Expand Down Expand Up @@ -364,4 +378,90 @@ public void Emit(OperationalAlert alert)
_alerts.Add(alert);
}
}

[Fact]
public async Task TerminalSweep_DeletesJobPastRetentionWindow()
{
var manager = GetManager();
var pastWindowMs = TimeProvider.System.GetUtcNow()
.Subtract(BackgroundJobManagerActor.TerminalJobRetentionWindow)
.Subtract(TimeSpan.FromMinutes(1))
.ToUnixTimeMilliseconds();

_store.Save(MakeTerminalDefinition("sweep-past", BackgroundJobStatus.Completed, pastWindowMs));

await manager.Ask<BackgroundJobManagerHealthResponse>(
GetBackgroundJobManagerHealth.Instance,
TimeSpan.FromSeconds(30), TestContext.Current.CancellationToken);
manager.Tell(BackgroundJobManagerActor.SweepTerminalJobs.Instance);
await manager.Ask<BackgroundJobManagerHealthResponse>(
GetBackgroundJobManagerHealth.Instance,
TimeSpan.FromSeconds(30), TestContext.Current.CancellationToken);

Assert.Null(_store.Get(new BackgroundJobId("sweep-past")));
}

[Fact]
public async Task TerminalSweep_KeepsJobWithinRetentionWindow()
{
var manager = GetManager();
var recentMs = TimeProvider.System.GetUtcNow().ToUnixTimeMilliseconds();

_store.Save(MakeTerminalDefinition("sweep-recent", BackgroundJobStatus.Completed, recentMs));

manager.Tell(BackgroundJobManagerActor.SweepTerminalJobs.Instance);
await manager.Ask<BackgroundJobManagerHealthResponse>(
GetBackgroundJobManagerHealth.Instance,
TimeSpan.FromSeconds(30), TestContext.Current.CancellationToken);

Assert.NotNull(_store.Get(new BackgroundJobId("sweep-recent")));
}

[Fact]
public async Task TerminalSweep_DoesNotTouchNonTerminalJobs()
{
var manager = GetManager();
var pastWindowMs = TimeProvider.System.GetUtcNow()
.Subtract(BackgroundJobManagerActor.TerminalJobRetentionWindow)
.Subtract(TimeSpan.FromMinutes(1))
.ToUnixTimeMilliseconds();

// A Running job with a stale CompletedAtMs must never be swept — the
// status guard is independent of the timestamp.
_store.Save(MakeTerminalDefinition("sweep-running", BackgroundJobStatus.Running, pastWindowMs));
_store.Save(MakeTerminalDefinition("sweep-pending", BackgroundJobStatus.Pending, pastWindowMs));

manager.Tell(BackgroundJobManagerActor.SweepTerminalJobs.Instance);
await manager.Ask<BackgroundJobManagerHealthResponse>(
GetBackgroundJobManagerHealth.Instance,
TimeSpan.FromSeconds(30), TestContext.Current.CancellationToken);

Assert.NotNull(_store.Get(new BackgroundJobId("sweep-running")));
Assert.NotNull(_store.Get(new BackgroundJobId("sweep-pending")));
}

[Fact]
public async Task TerminalSweep_DeletesOutputLogWithDefinition()
{
var manager = GetManager();
var pastWindowMs = TimeProvider.System.GetUtcNow()
.Subtract(BackgroundJobManagerActor.TerminalJobRetentionWindow)
.Subtract(TimeSpan.FromMinutes(1))
.ToUnixTimeMilliseconds();
var jobId = new BackgroundJobId("sweep-log");

var outputLogPath = _store.GetOutputLogPath(jobId);
File.WriteAllText(outputLogPath, "some output");

_store.Save(MakeTerminalDefinition("sweep-log", BackgroundJobStatus.Failed, pastWindowMs));

manager.Tell(BackgroundJobManagerActor.SweepTerminalJobs.Instance);
await manager.Ask<BackgroundJobManagerHealthResponse>(
GetBackgroundJobManagerHealth.Instance,
TimeSpan.FromSeconds(30), TestContext.Current.CancellationToken);

Assert.Null(_store.Get(jobId));
Assert.False(File.Exists(outputLogPath));
Assert.False(Directory.Exists(Path.GetDirectoryName(outputLogPath)));
}
}
31 changes: 30 additions & 1 deletion src/Netclaw.Actors/Jobs/BackgroundJobDefinitionStore.cs
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
// -----------------------------------------------------------------------
// -----------------------------------------------------------------------
// <copyright file="BackgroundJobDefinitionStore.cs" company="Petabridge, LLC">
// Copyright (C) 2026 - 2026 Petabridge, LLC <https://petabridge.com>
// </copyright>
Expand Down Expand Up @@ -145,6 +145,35 @@ public bool Delete(BackgroundJobId id)
}
}

/// <summary>
/// Deletes the job definition file AND its output-log directory. Used by
/// terminal-job cleanup once a job's retention window has elapsed. Returns
/// true when anything was removed.
/// </summary>
public bool DeleteJobArtifacts(BackgroundJobId id)
{
lock (_sync)
{
var removed = false;

var path = GetPath(id);
if (File.Exists(path))
{
File.Delete(path);
removed = true;
}

var outputDir = Path.Combine(_directory, Uri.EscapeDataString(id.Value));
Comment thread
github-code-quality[bot] marked this conversation as resolved.
Fixed
if (Directory.Exists(outputDir))
{
Directory.Delete(outputDir, recursive: true);
removed = true;
}

return removed;
}
}

/// <summary>
/// Returns the output log directory for a job, creating it if needed.
/// </summary>
Expand Down
102 changes: 100 additions & 2 deletions src/Netclaw.Actors/Jobs/BackgroundJobManagerActor.cs
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
// -----------------------------------------------------------------------
// -----------------------------------------------------------------------
// <copyright file="BackgroundJobManagerActor.cs" company="Petabridge, LLC">
// Copyright (C) 2026 - 2026 Petabridge, LLC <https://petabridge.com>
// </copyright>
Expand All @@ -21,11 +21,27 @@ namespace Netclaw.Actors.Jobs;
/// Infrastructure singleton that manages background job lifecycle independently
/// of any session. Follows the same pattern as <c>ReminderManagerActor</c>.
/// </summary>
public sealed class BackgroundJobManagerActor : ReceiveActor
public sealed class BackgroundJobManagerActor : ReceiveActor, IWithTimers
{
internal const int MaxConcurrentJobs = 5;
internal const int MaxOutputTailChars = 2000;

/// <summary>
/// How long a terminal job's definition and output log are retained after
/// completion before the cleanup sweep deletes them. The delivered result
/// message references the full output-log path, so the grace period must be
/// long enough for the owning session to poll <c>check_background_job</c> or
/// read the log after delivery. See <see cref="SweepTerminalJobs"/>.
/// </summary>
internal static TimeSpan TerminalJobRetentionWindow = TimeSpan.FromHours(24);
Comment thread
github-code-quality[bot] marked this conversation as resolved.
Fixed

/// <summary>
/// Cadence of the periodic terminal-job cleanup sweep. The sweep also runs
/// once during startup reconciliation, so a daemon restart immediately
/// purges anything past the retention window.
/// </summary>
internal static TimeSpan TerminalSweepInterval = TimeSpan.FromHours(1);
Comment thread
github-code-quality[bot] marked this conversation as resolved.
Fixed

// Capture ceiling for a job's output log: the execution actor drains each
// stream to this bound (head+tail) so a chatty long-running job can't buffer
// its full output in memory and OOM the daemon. The log holds a head+tail view
Expand All @@ -46,6 +62,17 @@ public sealed class BackgroundJobManagerActor : ReceiveActor
private readonly Queue<string> _deferredQueue = new();
private readonly Dictionary<string, BackgroundJobDefinition> _definitions = [];

/// <summary>
/// Timer identity for the periodic terminal-job sweep.
/// </summary>
private static readonly object TerminalSweepTimerKey = new();

/// <summary>
/// Timer scheduler injected by Akka for <see cref="IWithTimers"/>. Timers
/// are canceled automatically when the actor stops.
/// </summary>
public ITimerScheduler Timers { get; set; } = null!;

public BackgroundJobManagerActor(
BackgroundJobDefinitionStore store,
TimeProvider timeProvider,
Expand All @@ -62,6 +89,7 @@ public BackgroundJobManagerActor(
Receive<QueryBackgroundJob>(HandleQuery);
Receive<KillJobsForSession>(HandleKillJobsForSession);
Receive<GetBackgroundJobManagerHealth>(_ => HandleGetHealth());
Receive<SweepTerminalJobs>(_ => HandleSweepTerminalJobs());
}

protected override void PreStart()
Expand All @@ -72,6 +100,12 @@ protected override void PreStart()
// (macOS CI): ActorOf returns before PreStart executes, allowing external
// messages to queue ahead of the Reconcile message.
HandleReconcile();

// Periodic cleanup for terminal jobs past their retention window. The
// startup reconcile also sweeps, so a restart purges immediately; this
// timer keeps the store from growing during a long-lived daemon run.
// IWithTimers cancels the timer automatically when the actor stops.
Timers.StartPeriodicTimer(TerminalSweepTimerKey, SweepTerminalJobs.Instance, TerminalSweepInterval);
}

private async Task HandleStartAsync(StartBackgroundJob cmd)
Expand Down Expand Up @@ -322,8 +356,65 @@ private void HandleReconcile()

if (reconciled > 0)
_log.Info("Background job startup reconciliation: marked {0} orphaned job(s) as lost", reconciled);

// Startup sweep: purge terminal jobs whose retention window has elapsed,
// in the same pass as the orphan reconciliation.
HandleSweepTerminalJobs();
}

/// <summary>
/// Deletes definitions (and output logs) for terminal jobs whose
/// <c>CompletedAtMs</c> is older than <see cref="TerminalJobRetentionWindow"/>.
/// Active jobs are never touched. Terminal jobs stay queryable via
/// <c>check_background_job</c> during the window — the delivered result
/// message references the output-log path, and a slow polling session may
/// still read it — then get swept so the store cannot grow without bound.
/// </summary>
private void HandleSweepTerminalJobs()
{
var swept = 0;
var cutoffMs = _timeProvider.GetUtcNow()
.Subtract(TerminalJobRetentionWindow)
.ToUnixTimeMilliseconds();

foreach (var def in _store.List())
{
if (!IsTerminalStatus(def.Status))
continue;

// Belt-and-braces: never sweep a job this actor still considers
// active, even if the on-disk snapshot is stale.
if (_activeJobIds.Contains(def.Id.Value))
continue;

if (def.CompletedAtMs is null || def.CompletedAtMs.Value > cutoffMs)
continue;

if (_store.DeleteJobArtifacts(def.Id))
{
_definitions.Remove(def.Id.Value);
swept++;
_log.Info(
"Swept terminal background job {JobId} (status={Status}, completed_at={CompletedAtMs}) past retention window",
def.Id, def.Status, def.CompletedAtMs);
}
Comment on lines +415 to +423
}
Comment on lines +380 to +424

if (swept > 0)
_log.Info("Background job terminal sweep: deleted {0} job(s) past the retention window", swept);
}

private static bool IsTerminalStatus(BackgroundJobStatus status) => status switch
{
BackgroundJobStatus.Completed or
BackgroundJobStatus.Failed or
BackgroundJobStatus.Cancelled or
BackgroundJobStatus.TimedOut or
BackgroundJobStatus.Lost or
BackgroundJobStatus.Reaped => true,
_ => false
};

private void NotifyLostJob(BackgroundJobDefinition lost, long nowMs)
{
var outputFilePath = _store.GetOutputLogPathOnly(lost.Id);
Expand Down Expand Up @@ -497,4 +588,11 @@ private static string BuildResultContent(BackgroundJobCompleted completed, Backg
output + filePath;
}

/// <summary>
/// Internal self-message: periodic terminal-job cleanup sweep.
/// </summary>
internal sealed record SweepTerminalJobs : INoSerializationVerificationNeeded
{
public static readonly SweepTerminalJobs Instance = new();
}
}
Loading