Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
104 changes: 104 additions & 0 deletions src/kapacitor/Commands/WatchCommand.cs
Original file line number Diff line number Diff line change
@@ -1,5 +1,7 @@
using System.Runtime.InteropServices;
using System.Text;
using System.Text.Json;
using System.Text.Json.Nodes;
using System.Text.RegularExpressions;
using kapacitor.Auth;
using Microsoft.AspNetCore.SignalR.Client;
Expand Down Expand Up @@ -36,6 +38,17 @@ public static async Task<int> RunWatch(
};
PosixSignalRegistration.Create(PosixSignal.SIGTERM, _ => cts.Cancel());

// Tracks whether shutdown was triggered by the parent coding-agent process
// dying without firing session-end. When true (1), the watcher takes over the
// server's session-end POST (which the parent normally fires) so the read
// model doesn't keep the session "active" forever.
//
// Written by the parent-PID monitor task, read on the main thread —
// use Interlocked/Volatile so the C# memory model formally guarantees the
// write is observed even though awaits already act as memory barriers in
// practice.
var parentExited = 0;

// Watch the spawning claude process. If it dies without firing session-end
// (crash, force-kill, IDE-detach), self-terminate within ~5s instead of orphaning.
if (parentPid is { } ppid && ProcessHelpers.IsProcessAlive(ppid)) {
Expand All @@ -50,6 +63,7 @@ public static async Task<int> RunWatch(

if (!ProcessHelpers.IsProcessAlive(ppid)) {
Log($"Parent pid {ppid} exited; shutting down watcher");
Interlocked.Exchange(ref parentExited, 1);
cts.Cancel();
return;
}
Expand Down Expand Up @@ -232,11 +246,101 @@ public static async Task<int> RunWatch(
Log($"Done. {state.LinesProcessed} total lines processed.");

await hubConnection.DisposeAsync();

// Parent coding-agent died without firing session-end. Take over the role:
// POST /hooks/session-end so the server writes SessionEnded (otherwise the
// session stays "active" forever in the read model). Mirrors what the daemon
// already does for hosted agents via EndAgentSessionAsync. Skipped for:
// - agent watchers (agentId != null) — their lifecycle is handled by the
// parent session's own session-end hook plus the daemon path.
// - below-threshold sessions — no transcript lines were ever sent, so the
// server may not have a meaningful session to end.
// Runs after SignalR dispose so the server's StopAndDrainAsync skips the
// 10s drain wait (no live watcher connection to signal).
if (Volatile.Read(ref parentExited) == 1 && agentId is null && state.ThresholdReached) {
await PostSessionEndOnParentExitAsync(baseUrl, sessionId, transcriptPath, cwd, vendor, state.Repository);
}

await logWriter.DisposeAsync();

return 0;
}

/// <summary>
/// Whitelist of vendor values accepted by the server's session-end route.
/// Used to reject unexpected --vendor input before interpolating into the URL
/// path (defence-in-depth against path traversal even though the CLI runs locally).
/// </summary>
static readonly HashSet<string> KnownVendors = new(StringComparer.Ordinal) { "claude", "codex" };

/// <summary>
/// Total time budget for the parent-exit session-end POST. Covers /auth/config
/// discovery + retrying POST. Short by design — this runs on the watcher's
/// shutdown path; a stalled server must not block process termination, the whole
/// point of the parent-PID watchdog is to self-terminate within ~5s.
/// </summary>
static readonly TimeSpan ParentExitPostBudget = TimeSpan.FromSeconds(10);

internal static async Task PostSessionEndOnParentExitAsync(
string baseUrl,
string sessionId,
string transcriptPath,
string? cwd,
string vendor,
RepositoryPayload? repository
) {
if (!KnownVendors.Contains(vendor)) {
Log($"Parent-exit session-end skipped: unknown vendor '{vendor}'");
return;
}

using var budgetCts = new CancellationTokenSource(ParentExitPostBudget);

try {
var endHook = new JsonObject {
["session_id"] = sessionId,
["transcript_path"] = transcriptPath,
["cwd"] = cwd ?? "",
["hook_event_name"] = "session_end",
["reason"] = "parent_exited",
};

if (repository is not null) {
endHook["repository"] = JsonNode.Parse(
JsonSerializer.Serialize(repository, KapacitorJsonContext.Default.RepositoryPayload)
);
}

using var httpClient = await HttpClientExtensions.CreateAuthenticatedClientAsync(baseUrl, budgetCts.Token);
using var content = new StringContent(endHook.ToJsonString(), Encoding.UTF8, "application/json");

var url = $"{baseUrl}/hooks/session-end/{vendor}";
using var response = await httpClient.PostWithRetryAsync(url, content, timeout: ParentExitPostBudget, ct: budgetCts.Token);

if (!response.IsSuccessStatusCode) {
Log($"Parent-exit session-end POST returned HTTP {(int)response.StatusCode}");
return;
}

Log("Parent-exit session-end POST succeeded");

try {
var body = await response.Content.ReadAsStringAsync(budgetCts.Token);
var node = JsonNode.Parse(body);

if (node?["generate_whats_done"]?.GetValue<bool>() == true) {
WatcherManager.SpawnWhatsDoneGenerator(baseUrl, sessionId, vendor);
}
} catch (Exception ex) {
Log($"Parent-exit session-end response parse failed: {ex.Message}");
}
} catch (OperationCanceledException) {
Log($"Parent-exit session-end POST timed out after {ParentExitPostBudget.TotalSeconds:F0}s");
} catch (Exception ex) {
Log($"Parent-exit session-end POST failed: {ex.Message}");
}
}

static readonly Regex CommandNameRegex = CommandNameRx();
static bool parseErrorLogged;

Expand Down
137 changes: 137 additions & 0 deletions test/kapacitor.Tests.Integration/WatcherParentExitPostTests.cs
Original file line number Diff line number Diff line change
@@ -0,0 +1,137 @@
using System.Text.Json;
using kapacitor.Commands;
using WireMock.RequestBuilders;
using WireMock.ResponseBuilders;
using WireMock.Server;

namespace kapacitor.Tests.Integration;

/// <summary>
/// Validates the watcher's parent-exit session-end POST: when the watcher detects
/// its parent coding-agent process has died without firing session-end, it takes
/// over the role and POSTs to <c>/hooks/session-end/{vendor}</c> with
/// <c>reason: parent_exited</c>. The server's idempotent HandleSessionEnd writes
/// SessionEnded and tells us whether to spawn the what's-done generator.
/// </summary>
public class WatcherParentExitPostTests : IDisposable {
readonly WireMockServer _server = WireMockServer.Start();

public void Dispose() => _server.Stop();

[Test]
public async Task ParentExit_PostsSessionEnd_WithExpectedPayload() {
// Auth discovery: "None" means no Bearer token needed for subsequent calls.
_server.Given(Request.Create().WithPath("/auth/config").UsingGet())
.RespondWith(Response.Create().WithStatusCode(200).WithBody("""{"provider":"None"}"""));

_server.Given(Request.Create().WithPath("/hooks/session-end/claude").UsingPost())
.RespondWith(
Response.Create()
.WithStatusCode(200)
.WithHeader("Content-Type", "application/json")
.WithBody("""{"generate_whats_done":false}""")
);

var sessionId = $"test-{Guid.NewGuid():N}";

await WatchCommand.PostSessionEndOnParentExitAsync(
baseUrl: _server.Url!,
sessionId: sessionId,
transcriptPath: "/tmp/fake.jsonl",
cwd: "/repo",
vendor: "claude",
repository: null
);

var requests = _server.FindLogEntries(Request.Create().WithPath("/hooks/session-end/claude").UsingPost());
await Assert.That(requests.Count).IsEqualTo(1);

var body = JsonDocument.Parse(requests[0].RequestMessage.Body!).RootElement;
await Assert.That(body.GetProperty("session_id").GetString()).IsEqualTo(sessionId);
await Assert.That(body.GetProperty("transcript_path").GetString()).IsEqualTo("/tmp/fake.jsonl");
await Assert.That(body.GetProperty("cwd").GetString()).IsEqualTo("/repo");
await Assert.That(body.GetProperty("hook_event_name").GetString()).IsEqualTo("session_end");
await Assert.That(body.GetProperty("reason").GetString()).IsEqualTo("parent_exited");
}

[Test]
public async Task ParentExit_IncludesRepository_WhenProvided() {
_server.Given(Request.Create().WithPath("/auth/config").UsingGet())
.RespondWith(Response.Create().WithStatusCode(200).WithBody("""{"provider":"None"}"""));

_server.Given(Request.Create().WithPath("/hooks/session-end/claude").UsingPost())
.RespondWith(Response.Create().WithStatusCode(200).WithBody("""{"generate_whats_done":false}"""));

var sessionId = $"test-{Guid.NewGuid():N}";

await WatchCommand.PostSessionEndOnParentExitAsync(
baseUrl: _server.Url!,
sessionId: sessionId,
transcriptPath: "/tmp/fake.jsonl",
cwd: "/repo",
vendor: "claude",
repository: new RepositoryPayload {
Owner = "kurrent-io",
RepoName = "kapacitor",
Branch = "main",
}
);

var requests = _server.FindLogEntries(Request.Create().WithPath("/hooks/session-end/claude").UsingPost());
await Assert.That(requests.Count).IsEqualTo(1);

var body = JsonDocument.Parse(requests[0].RequestMessage.Body!).RootElement;
var repo = body.GetProperty("repository");
await Assert.That(repo.GetProperty("owner").GetString()).IsEqualTo("kurrent-io");
await Assert.That(repo.GetProperty("repo_name").GetString()).IsEqualTo("kapacitor");
await Assert.That(repo.GetProperty("branch").GetString()).IsEqualTo("main");
}

[Test]
public async Task ParentExit_SkipsPost_ForUnknownVendor() {
// Vendor is interpolated into the URL path. Validate against a known whitelist
// to defend against malformed --vendor values producing path traversal or
// hitting unintended endpoints.
_server.Given(Request.Create().WithPath("/auth/config").UsingGet())
.RespondWith(Response.Create().WithStatusCode(200).WithBody("""{"provider":"None"}"""));

// Catch-all for any session-end variant so we can assert nothing was posted.
_server.Given(Request.Create().WithPath("/hooks/session-end/*").UsingPost())
.RespondWith(Response.Create().WithStatusCode(200).WithBody("""{"generate_whats_done":false}"""));

await WatchCommand.PostSessionEndOnParentExitAsync(
baseUrl: _server.Url!,
sessionId: $"test-{Guid.NewGuid():N}",
transcriptPath: "/tmp/fake.jsonl",
cwd: "/repo",
vendor: "../admin",
repository: null
);

var hits = _server.FindLogEntries(Request.Create().WithPath("/hooks/*").UsingPost());
await Assert.That(hits.Count).IsEqualTo(0);
}

[Test]
public async Task ParentExit_PostsToVendorSpecificRoute_ForCodex() {
_server.Given(Request.Create().WithPath("/auth/config").UsingGet())
.RespondWith(Response.Create().WithStatusCode(200).WithBody("""{"provider":"None"}"""));

_server.Given(Request.Create().WithPath("/hooks/session-end/codex").UsingPost())
.RespondWith(Response.Create().WithStatusCode(200).WithBody("""{"generate_whats_done":false}"""));

await WatchCommand.PostSessionEndOnParentExitAsync(
baseUrl: _server.Url!,
sessionId: $"test-{Guid.NewGuid():N}",
transcriptPath: "/tmp/fake.jsonl",
cwd: "/repo",
vendor: "codex",
repository: null
);

var codexHits = _server.FindLogEntries(Request.Create().WithPath("/hooks/session-end/codex").UsingPost());
var claudeHits = _server.FindLogEntries(Request.Create().WithPath("/hooks/session-end/claude").UsingPost());
await Assert.That(codexHits.Count).IsEqualTo(1);
await Assert.That(claudeHits.Count).IsEqualTo(0);
}
}
Loading