Repository navigation
[AI-1861] Post live subagent-stop from Codex collab child watchers #526
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Changes from all commits
6643082
7ddcb38
7cfffd1
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,109 @@ | ||
| using System.Text.Json; | ||
|
|
||
| namespace Capacitor.Cli.Core; | ||
|
|
||
| /// <summary> | ||
| /// Turn-completion state a Codex collab CHILD watcher folds from its own rollout lines to | ||
| /// decide when to post the LIVE <c>/hooks/subagent-stop</c>. Codex fires no | ||
| /// per-child stop hook and <c>sub_agent_activity</c> carries no completed kind, so before | ||
| /// this the only stop was the parent's session-end teardown | ||
| /// (<c>CodexSubagentTeardown</c>) — every finished child card stayed "in progress" for the | ||
| /// parent's whole lifetime (hours, for a hosted reviewer). The deterministic per-turn | ||
| /// signal is in the child's own rollout: each turn ends with an | ||
| /// <c>event_msg.task_complete</c> line. | ||
| /// | ||
| /// A collab child is re-engageable (the parent can <c>send_message</c> again — observed | ||
| /// minutes after the first answer), but the server-side lifecycle is one-shot per | ||
| /// (session, agent): <c>AgentLifecycleDeterministicId</c> dedupes a second stop, and there | ||
| /// is no reopen. Two consequences shape this type: | ||
| /// | ||
| /// <list type="bullet"> | ||
| /// <item>The stop waits out a grace window after the last rollout activity, so a | ||
| /// same-round follow-up keeps the card honestly "in progress" instead of landing content | ||
| /// under an already-completed card. A child re-engaged AFTER the window still streams into | ||
| /// its subsession (appends are not lifecycle-gated) but its card stays completed — | ||
| /// accepted trade-off, far less wrong than spinning forever.</item> | ||
| /// <item><see cref="StopPosted"/> latches: once a stop succeeded, a later | ||
| /// <c>task_complete</c> never re-arms (the re-post would be a deduped no-op anyway). The | ||
| /// parent-end teardown remains the backstop and its duplicate stop dedupes the same | ||
| /// way.</item> | ||
| /// </list> | ||
| /// </summary> | ||
| public sealed class CodexSubagentTurnTracker { | ||
| /// <summary> | ||
| /// Default idle grace between the child's <c>task_complete</c> and the stop POST. | ||
| /// Sized to absorb the observed same-round re-engagement gaps (the parent's follow-up | ||
| /// <c>send_message</c> landed 2–6 minutes after a child's first answer in the | ||
| /// production session this was diagnosed on). | ||
| /// </summary> | ||
| public static readonly TimeSpan DefaultStopGrace = TimeSpan.FromMinutes(5); | ||
|
|
||
| /// <summary>The child rollout's last observed turn state: true once an | ||
| /// <c>event_msg.task_complete</c> landed with no later turn activity.</summary> | ||
| public bool TurnCompleted { get; private set; } | ||
|
|
||
| /// <summary>Set by the watcher once a stop POST succeeded; permanently disarms | ||
| /// <see cref="ShouldPostStop"/> (the lifecycle is one-shot — see class doc).</summary> | ||
| public bool StopPosted { get; set; } | ||
|
|
||
| /// <summary> | ||
| /// Folds one rollout line. <c>event_msg.task_complete</c> marks the turn completed; | ||
| /// real turn activity — any <c>response_item</c>, or an explicit | ||
| /// <c>event_msg.task_started</c> — re-opens it (re-engagement). Everything else | ||
| /// (<c>token_count</c> and other trailing <c>event_msg</c> noise, <c>turn_context</c>, | ||
| /// <c>world_state</c>, malformed/non-JSON lines) leaves the state unchanged, so this is | ||
| /// safe to call for every drained line. | ||
| /// </summary> | ||
| public void Observe(string line) { | ||
| try { | ||
| using var doc = JsonDocument.Parse(line); | ||
| var root = doc.RootElement; | ||
|
|
||
| switch (root.Str("type")) { | ||
| case "event_msg": | ||
| switch (root.Obj("payload")?.Str("type")) { | ||
| case "task_complete": | ||
| TurnCompleted = true; | ||
| break; | ||
|
|
||
| case "task_started": | ||
| TurnCompleted = false; | ||
| break; | ||
| } | ||
|
|
||
| break; | ||
|
|
||
| case "response_item": | ||
| TurnCompleted = false; | ||
| break; | ||
| } | ||
| } catch { | ||
| // Malformed / non-JSON lines never break the drain loop. | ||
| } | ||
| } | ||
|
|
||
| /// <summary> | ||
| /// True when the watcher should post the child's <c>subagent-stop</c> now: not yet | ||
| /// posted, last turn completed, no tool call in flight (a long-running command produces | ||
| /// no rollout line between its call and output — mirrors the idle-end guard), and the | ||
| /// rollout has been quiet for <paramref name="grace"/> since | ||
| /// <paramref name="lastActivityAt"/>. | ||
| /// </summary> | ||
| public bool ShouldPostStop(int pendingToolCallCount, DateTimeOffset lastActivityAt, DateTimeOffset now, TimeSpan grace) => | ||
| !StopPosted | ||
| && TurnCompleted | ||
| && pendingToolCallCount == 0 | ||
| && now - lastActivityAt >= grace; | ||
|
|
||
| /// <summary> | ||
| /// Resolves the stop grace from <c>KCAP_CODEX_SUBAGENT_IDLE_MINUTES</c>: non-negative | ||
| /// minutes (<c>0</c> = post immediately at <c>task_complete</c>), anything | ||
| /// unset/blank/non-numeric/negative falls back to <see cref="DefaultStopGrace"/>. Pure | ||
| /// so the parsing is unit-testable (mirrors | ||
| /// <c>WatchCommand.ResolveCodexIdleTimeout</c>). | ||
| /// </summary> | ||
| public static TimeSpan ResolveStopGrace(string? envValue) => | ||
| int.TryParse(envValue, out var minutes) && minutes >= 0 | ||
| ? TimeSpan.FromMinutes(minutes) | ||
| : DefaultStopGrace; | ||
| } | ||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -471,6 +471,19 @@ void ArmParentMonitor(int ppid) { | |
| }; | ||
| var idleExit = false; | ||
|
|
||
| // Idle grace between a Codex collab child's task_complete and its live | ||
| // subagent-stop POST (codex child watchers only — see the loop's stop check). The | ||
| // agent type is read lazily from the child rollout's own header, once. The seed | ||
| // rebuilds turn/pending state from the acknowledged prefix, which a restarted | ||
| // watcher's watermark-resumed drain would otherwise never re-deliver. | ||
| var codexSubagentStopGrace = CodexSubagentTurnTracker.ResolveStopGrace(Environment.GetEnvironmentVariable("KCAP_CODEX_SUBAGENT_IDLE_MINUTES")); | ||
| string? codexChildAgentType = null; | ||
| var codexStopNextAttemptAt = DateTimeOffset.MinValue; | ||
|
|
||
| if (vendor == "codex" && agentId is not null) { | ||
| SeedCodexSubagentTurnState(state, transcriptPath); | ||
| } | ||
|
|
||
| if (skipTitle) { | ||
| state.TitleGenerated = true; | ||
| } | ||
|
|
@@ -710,6 +723,32 @@ Task<IReadOnlyList<string>> DrainNewLinesGatedAsync(bool isFinalDrainLocal, Canc | |
| await ScanAntigravitySubagentLinks(baseUrl, sessionId, drained, state.PostedSubagentLinks, cts.Token); | ||
| } | ||
|
|
||
| // A Codex collab CHILD posts its own subagent-stop once its rollout's | ||
| // last turn completed (task_complete, no tool call in flight) and the grace | ||
| // window elapsed — without this, the parent's session-end teardown is the only | ||
| // stop, so a finished child's chat card spins for the parent's whole lifetime. | ||
| // One-shot: the server dedupes lifecycle events per (session, agent), so the | ||
| // tracker latches after the first successful POST; a failed POST retries on the | ||
| // CodexSubagentStopRetryInterval cadence (never every 1s tick), and the | ||
| // parent-end teardown remains the backstop either way. The child keeps | ||
| // streaming after the stop — appends are not lifecycle-gated, so a re-engaged | ||
| // child's content still lands in its subsession. | ||
| if (agentId is not null && vendor == "codex" | ||
| && DateTimeOffset.UtcNow >= codexStopNextAttemptAt | ||
| && state.CodexSubagentTurn.ShouldPostStop( | ||
| state.PendingCodexToolCalls.Count, state.LastActivityAt, | ||
| DateTimeOffset.UtcNow, codexSubagentStopGrace)) { | ||
| codexChildAgentType ??= ResolveCodexChildAgentType(transcriptPath); | ||
|
|
||
| if (await PostCodexSubagentStopAsync(baseUrl, sessionId, agentId, codexChildAgentType, transcriptPath, cts.Token)) { | ||
| state.CodexSubagentTurn.StopPosted = true; | ||
| Log($"Codex subagent {agentId} ({codexChildAgentType}) turn complete + idle " | ||
| + $"{codexSubagentStopGrace.TotalMinutes:F0}m; posted subagent-stop"); | ||
|
qodo-code-review[bot] marked this conversation as resolved.
Resolved
|
||
| } else { | ||
| codexStopNextAttemptAt = DateTimeOffset.UtcNow + CodexSubagentStopRetryInterval; | ||
| } | ||
| } | ||
|
|
||
| // the Cursor idle clock must be the LATER of transcript | ||
| // activity AND the hook heartbeat mtime. Keyed on the CHILD's own session id for | ||
| // a child watcher (agentId), matching how CursorHookCommand actually writes the | ||
|
|
@@ -1051,6 +1090,98 @@ static async Task<bool> PostCodexSubagentStartAsync( | |
| } | ||
| } | ||
|
|
||
| /// <summary> | ||
| /// Overall budget for one live child-stop POST attempt (auth client creation + the POST | ||
| /// itself). Must stay well inside <see cref="WatcherHeartbeat.Threshold"/> (20s): the | ||
| /// main loop touches the heartbeat once per iteration and awaits this call inline, so a | ||
| /// slow attempt on the default 30s retry budget could mark the watcher stale and churn | ||
| /// it exactly at turn completion. Retries ride later loop iterations instead, on the | ||
| /// <see cref="CodexSubagentStopRetryInterval"/> cadence. | ||
| /// </summary> | ||
| static readonly TimeSpan CodexSubagentStopPostBudget = TimeSpan.FromSeconds(5); | ||
|
|
||
| /// <summary> | ||
| /// Minimum spacing between live child-stop POST attempts, so a persistent failure | ||
| /// (outage, auth trouble) never hammers <c>/hooks/subagent-stop</c> at the loop's 1s | ||
| /// cadence. Deliberately no terminal give-up: an auth refresh can heal a 401 and the | ||
| /// parent-end teardown dedupes whatever this never manages to post. | ||
| /// </summary> | ||
| static readonly TimeSpan CodexSubagentStopRetryInterval = TimeSpan.FromSeconds(60); | ||
|
|
||
| /// <summary> | ||
| /// The live per-child stop, posted by the CHILD watcher itself once its turn | ||
| /// completed and idled — the same wire shape <see cref="CodexSubagentTeardown"/> posts at | ||
| /// parent end (which stays the backstop; the server dedupes the duplicate stop via its | ||
| /// deterministic lifecycle event ids). Bounded by | ||
| /// <see cref="CodexSubagentStopPostBudget"/>; failures are logged and reported to the | ||
| /// caller, never thrown. | ||
| /// </summary> | ||
| static async Task<bool> PostCodexSubagentStopAsync( | ||
| string baseUrl, string sessionId, string agentId, string agentType, string subFile, CancellationToken ct | ||
| ) { | ||
| try { | ||
| using var budgetCts = CancellationTokenSource.CreateLinkedTokenSource(ct); | ||
| budgetCts.CancelAfter(CodexSubagentStopPostBudget); | ||
|
|
||
| using var client = await HttpClientExtensions.CreateAuthenticatedClientAsync(baseUrl, budgetCts.Token); | ||
| var payload = CodexSubagentDiscovery.BuildStopPayload(sessionId, agentId, agentType, subFile); | ||
| using var content = new StringContent(payload.ToJsonString(), Encoding.UTF8, "application/json"); | ||
| using var resp = await client.PostWithRetryAsync($"{baseUrl}/hooks/subagent-stop", content, timeout: CodexSubagentStopPostBudget, ct: budgetCts.Token); | ||
|
|
||
| if (!resp.IsSuccessStatusCode) { | ||
| Log($"Codex subagent {agentId} stop POST returned {(int)resp.StatusCode}; " | ||
| + $"retrying in {CodexSubagentStopRetryInterval.TotalSeconds:F0}s"); | ||
| } | ||
|
|
||
| return resp.IsSuccessStatusCode; | ||
| } catch (Exception ex) { | ||
| Log($"Codex subagent {agentId} stop POST failed: {ex.Message}; " | ||
| + $"retrying in {CodexSubagentStopRetryInterval.TotalSeconds:F0}s"); | ||
|
|
||
| return false; | ||
| } | ||
|
qodo-code-review[bot] marked this conversation as resolved.
Resolved
|
||
| } | ||
|
|
||
| /// <summary> | ||
| /// Agent type for a child watcher's own stop payload, read from the child rollout's | ||
| /// <c>session_meta</c> header — the child watcher isn't handed the type at spawn (only the | ||
| /// parent's scan derives it), and <see cref="CodexSubagentDiscovery.ReadHeader"/> never | ||
| /// throws, so a still-unreadable header just yields the generic fallback. | ||
| /// </summary> | ||
| static string ResolveCodexChildAgentType(string transcriptPath) { | ||
| var (_, meta) = CodexSubagentDiscovery.ReadHeader(transcriptPath); | ||
|
|
||
| return CodexSubagentDiscovery.AgentTypeFrom(meta.AgentPath, meta.AgentNickname); | ||
| } | ||
|
|
||
| /// <summary> | ||
| /// Startup restart-recovery for a Codex collab CHILD watcher's live-stop state: | ||
| /// a restarted watcher resumes from the server watermark | ||
| /// (<c>WatcherConnect</c>), so the drain never re-delivers already-acknowledged lines — | ||
| /// including the <c>task_complete</c> that should drive the live stop. Without this | ||
| /// seed, a child watcher dying between that ack and the grace-delayed stop POST left | ||
| /// the card spinning until the parent-end teardown. Folds the rollout's full on-disk | ||
| /// prefix through both the turn tracker and the pending-call set; the first drain | ||
| /// re-observing the unacknowledged suffix is harmless — both folds converge on the | ||
| /// same last-state. Best-effort: an unreadable rollout just leaves fresh-start state | ||
| /// (the pre-seed behavior), with the parent-end teardown as backstop. | ||
| /// </summary> | ||
| internal static void SeedCodexSubagentTurnState(WatchState state, string transcriptPath) { | ||
| try { | ||
| using var fs = new FileStream(transcriptPath, FileMode.Open, FileAccess.Read, FileShare.ReadWrite | FileShare.Delete); | ||
| using var reader = new StreamReader(fs); | ||
|
|
||
| while (reader.ReadLine() is { } line) { | ||
| if (string.IsNullOrWhiteSpace(line)) continue; | ||
|
|
||
| UpdateCodexPendingToolCalls(state.PendingCodexToolCalls, line); | ||
| state.CodexSubagentTurn.Observe(line); | ||
| } | ||
| } catch { | ||
| // Missing/locked rollout — seed is best-effort, see doc comment. | ||
| } | ||
| } | ||
|
|
||
| /// <summary> | ||
| /// Whitelist of vendor values accepted by the server's session-end route. | ||
| /// Used to reject unexpected --vendor input before interpolating into the URL | ||
|
|
@@ -1632,6 +1763,16 @@ internal static async Task<IReadOnlyList<string>> DrainNewLines( | |
| UpdateCodexPendingToolCalls(state.PendingCodexToolCalls, line); | ||
| } | ||
|
|
||
| // A Codex collab CHILD watcher additionally folds its own rollout's turn state | ||
| // (task_complete vs renewed activity) so the polling loop can post a live | ||
| // subagent-stop once the child is done + idle. Gated, unlike the | ||
| // pending-call tracking above, because only codex child watchers ever consult it. | ||
| if (vendor == "codex" && agentId is not null) { | ||
| foreach (var line in newLines) { | ||
| state.CodexSubagentTurn.Observe(line); | ||
|
Comment on lines
+1770
to
+1772
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more.
When a child watcher restarts after the batch containing Useful? React with 👍 / 👎. |
||
| } | ||
| } | ||
|
|
||
| // Capture first user text (needed for title generation) | ||
| if (state is { TitleGenerated: false, FirstUserText: null } && agentId is null) { | ||
| // If we resumed from a later position, scan from the beginning of the file | ||
|
|
||
Uh oh!
There was an error while loading. Please reload this page.