diff --git a/docs/spec/SPEC-011-daemon-architecture.md b/docs/spec/SPEC-011-daemon-architecture.md index 2b3874244..d792879ec 100644 --- a/docs/spec/SPEC-011-daemon-architecture.md +++ b/docs/spec/SPEC-011-daemon-architecture.md @@ -210,6 +210,13 @@ The daemon writes its PID to `~/.netclaw/netclaw.pid` for lifecycle management. shutdown. The daemon handles SIGTERM by draining active sessions and stopping the actor system cleanly. +During drain, a session can stop a tool task that waits only for durable +approval prompts. The session waits for the tool task to stop before it +acknowledges drain. Its journal retains the prompts and completed sibling +results. An approval after restart resumes the original turn under its +recorded authority. Active tools, accepted buffered input, and incomplete +results keep the current bounded drain path. + `netclaw daemon status` checks the PID file and verifies the process is alive. Reports: running/stopped, PID, uptime, port, number of active sessions. diff --git a/feeds/skills/.system/files/netclaw-operations/SKILL.md b/feeds/skills/.system/files/netclaw-operations/SKILL.md index 4799e2e97..a82968ed5 100644 --- a/feeds/skills/.system/files/netclaw-operations/SKILL.md +++ b/feeds/skills/.system/files/netclaw-operations/SKILL.md @@ -3,7 +3,7 @@ name: netclaw-operations description: "REQUIRED when the user asks about scheduling, reminders, cron jobs, timers, background jobs, diagnostics, troubleshooting, MCP tools, daemon health, identity updates, or Netclaw capabilities and self-maintenance." metadata: author: netclaw - version: "2.74.7" + version: "2.74.8" --- # Netclaw Operations @@ -425,6 +425,11 @@ failed or was superseded); the session then posts a visible "approval prompt has expired" notice rather than silently dropping the click. If a user reports a stale button, ask them to re-issue the request. +During a graceful stop, the session can stop a tool task that waits only for +journaled approval prompts. The session waits for that task to stop before it +acknowledges drain. The original requester can still approve after restart. +An active tool or accepted buffered input keeps the current bounded drain path. + **Why you may not see a prompt at all.** If the user invokes a read-only verb (say `grep`) with a path argument under a tree the operator has previously trusted, the safe-verb short-circuit applies and there is no prompt. This diff --git a/openspec/changes/stop-durable-approval-waits/.openspec.yaml b/openspec/changes/stop-durable-approval-waits/.openspec.yaml new file mode 100644 index 000000000..f2cbbe6a6 --- /dev/null +++ b/openspec/changes/stop-durable-approval-waits/.openspec.yaml @@ -0,0 +1,2 @@ +schema: spec-driven +created: 2026-09-18 diff --git a/openspec/changes/stop-durable-approval-waits/design.md b/openspec/changes/stop-durable-approval-waits/design.md new file mode 100644 index 000000000..24d039c36 --- /dev/null +++ b/openspec/changes/stop-durable-approval-waits/design.md @@ -0,0 +1,62 @@ +## Context + +See [the proposal](proposal.md) for the delay. `LlmSessionActor` keeps the active tool task alive while `SessionToolExecutionPipeline` waits for an approval. A `ToolApprovalRequested` event records a durable parent-session prompt. `ToolCallRecorded` records a completed sibling result. The pipeline now has no explicit stop acknowledgment. + +The session actor owns the live task and the drain decision. The session journal owns approval and result state. The actor buffer and deferred approval response are actor-local data. The [engineering glossary](../../../docs/spec/GLOSSARY.md) defines shared terms. + +## Goals / Non-Goals + +**Goals:** + +- Stop a tool task after all unfinished calls have durable approval records. +- Acknowledge drain only after that task stops. +- Keep approved tool work safe for cold recovery under its original turn authority. + +**Non-Goals:** + +- Resume a canceled model call or accepted input automatically. +- Shorten the global stop limit before other active states have safe paths. + +## Decisions + +### The actor checks each call at the journal boundary + +The active batch tracker knows expected call IDs and applied result IDs. The approval state knows each pending call and whether its prompt is durable. The actor uses both sets after event callbacks apply. It does not trust a prompt that has only entered the actor mailbox. + +The actor can use the fast path only when each expected call has an applied result or an unresolved durable approval. At least one call must await approval. The actor excludes accepted buffered input and a deferred approval response. This rule rejects active siblings and non-durable child approvals. + +### The tool task supplies the stop signal + +The actor retains the task returned by the current tool pipeline. After eligibility, the actor cancels that task's token. A task completion signal returns to the actor mailbox. The actor then passivates and acknowledges drain. The actor does not treat token cancellation alone as proof that work stopped. + +The pipeline can report cancellation as a failed batch. The actor suppresses only the cancellation that belongs to this verified drain attempt. An unrelated failure keeps the existing failure path. A stale task signal cannot complete a newer batch. + +### The journal remains the recovery source + +The actor adds no new persisted event in this slice. The existing approval and result events restore the parked batch. A recovered approval response uses its recorded `TurnContextRecord`. An incomplete legacy context cannot grant broader authority. + +### Schematic sequence + +```text +graceful stop -> session actor marks drain requested +approval/result journal callback -> actor checks every call +if all unfinished calls await durable approval and local buffers are empty: + actor cancels the current tool task + tool task stops -> actor receives the stop signal + actor passivates -> drain ack +else: + current bounded drain path remains +``` + +The sequence omits ordinary policy checks and journal callbacks. + +## Risks / Trade-offs + +- A sibling can finish while the actor checks eligibility. The actor uses applied result events and mailbox order to keep the result before drain. +- A tool can ignore cancellation. The actor then gives no early acknowledgment. The global stop limit still applies. +- A resolved approval can race with drain. The actor excludes an unfinished resolved call and keeps the bounded path. +- An accepted buffered message has no durable admission record today. The actor excludes that state until the later input-admission slice. + +## Migration Plan + +This change needs no journal migration or configuration update. A rollback removes the fast path. Journaled approvals keep their current recovery behavior. diff --git a/openspec/changes/stop-durable-approval-waits/proposal.md b/openspec/changes/stop-durable-approval-waits/proposal.md new file mode 100644 index 000000000..39ffd4a12 --- /dev/null +++ b/openspec/changes/stop-durable-approval-waits/proposal.md @@ -0,0 +1,33 @@ +## Why + +Source PRD: [PRD-001 FR-016](../../../docs/prd/PRD-001-netclaw-mvp.md#fr-016-config-change-restart-coordination). + +A session can wait for a tool approval until the daemon reaches its 190-second stop limit. The approval already has a journal record. Netclaw can stop that wait after the tool task stops. + +## What Changes + +- Let a session stop promptly when every unfinished tool call waits on a durable approval. +- Require each completed sibling tool call to have a journaled result before the session stops. +- Require the tool task to stop before the actor acknowledges drain. +- Keep the current bounded drain path for model calls, active tools, unresolved results, accepted buffered input, and deferred approval responses. +- Preserve the original approval and its turn authority after cold recovery. + +This slice does not add automatic session wakeups, new input admission records, or a shorter global stop limit. + +## Capabilities + +### New Capabilities + +None. + +### Modified Capabilities + +- `session-resume`: A graceful daemon stop can passivate a session that waits only for durable tool approvals. + +## Impact + +The change affects `LlmSessionActor`, its tool task boundary, the session-resume contract, and actor tests. It adds no public API or configuration property. + +### Security and operational impact + +The actor cannot stop before every unfinished call has a journaled approval. It cannot replay a call with an uncertain effect. The daemon retains its current timeout for other states. diff --git a/openspec/changes/stop-durable-approval-waits/specs/session-resume/spec.md b/openspec/changes/stop-durable-approval-waits/specs/session-resume/spec.md new file mode 100644 index 000000000..a52cc9a7f --- /dev/null +++ b/openspec/changes/stop-durable-approval-waits/specs/session-resume/spec.md @@ -0,0 +1,47 @@ +## ADDED Requirements + +### Requirement: Graceful stop of durable approval waits + +During a graceful daemon stop, a session SHALL finish drain when every unfinished tool call waits on a durable approval. The session SHALL wait for the tool task to stop before it acknowledges drain. The journal SHALL retain approval requests and completed tool results for cold recovery. + +Use the [engineering glossary](../../../../../docs/spec/GLOSSARY.md) for shared terms. + +#### Scenario: One durable approval stops promptly + +- **GIVEN** a session has one unfinished tool call with a journaled approval request +- **WHEN** the daemon requests a graceful drain +- **THEN** the session stops the tool task and acknowledges drain without an approval response +- **AND** the original requester can approve the call after cold recovery +- **AND** the recovered turn keeps its original authority + +#### Scenario: Completed sibling retains its result + +- **GIVEN** one tool result has a journal record and another tool call waits on a journaled approval +- **WHEN** the daemon requests a graceful drain +- **THEN** the session acknowledges drain after the tool task stops +- **AND** recovery does not execute the completed sibling again + +#### Scenario: Active sibling prevents the fast path + +- **GIVEN** one tool call waits on a journaled approval and another tool call has no journaled result or approval +- **WHEN** the daemon requests a graceful drain +- **THEN** the session does not acknowledge drain through the approval fast path +- **AND** the current bounded drain path remains in effect + +#### Scenario: Accepted buffered input prevents the fast path + +- **GIVEN** a session has accepted user input in its actor buffer +- **WHEN** the daemon requests a graceful drain +- **THEN** the session does not acknowledge drain through the approval fast path + +#### Scenario: Non-durable or resolved approval prevents the fast path + +- **GIVEN** an unfinished call has a non-durable approval or a resolved approval without a result +- **WHEN** the daemon requests a graceful drain +- **THEN** the session does not acknowledge drain through the approval fast path + +#### Scenario: Deferred approval response prevents the fast path + +- **GIVEN** the session has a deferred approval response +- **WHEN** the daemon requests a graceful drain +- **THEN** the session does not acknowledge drain through the approval fast path diff --git a/openspec/changes/stop-durable-approval-waits/tasks.md b/openspec/changes/stop-durable-approval-waits/tasks.md new file mode 100644 index 000000000..b8e8aebdc --- /dev/null +++ b/openspec/changes/stop-durable-approval-waits/tasks.md @@ -0,0 +1,17 @@ +## 1. Actor proof + +- [x] 1.1 Add an actor test for a journaled approval during drain. Run it against the baseline and confirm that the drain ack does not arrive. +- [x] 1.2 Add the tool task stop signal and the durable approval gate. Verify that the actor test receives the ack after task stop. +- [x] 1.3 Cold-recover the session, approve the parked call, and verify one execution under the original trust context. + +## 2. Negative boundaries + +- [x] 2.1 Add a sibling tool gate and verify that an active sibling prevents an early drain ack. +- [x] 2.2 Verify that an unresolved result, non-durable approval, accepted buffer, or deferred response prevents the fast path. +- [x] 2.3 Run the focused actor tests and verify no duplicate tool action or incomplete journal result. + +## 3. Contract and gates + +- [x] 3.1 Update `SPEC-011` and the `netclaw-operations` system skill. Verify that both describe the approval-only drain rule. +- [x] 3.2 Run the behavioral eval suite for the skill change and verify its result. +- [x] 3.3 Run the actor test project, Slopwatch, header verification, OpenSpec validation, and `git diff --check`. diff --git a/openspec/specs/session-resume/spec.md b/openspec/specs/session-resume/spec.md index 491ca6133..2b2f5020f 100644 --- a/openspec/specs/session-resume/spec.md +++ b/openspec/specs/session-resume/spec.md @@ -178,6 +178,52 @@ truth for in-flight approval state. - **THEN** recovery SHALL succeed with an empty pending-interaction set - **AND** SHALL NOT fail or error on the missing field +### Requirement: Graceful stop of durable approval waits + +During a graceful daemon stop, a session SHALL finish drain when every unfinished tool call waits on a durable approval. The session SHALL wait for the tool task to stop before it acknowledges drain. The journal SHALL retain approval requests and completed tool results for cold recovery. + +Use the [engineering glossary](../../../docs/spec/GLOSSARY.md) for shared terms. + +#### Scenario: One durable approval stops promptly + +- **GIVEN** a session has one unfinished tool call with a journaled approval request +- **WHEN** the daemon requests a graceful drain +- **THEN** the session stops the tool task and acknowledges drain without an approval response +- **AND** the original requester can approve the call after cold recovery +- **AND** the recovered turn keeps its original authority + +#### Scenario: Completed sibling retains its result + +- **GIVEN** one tool result has a journal record and another tool call waits on a journaled approval +- **WHEN** the daemon requests a graceful drain +- **THEN** the session acknowledges drain after the tool task stops +- **AND** recovery does not execute the completed sibling again + +#### Scenario: Active sibling prevents the fast path + +- **GIVEN** one tool call waits on a journaled approval and another tool call has no journaled result or approval +- **WHEN** the daemon requests a graceful drain +- **THEN** the session does not acknowledge drain through the approval fast path +- **AND** the current bounded drain path remains in effect + +#### Scenario: Accepted buffered input prevents the fast path + +- **GIVEN** a session has accepted user input in its actor buffer +- **WHEN** the daemon requests a graceful drain +- **THEN** the session does not acknowledge drain through the approval fast path + +#### Scenario: Non-durable or resolved approval prevents the fast path + +- **GIVEN** an unfinished call has a non-durable approval or a resolved approval without a result +- **WHEN** the daemon requests a graceful drain +- **THEN** the session does not acknowledge drain through the approval fast path + +#### Scenario: Deferred approval response prevents the fast path + +- **GIVEN** the session has a deferred approval response +- **WHEN** the daemon requests a graceful drain +- **THEN** the session does not acknowledge drain through the approval fast path + ### Requirement: Recovered pending approvals restore turn context When a session recovers pending tool approvals from the journal, it SHALL also restore the original turn context for each pending approval. The restored context SHALL include the requester, audience, boundary, channel type, approval capability, principal classification, provenance, and adopted-context safety state needed to resume the original request faithfully. diff --git a/src/Netclaw.Actors.Tests/Sessions/ApprovalRehydrationTests.cs b/src/Netclaw.Actors.Tests/Sessions/ApprovalRehydrationTests.cs index 606ffa4a7..740f59d8c 100644 --- a/src/Netclaw.Actors.Tests/Sessions/ApprovalRehydrationTests.cs +++ b/src/Netclaw.Actors.Tests/Sessions/ApprovalRehydrationTests.cs @@ -68,6 +68,211 @@ protected override void ConfigureSessionServices(IServiceCollection services) services.AddSingleton(registry); } + [Fact] + public async Task Restart_drain_stops_durable_approval_wait_and_cold_response_resumes_original_turn() + { + const string callId = "call-shell-restart-drain"; + _toolExecutor.GatedTools.Add("shell_execute"); + _fakeChatClient.ToolCallsOnFirstCall = + [ + new FunctionCallContent(callId, "shell_execute", + new Dictionary { ["command"] = "git status" }) + ]; + + var sessionId = new SessionId("test-channel/restart-drain-pending-approval"); + var sessionManager = ActorRegistry.Get(); + var subscriber = CreateTestProbe("restart-drain-pending-sub"); + + await sessionManager.Ask(new JoinSession(subscriber) + { + SessionId = sessionId, + Filter = OutputFilter.Full + }, TimeSpan.FromSeconds(10), TestContext.Current.CancellationToken); + await subscriber.ExpectMsgAsync(cancellationToken: TestContext.Current.CancellationToken); + + await sessionManager.Ask(new SendUserMessage + { + SessionId = sessionId, + Content = "Run git status", + Source = RequesterSource("U-requester") + }, TimeSpan.FromSeconds(5), TestContext.Current.CancellationToken); + + await subscriber.ExpectMsgAsync( + TimeSpan.FromSeconds(5), cancellationToken: TestContext.Current.CancellationToken); + await subscriber.ExpectMsgAsync( + TimeSpan.FromSeconds(5), cancellationToken: TestContext.Current.CancellationToken); + Assert.Equal(0, _toolExecutor.SuccessfulExecutions); + + var escapedId = Uri.EscapeDataString(sessionId.Value); + var child = await Sys.ActorSelection($"/user/session-manager/{escapedId}") + .ResolveOne(TimeSpan.FromSeconds(5), TestContext.Current.CancellationToken); + Watch(child); + + var ack = await sessionManager.Ask( + new PrepareForDaemonRestart(sessionId, "config-reload"), + TimeSpan.FromSeconds(5), TestContext.Current.CancellationToken); + Assert.Equal(sessionId, ack.SessionId); + await ExpectTerminatedAsync(child, TimeSpan.FromSeconds(5), + cancellationToken: TestContext.Current.CancellationToken); + + var recoveredSubscriber = CreateTestProbe("restart-drain-recovered-sub"); + await sessionManager.Ask(new JoinSession(recoveredSubscriber) + { + SessionId = sessionId, + Filter = OutputFilter.Full + }, TimeSpan.FromSeconds(10), TestContext.Current.CancellationToken); + await recoveredSubscriber.ExpectMsgAsync( + cancellationToken: TestContext.Current.CancellationToken); + + sessionManager.Tell(new ToolInteractionResponse + { + SessionId = sessionId, + CallId = new Netclaw.Tools.ToolCallId(callId), + SelectedKey = new ApprovalOptionKey(ApprovalOptionKeys.ApproveOnce), + SenderId = new SenderId("U-requester") + }, ActorRefs.Nobody); + + await recoveredSubscriber.ExpectMsgAsync( + TimeSpan.FromSeconds(5), cancellationToken: TestContext.Current.CancellationToken); + await recoveredSubscriber.ExpectMsgAsync( + TimeSpan.FromSeconds(5), cancellationToken: TestContext.Current.CancellationToken); + var completed = await recoveredSubscriber.ExpectMsgAsync( + TimeSpan.FromSeconds(5), cancellationToken: TestContext.Current.CancellationToken); + Assert.Equal(TurnOutcome.Completed, completed.Outcome); + Assert.Equal(1, _toolExecutor.SuccessfulExecutions); + Assert.Equal(TrustAudience.Team, _toolExecutor.LastExecutionAudience); + } + + [Fact] + public async Task Restart_drain_waits_for_active_sibling_then_preserves_its_result() + { + const string shellCallId = "call-shell-restart-sibling"; + _toolExecutor.GatedTools.Add("shell_execute"); + _toolExecutor.BlockNextSuccessfulExecution("read_file"); + _fakeChatClient.ToolCallsOnFirstCall = + [ + new FunctionCallContent(shellCallId, "shell_execute", + new Dictionary { ["command"] = "git status" }), + new FunctionCallContent("call-read-restart-sibling", "read_file", + new Dictionary { ["path"] = "README.md" }) + ]; + + var sessionId = new SessionId("test-channel/restart-drain-active-sibling"); + var sessionManager = ActorRegistry.Get(); + var subscriber = CreateTestProbe("restart-drain-sibling-sub"); + await sessionManager.Ask(new JoinSession(subscriber) + { + SessionId = sessionId, + Filter = OutputFilter.Full + }, TimeSpan.FromSeconds(10), TestContext.Current.CancellationToken); + await subscriber.ExpectMsgAsync(cancellationToken: TestContext.Current.CancellationToken); + + await sessionManager.Ask(new SendUserMessage + { + SessionId = sessionId, + Content = "Read the file and run git status", + Source = RequesterSource("U-requester") + }, TimeSpan.FromSeconds(5), TestContext.Current.CancellationToken); + + await subscriber.ExpectMsgAsync( + TimeSpan.FromSeconds(5), cancellationToken: TestContext.Current.CancellationToken); + await subscriber.ExpectMsgAsync( + TimeSpan.FromSeconds(5), cancellationToken: TestContext.Current.CancellationToken); + await subscriber.ExpectMsgAsync( + TimeSpan.FromSeconds(5), cancellationToken: TestContext.Current.CancellationToken); + await _toolExecutor.BlockedExecutionStarted.Task.WaitAsync( + TimeSpan.FromSeconds(5), TestContext.Current.CancellationToken); + + var escapedId = Uri.EscapeDataString(sessionId.Value); + var child = await Sys.ActorSelection($"/user/session-manager/{escapedId}") + .ResolveOne(TimeSpan.FromSeconds(5), TestContext.Current.CancellationToken); + Watch(child); + var drainProbe = CreateTestProbe("restart-drain-sibling-ack"); + child.Tell(new PrepareForDaemonRestart(sessionId, "config-reload"), drainProbe.Ref); + await drainProbe.ExpectNoMsgAsync(TimeSpan.FromMilliseconds(250), + TestContext.Current.CancellationToken); + + _toolExecutor.ReleaseBlockedExecution(); + await subscriber.ExpectMsgAsync( + TimeSpan.FromSeconds(5), cancellationToken: TestContext.Current.CancellationToken); + await drainProbe.ExpectMsgAsync( + TimeSpan.FromSeconds(5), cancellationToken: TestContext.Current.CancellationToken); + await ExpectTerminatedAsync(child, TimeSpan.FromSeconds(5), + cancellationToken: TestContext.Current.CancellationToken); + + var recoveredSubscriber = CreateTestProbe("restart-drain-sibling-recovered-sub"); + await sessionManager.Ask(new JoinSession(recoveredSubscriber) + { + SessionId = sessionId, + Filter = OutputFilter.Full + }, TimeSpan.FromSeconds(10), TestContext.Current.CancellationToken); + await recoveredSubscriber.ExpectMsgAsync( + cancellationToken: TestContext.Current.CancellationToken); + + sessionManager.Tell(new ToolInteractionResponse + { + SessionId = sessionId, + CallId = new Netclaw.Tools.ToolCallId(shellCallId), + SelectedKey = new ApprovalOptionKey(ApprovalOptionKeys.ApproveOnce), + SenderId = new SenderId("U-requester") + }, ActorRefs.Nobody); + + await recoveredSubscriber.FishForMessageAsync( + _ => true, TimeSpan.FromSeconds(5), + cancellationToken: TestContext.Current.CancellationToken); + Assert.Equal(1, _toolExecutor.ExecutionsFor("read_file")); + Assert.Equal(1, _toolExecutor.ExecutionsFor("shell_execute")); + } + + [Fact] + public async Task Restart_drain_keeps_the_bounded_path_when_user_input_is_buffered() + { + _toolExecutor.GatedTools.Add("shell_execute"); + _fakeChatClient.ToolCallsOnFirstCall = + [ + new FunctionCallContent("call-shell-buffered-input", "shell_execute", + new Dictionary { ["command"] = "git status" }) + ]; + + var sessionId = new SessionId("test-channel/restart-drain-buffered-input"); + var sessionManager = ActorRegistry.Get(); + var subscriber = CreateTestProbe("restart-drain-buffered-sub"); + await sessionManager.Ask(new JoinSession(subscriber) + { + SessionId = sessionId, + Filter = OutputFilter.Full + }, TimeSpan.FromSeconds(10), TestContext.Current.CancellationToken); + await subscriber.ExpectMsgAsync(cancellationToken: TestContext.Current.CancellationToken); + + await sessionManager.Ask(new SendUserMessage + { + SessionId = sessionId, + Content = "Run git status", + Source = RequesterSource("U-requester") + }, TimeSpan.FromSeconds(5), TestContext.Current.CancellationToken); + await subscriber.ExpectMsgAsync( + TimeSpan.FromSeconds(5), cancellationToken: TestContext.Current.CancellationToken); + await subscriber.ExpectMsgAsync( + TimeSpan.FromSeconds(5), cancellationToken: TestContext.Current.CancellationToken); + + await sessionManager.Ask(new SendUserMessage + { + SessionId = sessionId, + Content = "Also check the branch", + Source = RequesterSource("U-requester") + }, TimeSpan.FromSeconds(5), TestContext.Current.CancellationToken); + + var escapedId = Uri.EscapeDataString(sessionId.Value); + var child = await Sys.ActorSelection($"/user/session-manager/{escapedId}") + .ResolveOne(TimeSpan.FromSeconds(5), TestContext.Current.CancellationToken); + var drainProbe = CreateTestProbe("restart-drain-buffered-ack"); + child.Tell(new PrepareForDaemonRestart(sessionId, "config-reload"), drainProbe.Ref); + + await drainProbe.ExpectNoMsgAsync(TimeSpan.FromMilliseconds(250), + TestContext.Current.CancellationToken); + Assert.Equal(0, _toolExecutor.SuccessfulExecutions); + } + [Fact] public async Task Passivated_session_resumes_tool_batch_when_approval_arrives() { diff --git a/src/Netclaw.Actors.Tests/Sessions/SessionPhaseTransitionTests.cs b/src/Netclaw.Actors.Tests/Sessions/SessionPhaseTransitionTests.cs index be88c1a8e..c21280fcb 100644 --- a/src/Netclaw.Actors.Tests/Sessions/SessionPhaseTransitionTests.cs +++ b/src/Netclaw.Actors.Tests/Sessions/SessionPhaseTransitionTests.cs @@ -17,6 +17,7 @@ public sealed class SessionPhaseTransitionTests [InlineData(SessionPhase.Ready, SessionPhase.Passivating)] [InlineData(SessionPhase.Processing, SessionPhase.Ready)] [InlineData(SessionPhase.Processing, SessionPhase.Compacting)] + [InlineData(SessionPhase.Processing, SessionPhase.Passivating)] [InlineData(SessionPhase.Compacting, SessionPhase.Ready)] [InlineData(SessionPhase.Compacting, SessionPhase.Processing)] [InlineData(SessionPhase.Passivating, SessionPhase.Ready)] diff --git a/src/Netclaw.Actors.Tests/Sessions/ToolApprovalStateTests.cs b/src/Netclaw.Actors.Tests/Sessions/ToolApprovalStateTests.cs index 4e2a0b0d7..0642292aa 100644 --- a/src/Netclaw.Actors.Tests/Sessions/ToolApprovalStateTests.cs +++ b/src/Netclaw.Actors.Tests/Sessions/ToolApprovalStateTests.cs @@ -36,6 +36,30 @@ public void Resolve_moves_one_call_from_pending_to_resolved() Assert.False(state.Resolve(request.CallId, ApprovalDecision.Denied, out _)); } + [Fact] + public void Restart_stop_requires_an_unresolved_durable_prompt_with_restored_authority() + { + var state = new ToolApprovalState(); + var request = CreateRequest("call-1", requestedAtMs: 10); + + state.Request(request, persistApprovalState: false, recovered: false); + Assert.False(state.HasRecoverablePending(request.CallId)); + + state.Request(request, persistApprovalState: true, recovered: false); + Assert.True(state.HasRecoverablePending(request.CallId)); + + Assert.True(state.Resolve(request.CallId, ApprovalDecision.ApprovedOnce, out _)); + Assert.False(state.HasRecoverablePending(request.CallId)); + + var legacy = request with { CallId = "legacy-restorable", TurnContext = null }; + state.Request(legacy, persistApprovalState: true, recovered: true); + Assert.False(state.HasRecoverablePending(legacy.CallId)); + + var incomplete = request with { CallId = "legacy-call", TurnContext = null, ChannelType = null }; + state.Request(incomplete, persistApprovalState: true, recovered: true); + Assert.False(state.HasRecoverablePending(incomplete.CallId)); + } + [Fact] public void Concurrent_requests_wait_until_the_last_call_resolves() { diff --git a/src/Netclaw.Actors/Sessions/ActiveToolBatchTracker.cs b/src/Netclaw.Actors/Sessions/ActiveToolBatchTracker.cs index 7d92b9bf6..d47a91f96 100644 --- a/src/Netclaw.Actors/Sessions/ActiveToolBatchTracker.cs +++ b/src/Netclaw.Actors/Sessions/ActiveToolBatchTracker.cs @@ -17,6 +17,12 @@ internal sealed class ActiveToolBatchTracker private readonly Dictionary _cycleResults = new(StringComparer.Ordinal); private PreparedToolCycleBatch? _preparedCycleBatch; + public long Generation { get; private set; } + + public Task? ExecutionTask { get; private set; } + + public bool RestartStopRequested { get; private set; } + public int CompletedCount => _completedCallIds.Count; public bool HasAllResults => _expectedCallIds.Count > 0 @@ -31,6 +37,7 @@ public void Start( SerializableChatMessage assistantMessage, IEnumerable existingResults) { + ResetExecution(); _preparedCycleBatch = null; _cycleResults.Clear(); ClearExpectedCallIds(); @@ -51,6 +58,7 @@ public void Start( IEnumerable toolCalls, PreparedToolCycleBatch? preparedCycleBatch) { + ResetExecution(); _preparedCycleBatch = preparedCycleBatch; _cycleResults.Clear(); ClearExpectedCallIds(); @@ -78,8 +86,39 @@ public void RecordCycleResult( public void MarkExecutionTaskCompleted() => ExecutionTaskCompleted = true; + public void SetExecutionTask(Task task) + { + ArgumentNullException.ThrowIfNull(task); + ExecutionTask = task; + } + + public bool CanStopForDurableApprovals(Func hasDurableApproval) + { + if (_expectedCallIds.Count == 0 || ExecutionTask is null + || ExecutionTask.IsCompleted || RestartStopRequested) + return false; + + var hasUnfinishedApproval = false; + foreach (var callId in _expectedCallIds) + { + if (_completedCallIds.Contains(callId)) + continue; + + if (!hasDurableApproval(callId)) + return false; + + hasUnfinishedApproval = true; + } + + return hasUnfinishedApproval; + } + + public void MarkRestartStopRequested() + => RestartStopRequested = true; + public void Clear() { + ResetExecution(); ClearExpectedCallIds(); ClearCompletedCallIds(); _preparedCycleBatch = null; @@ -87,6 +126,13 @@ public void Clear() ExecutionTaskCompleted = false; } + private void ResetExecution() + { + Generation++; + ExecutionTask = null; + RestartStopRequested = false; + } + private void ClearExpectedCallIds() => _expectedCallIds.Clear(); diff --git a/src/Netclaw.Actors/Sessions/LlmSessionActor.cs b/src/Netclaw.Actors/Sessions/LlmSessionActor.cs index be142d93e..2a97ea01b 100644 --- a/src/Netclaw.Actors/Sessions/LlmSessionActor.cs +++ b/src/Netclaw.Actors/Sessions/LlmSessionActor.cs @@ -192,6 +192,9 @@ public sealed class LlmSessionActor : ReceivePersistentActor, IWithTimers private bool _passivationCompleted; private bool _passivationFinalStopScheduled; + private sealed record ToolPipelineStoppedForRestart(long Generation, Exception? Failure) + : INoSerializationVerificationNeeded; + // Reap-on-passivation handshake: while a KillJobsForSession ask is in // flight, the final snapshot is deferred so it captures the reaped marks. // _jobReapEpoch is bumped per reap request so a late reply from a @@ -464,6 +467,7 @@ private void Ready() Command(msg => Sender.Tell(Context.ActorOf(msg.Props, msg.ActorName))); Command(HandleDeliveryFailedWhenReady); Command(_ => RequestRestartDrain()); + Command(_ => { }); // Approval click for a tool batch that parked while the session was // idle (deferred passivation) or that survived cold recovery. The @@ -553,6 +557,7 @@ private void Processing() ApplyToolCallRecorded(evt); ProcessToolCallResult(result); TryCompleteStreamedToolBatch(); + TryStopDurableApprovalWaits(); }); }); @@ -568,6 +573,10 @@ private void Processing() Command(msg => { + if (_activeToolBatch.RestartStopRequested + && msg.Cause is TimeoutException { InnerException: OperationCanceledException }) + return; + _watchdog.Stop(Timers); CancelAndDisposeToolExecutionCts(); _mediaBuffer.Clear(); @@ -741,6 +750,7 @@ private void Processing() }, OutputFilter.ToolCalls); }); Command(_ => RequestRestartDrain()); + Command(HandleToolPipelineStoppedForRestart); CommandDistillationAckNoOp(); CommandJobReapResolved(); } @@ -2089,7 +2099,7 @@ await self.Ask( CancellationToken = toolExecutionCt }; - _ = pipeline.ExecuteAsync(batch); + _activeToolBatch.SetExecutionTask(pipeline.ExecuteAsync(batch)); } private void HandleTextResponse( @@ -3629,6 +3639,7 @@ private void HandleToolInteractionRequestDispatch(ToolInteractionRequestDispatch { ApplyToolApprovalRequested(e); EmitOutput(msg); + TryStopDurableApprovalWaits(); }); } @@ -4977,6 +4988,55 @@ private void RequestRestartDrain() if (_phase.Current == SessionPhase.Ready) TransitionTo(SessionPhase.Passivating); + else if (_phase.Current == SessionPhase.Processing) + TryStopDurableApprovalWaits(); + } + + private void TryStopDurableApprovalWaits() + { + if (!_restartDrainRequested || _phase.Current != SessionPhase.Processing + || _buffer.Count > 0 || _deferredApprovalResponse is not null + || _activeToolExecutionCts is null + || !_activeToolBatch.CanStopForDurableApprovals(_toolApprovals.HasRecoverablePending)) + return; + + var task = _activeToolBatch.ExecutionTask + ?? throw new InvalidOperationException("An eligible tool batch requires an execution task."); + var generation = _activeToolBatch.Generation; + _activeToolBatch.MarkRestartStopRequested(); + _log.Info("Stopping a tool batch that waits only for durable approvals before restart drain"); + _activeToolExecutionCts.Cancel(); + _ = ReportToolPipelineStopAsync(task, Self, generation); + } + + private static async Task ReportToolPipelineStopAsync(Task task, IActorRef actor, long generation) + { + try + { + await task.ConfigureAwait(false); + actor.Tell(new ToolPipelineStoppedForRestart(generation, null)); + } + catch (Exception ex) + { + actor.Tell(new ToolPipelineStoppedForRestart(generation, ex)); + } + } + + private void HandleToolPipelineStoppedForRestart(ToolPipelineStoppedForRestart stopped) + { + if (!_activeToolBatch.RestartStopRequested || stopped.Generation != _activeToolBatch.Generation) + return; + + if (stopped.Failure is { } failure) + { + _log.Error(failure, "The tool task failed during approval-only restart drain"); + FailCurrentTurn("The tool task failed during restart drain.", failure, ErrorCategory.ToolFailure); + return; + } + + CancelAndDisposeToolExecutionCts(); + ClearActiveToolBatchTracking(); + TransitionTo(SessionPhase.Passivating); } private void ClearBufferedMessagesForRestartDrain() diff --git a/src/Netclaw.Actors/Sessions/SessionPhase.cs b/src/Netclaw.Actors/Sessions/SessionPhase.cs index 5b10bf269..17b7e50c3 100644 --- a/src/Netclaw.Actors/Sessions/SessionPhase.cs +++ b/src/Netclaw.Actors/Sessions/SessionPhase.cs @@ -33,7 +33,7 @@ internal static class SessionPhaseTransitions { SessionPhase.Recovering => to == SessionPhase.Ready, SessionPhase.Ready => to is SessionPhase.Processing or SessionPhase.Compacting or SessionPhase.Passivating, - SessionPhase.Processing => to is SessionPhase.Ready or SessionPhase.Compacting, + SessionPhase.Processing => to is SessionPhase.Ready or SessionPhase.Compacting or SessionPhase.Passivating, SessionPhase.Compacting => to is SessionPhase.Ready or SessionPhase.Processing, SessionPhase.Passivating => to is SessionPhase.Ready, _ => false diff --git a/src/Netclaw.Actors/Sessions/ToolApprovalState.cs b/src/Netclaw.Actors/Sessions/ToolApprovalState.cs index 9c337c12c..b35908ec2 100644 --- a/src/Netclaw.Actors/Sessions/ToolApprovalState.cs +++ b/src/Netclaw.Actors/Sessions/ToolApprovalState.cs @@ -135,6 +135,13 @@ public bool TryGetResolved(string callId, out ResolvedToolApproval resolved) public bool HasPending(string callId) => _calls.TryGetValue(callId, out var call) && call is PendingToolApproval; + public bool HasRecoverablePending(string callId) + => TryGetPending(callId, out var pending) + && pending.PersistApprovalState + && pending.Request.TurnContext is not null + && pending.TurnContext is not null + && pending.TurnContextRestoreFailure is null; + public bool HasResolved(string callId) => _calls.TryGetValue(callId, out var call) && call is ResolvedToolApproval;