From 4634b0e875947105b46c3cf5c29f1db7ed848a08 Mon Sep 17 00:00:00 2001 From: Aaron Stannard Date: Tue, 14 Jul 2026 21:48:56 +0000 Subject: [PATCH 1/4] docs(tools): record stage one delivery --- openspec/changes/simplify-tool-execution-context/tasks.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/openspec/changes/simplify-tool-execution-context/tasks.md b/openspec/changes/simplify-tool-execution-context/tasks.md index 0da879d5a..a93c4293a 100644 --- a/openspec/changes/simplify-tool-execution-context/tasks.md +++ b/openspec/changes/simplify-tool-execution-context/tasks.md @@ -6,7 +6,7 @@ - [x] 1.4 Split mutable tool outputs into a per-invocation append-only sink and approval retry/match state into a pipeline-owned attempt object while sharing only immutable run authority across a batch. - [x] 1.5 Add focused tests proving invalid scope values fail before dispatch, missing authority has no dispatch path, and parallel calls cannot observe each other's mutable state. - [x] 1.6 Update affected engineering documentation; review the mapped `netclaw-operations` system skill and leave it unchanged because the internal refactor must not alter model-visible guidance. -- [ ] 1.7 Run targeted tests, tool-related evals/full eval suite as required, `dotnet test`, Slopwatch, file-header verification, and `git diff --check`; open and babysit Stage 1 through review, CI, merge, and post-merge `dev` verification. +- [x] 1.7 Run targeted tests, tool-related evals/full eval suite as required, `dotnet test`, Slopwatch, file-header verification, and `git diff --check`; open and babysit Stage 1 through review, CI, merge, and post-merge `dev` verification. ## 2. Stage 2 — Composed Session Pipeline From 6281312fd0f5ebec04e980b0679ff04d239fb3b1 Mon Sep 17 00:00:00 2001 From: Aaron Stannard Date: Tue, 14 Jul 2026 22:18:21 +0000 Subject: [PATCH 2/4] refactor(tools): compose session execution pipeline --- IMPLEMENTATION_PLAN.md | 2 +- ...SPEC-002-session-lifecycle-and-protocol.md | 15 + .../specs/netclaw-session/spec.md | 51 ++ .../simplify-tool-execution-context/tasks.md | 10 +- .../Sessions/LlmSessionTestExtensions.cs | 2 +- .../Pipelines/BackgroundRoutingTests.cs | 210 ++++--- .../Pipelines/MetaValidationAndNoticeTests.cs | 57 +- .../SessionToolPipelineTestFixture.cs | 204 +++++++ .../SessionToolExecutionPipelineTests.cs | 418 ++++++------- .../Sessions/LlmSessionActor.cs | 68 ++- .../Pipelines/SessionToolExecutionPipeline.cs | 550 ++++++++++-------- .../Sessions/SessionDependencies.cs | 2 +- src/Netclaw.Actors/SubAgents/SubAgentActor.cs | 2 +- src/Netclaw.Actors/Tools/IToolExecutor.cs | 18 + src/Netclaw.Daemon/Program.cs | 2 +- 15 files changed, 943 insertions(+), 668 deletions(-) create mode 100644 openspec/changes/simplify-tool-execution-context/specs/netclaw-session/spec.md create mode 100644 src/Netclaw.Actors.Tests/Sessions/Pipelines/SessionToolPipelineTestFixture.cs diff --git a/IMPLEMENTATION_PLAN.md b/IMPLEMENTATION_PLAN.md index 7fa809201..302d89b20 100644 --- a/IMPLEMENTATION_PLAN.md +++ b/IMPLEMENTATION_PLAN.md @@ -125,7 +125,7 @@ working state with gated asynchronous Git enrichment. Done when: -- [ ] Stage 1 lands required run scopes, per-call isolation, and non-null +- [x] Stage 1 lands required run scopes, per-call isolation, and non-null security/authority dependencies without compatibility shims. - [ ] Stage 2 lands the composed pipeline without changing existing background, fallback, authorization, approval, MCP, or model-visible behavior. diff --git a/docs/spec/SPEC-002-session-lifecycle-and-protocol.md b/docs/spec/SPEC-002-session-lifecycle-and-protocol.md index c503b0f9d..b70daccca 100644 --- a/docs/spec/SPEC-002-session-lifecycle-and-protocol.md +++ b/docs/spec/SPEC-002-session-lifecycle-and-protocol.md @@ -41,6 +41,21 @@ This enables: 5. Actor emits typed `SessionOutput` events to subscribers. 6. Actor checks compaction threshold. +### Tool Execution Pipeline + +Tool-enabled sessions compose one `SessionToolExecutionPipeline` from required +execution, audit, time, and logging services. Each admitted tool-call response +is submitted as one `SessionToolBatch`; the batch derives its immutable tool +authority from the admitted `TurnContext` and carries environment and +per-batch capabilities separately. Callers cannot supply a second authority +object that disagrees with the admitted turn. + +The pipeline executes calls concurrently with fresh invocation state per call. +Unavailable background-job infrastructure is an explicit capability state and +retains synchronous execution behavior. This internal composition does not +change MCP schemas, persisted actor messages, approval outcomes, or model-facing +tool results. + ## Subscriber Model Subscribers join via `JoinSession` with an `OutputFilter` bitmask controlling diff --git a/openspec/changes/simplify-tool-execution-context/specs/netclaw-session/spec.md b/openspec/changes/simplify-tool-execution-context/specs/netclaw-session/spec.md new file mode 100644 index 000000000..165200a16 --- /dev/null +++ b/openspec/changes/simplify-tool-execution-context/specs/netclaw-session/spec.md @@ -0,0 +1,51 @@ +## MODIFIED Requirements + +### Requirement: Tool execution encapsulation + +Tool execution SHALL be encapsulated in a composed `SessionToolExecutionPipeline` +whose unconditional production services are required constructor dependencies. +The session actor SHALL submit one cohesive batch command whose tool authority is +derived from the admitted `TurnContext`; callers SHALL NOT be able to supply a +second, conflicting authority source. Genuinely unavailable runtime capabilities, +including background-job dispatch, SHALL be represented explicitly while retaining +their existing behavior. The pipeline SHALL execute tool calls in parallel, track +sub-agent activity, and send completion or failure messages back to the actor. + +The pipeline SHALL NOT itself bound or clamp tool-result size. Bounding to the +inline budget and spilling overflow remains centralized in +`DispatchingToolExecutor`, so the pipeline stores the result already bounded by +the dispatcher. `SessionTuning.MaxInlineToolResultChars` remains the session +content budget used for tools without a smaller per-tool override. + +#### Scenario: Parallel tool execution + +- **GIVEN** an admitted turn whose LLM response contains three tool calls +- **WHEN** the session submits its `SessionToolBatch` +- **THEN** all three tool calls execute in parallel with fresh call-local state +- **AND** results are collected and returned through the existing actor protocol + +#### Scenario: Conflicting authority cannot be supplied + +- **GIVEN** a session constructs a tool batch from an admitted `TurnContext` +- **WHEN** the batch derives its tool run scope +- **THEN** session, audience, boundary, channel, delivery, and interactive-approval authority come from that turn context +- **AND** the caller has no initializer or alternate constructor for replacing the derived authority + +#### Scenario: Background manager is unavailable + +- **GIVEN** a valid background-capable shell request and no registered background-job manager +- **WHEN** the batch executes +- **THEN** the request executes synchronously as it did before the composition refactor +- **AND** manager absence is not inferred from a nullable security dependency + +#### Scenario: Tool execution timeout + +- **GIVEN** tool execution is in progress +- **WHEN** the configured `ToolExecutionTimeout` elapses +- **THEN** the pipeline sends `ToolExecutionFailed` with a `TimeoutException` + +#### Scenario: Oversized result already bounded by the dispatcher + +- **GIVEN** a tool returns an oversized result +- **WHEN** it reaches the pipeline +- **THEN** the pipeline stores it as-is without re-clamping diff --git a/openspec/changes/simplify-tool-execution-context/tasks.md b/openspec/changes/simplify-tool-execution-context/tasks.md index a93c4293a..12d9bc9bf 100644 --- a/openspec/changes/simplify-tool-execution-context/tasks.md +++ b/openspec/changes/simplify-tool-execution-context/tasks.md @@ -10,11 +10,11 @@ ## 2. Stage 2 — Composed Session Pipeline -- [ ] 2.1 Replace the broad session tool-call parameter list with a cohesive batch command and a composed `SessionToolExecutionPipeline` whose production dependencies are required. -- [ ] 2.2 Trace each nullable pipeline service through every intended production composition path; make proven-unconditional services required, model genuinely production-reachable absence explicitly with unchanged behavior, and keep test-only fixture states out of the production API. -- [ ] 2.3 Preserve existing `_background` behavior for shell, non-shell, missing-manager, and dispatch-failure paths while removing redundant parameter plumbing. -- [ ] 2.4 Add characterization tests for audit/logging/approval/background infrastructure, malformed metadata, ACL and approval denial, supported background routing, missing-manager fallback, dispatch failure, and non-shell fallback. -- [ ] 2.5 Verify MCP request/response schemas and persisted actor contracts remain compatible; update affected engineering docs, specs, and the versioned `netclaw-operations` system skill. +- [x] 2.1 Replace the broad session tool-call parameter list with a cohesive batch command and a composed `SessionToolExecutionPipeline` whose production dependencies are required. +- [x] 2.2 Trace each nullable pipeline service through every intended production composition path; make proven-unconditional services required, model genuinely production-reachable absence explicitly with unchanged behavior, and keep test-only fixture states out of the production API. +- [x] 2.3 Preserve existing `_background` behavior for shell, non-shell, missing-manager, and dispatch-failure paths while removing redundant parameter plumbing. +- [x] 2.4 Add characterization tests for audit/logging/approval/background infrastructure, malformed metadata, ACL and approval denial, supported background routing, missing-manager fallback, dispatch failure, and non-shell fallback. +- [x] 2.5 Verify MCP request/response schemas and persisted actor contracts remain compatible; update affected engineering docs and specs, and review the versioned `netclaw-operations` system skill without changing model-visible guidance for an internal behavior-preserving refactor. - [ ] 2.6 Run targeted tests, the tool-definition eval suite, `dotnet test`, Slopwatch, file-header verification, and `git diff --check`; open and babysit Stage 2 through review, CI, merge, and post-merge `dev` verification. ## 3. Stage 3 — Child Context and Async Git diff --git a/src/Netclaw.Actors.Tests/Sessions/LlmSessionTestExtensions.cs b/src/Netclaw.Actors.Tests/Sessions/LlmSessionTestExtensions.cs index ff1a93492..eaf453f0b 100644 --- a/src/Netclaw.Actors.Tests/Sessions/LlmSessionTestExtensions.cs +++ b/src/Netclaw.Actors.Tests/Sessions/LlmSessionTestExtensions.cs @@ -44,7 +44,7 @@ public static IServiceCollection AddLlmSessionCompositeRecords(this IServiceColl { services.TryAddSingleton(sp => new SessionToolServices( sp.GetRequiredService(), - sp.GetService(), + sp.GetService() ?? NullToolAuditLogger.Instance, sp.GetRequiredService(), sp.GetService(), sp.GetService(), diff --git a/src/Netclaw.Actors.Tests/Sessions/Pipelines/BackgroundRoutingTests.cs b/src/Netclaw.Actors.Tests/Sessions/Pipelines/BackgroundRoutingTests.cs index 12582ae6e..823b1c446 100644 --- a/src/Netclaw.Actors.Tests/Sessions/Pipelines/BackgroundRoutingTests.cs +++ b/src/Netclaw.Actors.Tests/Sessions/Pipelines/BackgroundRoutingTests.cs @@ -41,20 +41,10 @@ public async Task TimeoutAlone_DoesNotRouteToBackground() }) }; - await SessionToolExecutionPipeline.ExecuteToolsAsync( - executor, toolCalls, - new SessionId("test/timeout-only"), - source: null, - auditLogger: null, - timeProvider: TimeProvider.System, - sessionDir: Path.GetTempPath(), - maxInlineToolResultChars: 4096, - timeout: TimeSpan.FromSeconds(5), - self: probe.Ref, - emitSubAgentOutput: _ => { }, - spawnChildActor: static (_, _, _) => Task.FromResult(new object()), - backgroundJobManager: jobManagerProbe.Ref, - ct: TestContext.Current.CancellationToken); + await new SessionToolPipelineTestFixture( + executor, toolCalls, new SessionId("test/timeout-only"), probe.Ref) + .WithBackgroundJobs(jobManagerProbe.Ref) + .ExecuteAsync(TestContext.Current.CancellationToken); var completed = await probe.ExpectMsgAsync( TimeSpan.FromSeconds(5), @@ -88,20 +78,11 @@ public async Task ExplicitBackground_RoutesShellToBackgroundManager() }) }; - await SessionToolExecutionPipeline.ExecuteToolsAsync( - executor, toolCalls, - new SessionId("test/background"), - source: TestMessageSource(), - auditLogger: null, - timeProvider: TimeProvider.System, - sessionDir: Path.GetTempPath(), - maxInlineToolResultChars: 4096, - timeout: TimeSpan.FromSeconds(5), - self: probe.Ref, - emitSubAgentOutput: _ => { }, - spawnChildActor: static (_, _, _) => Task.FromResult(new object()), - backgroundJobManager: fakeJobManager, - ct: TestContext.Current.CancellationToken); + await new SessionToolPipelineTestFixture( + executor, toolCalls, new SessionId("test/background"), probe.Ref) + .From(TestMessageSource()) + .WithBackgroundJobs(fakeJobManager) + .ExecuteAsync(TestContext.Current.CancellationToken); var completed = await probe.ExpectMsgAsync( TimeSpan.FromSeconds(5), @@ -139,20 +120,11 @@ public async Task ExplicitBackground_HonorsRequestedTimeout() }) }; - await SessionToolExecutionPipeline.ExecuteToolsAsync( - executor, toolCalls, - new SessionId("test/background-timeout"), - source: TestMessageSource(), - auditLogger: null, - timeProvider: TimeProvider.System, - sessionDir: Path.GetTempPath(), - maxInlineToolResultChars: 4096, - timeout: TimeSpan.FromSeconds(5), - self: probe.Ref, - emitSubAgentOutput: _ => { }, - spawnChildActor: static (_, _, _) => Task.FromResult(new object()), - backgroundJobManager: fakeJobManager, - ct: TestContext.Current.CancellationToken); + await new SessionToolPipelineTestFixture( + executor, toolCalls, new SessionId("test/background-timeout"), probe.Ref) + .From(TestMessageSource()) + .WithBackgroundJobs(fakeJobManager) + .ExecuteAsync(TestContext.Current.CancellationToken); await probe.ExpectMsgAsync( TimeSpan.FromSeconds(5), @@ -185,20 +157,11 @@ public async Task ExplicitBackground_OmittedTimeout_ArmsNoKillTimer() }) }; - await SessionToolExecutionPipeline.ExecuteToolsAsync( - executor, toolCalls, - new SessionId("test/background-notimer"), - source: TestMessageSource(), - auditLogger: null, - timeProvider: TimeProvider.System, - sessionDir: Path.GetTempPath(), - maxInlineToolResultChars: 4096, - timeout: TimeSpan.FromSeconds(5), - self: probe.Ref, - emitSubAgentOutput: _ => { }, - spawnChildActor: static (_, _, _) => Task.FromResult(new object()), - backgroundJobManager: fakeJobManager, - ct: TestContext.Current.CancellationToken); + await new SessionToolPipelineTestFixture( + executor, toolCalls, new SessionId("test/background-notimer"), probe.Ref) + .From(TestMessageSource()) + .WithBackgroundJobs(fakeJobManager) + .ExecuteAsync(TestContext.Current.CancellationToken); await probe.ExpectMsgAsync( TimeSpan.FromSeconds(5), @@ -228,20 +191,11 @@ public async Task ExplicitBackground_SubmitAckIncludesOutputLogPath() }) }; - await SessionToolExecutionPipeline.ExecuteToolsAsync( - executor, toolCalls, - new SessionId("test/background-logpath"), - source: TestMessageSource(), - auditLogger: null, - timeProvider: TimeProvider.System, - sessionDir: Path.GetTempPath(), - maxInlineToolResultChars: 4096, - timeout: TimeSpan.FromSeconds(5), - self: probe.Ref, - emitSubAgentOutput: _ => { }, - spawnChildActor: static (_, _, _) => Task.FromResult(new object()), - backgroundJobManager: fakeJobManager, - ct: TestContext.Current.CancellationToken); + await new SessionToolPipelineTestFixture( + executor, toolCalls, new SessionId("test/background-logpath"), probe.Ref) + .From(TestMessageSource()) + .WithBackgroundJobs(fakeJobManager) + .ExecuteAsync(TestContext.Current.CancellationToken); var completed = await probe.ExpectMsgAsync( TimeSpan.FromSeconds(5), @@ -272,20 +226,11 @@ public async Task ExplicitBackground_PreservesWorkingDirectory() }) }; - await SessionToolExecutionPipeline.ExecuteToolsAsync( - executor, toolCalls, - new SessionId("test/background-dir"), - source: TestMessageSource(), - auditLogger: null, - timeProvider: TimeProvider.System, - sessionDir: Path.GetTempPath(), - maxInlineToolResultChars: 4096, - timeout: TimeSpan.FromSeconds(5), - self: probe.Ref, - emitSubAgentOutput: _ => { }, - spawnChildActor: static (_, _, _) => Task.FromResult(new object()), - backgroundJobManager: fakeJobManager, - ct: TestContext.Current.CancellationToken); + await new SessionToolPipelineTestFixture( + executor, toolCalls, new SessionId("test/background-dir"), probe.Ref) + .From(TestMessageSource()) + .WithBackgroundJobs(fakeJobManager) + .ExecuteAsync(TestContext.Current.CancellationToken); await probe.ExpectMsgAsync( TimeSpan.FromSeconds(5), @@ -314,20 +259,10 @@ public async Task ExplicitBackground_DeniedByAuthorization_DoesNotRouteToBackgro }) }; - await SessionToolExecutionPipeline.ExecuteToolsAsync( - executor, toolCalls, - new SessionId("test/background-denied"), - source: null, - auditLogger: null, - timeProvider: TimeProvider.System, - sessionDir: Path.GetTempPath(), - maxInlineToolResultChars: 4096, - timeout: TimeSpan.FromSeconds(5), - self: probe.Ref, - emitSubAgentOutput: _ => { }, - spawnChildActor: static (_, _, _) => Task.FromResult(new object()), - backgroundJobManager: jobManagerProbe.Ref, - ct: TestContext.Current.CancellationToken); + await new SessionToolPipelineTestFixture( + executor, toolCalls, new SessionId("test/background-denied"), probe.Ref) + .WithBackgroundJobs(jobManagerProbe.Ref) + .ExecuteAsync(TestContext.Current.CancellationToken); var completed = await probe.ExpectMsgAsync( TimeSpan.FromSeconds(5), @@ -357,20 +292,10 @@ public async Task NonShellToolWithBackground_ExecutesSynchronously() }) }; - await SessionToolExecutionPipeline.ExecuteToolsAsync( - executor, toolCalls, - new SessionId("test/nonshell-bg"), - source: null, - auditLogger: null, - timeProvider: TimeProvider.System, - sessionDir: Path.GetTempPath(), - maxInlineToolResultChars: 4096, - timeout: TimeSpan.FromSeconds(5), - self: probe.Ref, - emitSubAgentOutput: _ => { }, - spawnChildActor: static (_, _, _) => Task.FromResult(new object()), - backgroundJobManager: jobManagerProbe.Ref, - ct: TestContext.Current.CancellationToken); + await new SessionToolPipelineTestFixture( + executor, toolCalls, new SessionId("test/nonshell-bg"), probe.Ref) + .WithBackgroundJobs(jobManagerProbe.Ref) + .ExecuteAsync(TestContext.Current.CancellationToken); var completed = await probe.ExpectMsgAsync( TimeSpan.FromSeconds(5), @@ -384,6 +309,58 @@ await jobManagerProbe.ExpectNoMsgAsync( cancellationToken: TestContext.Current.CancellationToken); } + [Fact] + public async Task ExplicitBackground_WithoutManager_ExecutesSynchronously() + { + var executor = new EchoExecutor(); + var probe = CreateTestProbe("pipeline-no-background-manager"); + var toolCalls = new List + { + new("call-bg-no-manager", "shell_execute", new Dictionary + { + ["command"] = "echo fallback", + ["_background"] = true + }) + }; + + await new SessionToolPipelineTestFixture( + executor, toolCalls, new SessionId("test/background-no-manager"), probe.Ref) + .From(TestMessageSource()) + .ExecuteAsync(TestContext.Current.CancellationToken); + + var completed = await probe.ExpectMsgAsync( + TimeSpan.FromSeconds(5), + cancellationToken: TestContext.Current.CancellationToken); + Assert.Equal("echo:echo fallback", Assert.Single(completed.ToolResults).Content); + } + + [Fact] + public async Task BackgroundManagerFailure_ReturnsSubmissionErrorWithoutSynchronousRetry() + { + var executor = new EchoExecutor(); + var probe = CreateTestProbe("pipeline-failing-background-manager"); + var manager = Sys.ActorOf(Props.Create(() => new FailingJobManager())); + var toolCalls = new List + { + new("call-bg-failure", "shell_execute", new Dictionary + { + ["command"] = "long-running-command", + ["_background"] = true + }) + }; + + await new SessionToolPipelineTestFixture( + executor, toolCalls, new SessionId("test/background-failure"), probe.Ref) + .From(TestMessageSource()) + .WithBackgroundJobs(manager) + .ExecuteAsync(TestContext.Current.CancellationToken); + + var completed = await probe.ExpectMsgAsync( + TimeSpan.FromSeconds(5), + cancellationToken: TestContext.Current.CancellationToken); + Assert.Contains("Error submitting background job", Assert.Single(completed.ToolResults).Content); + } + // Background-job submission now requires a trust context — source cannot be null. // This factory produces a minimal Personal-audience source for tests that route // to the background job manager and don't need to assert on trust-context values. @@ -434,4 +411,13 @@ public FakeJobManager(IActorRef probe) }); } } + + private sealed class FailingJobManager : ReceiveActor + { + public FailingJobManager() + { + Receive(_ => + Sender.Tell(new Status.Failure(new InvalidOperationException("dispatch failed")))); + } + } } diff --git a/src/Netclaw.Actors.Tests/Sessions/Pipelines/MetaValidationAndNoticeTests.cs b/src/Netclaw.Actors.Tests/Sessions/Pipelines/MetaValidationAndNoticeTests.cs index 6680068ff..989454beb 100644 --- a/src/Netclaw.Actors.Tests/Sessions/Pipelines/MetaValidationAndNoticeTests.cs +++ b/src/Netclaw.Actors.Tests/Sessions/Pipelines/MetaValidationAndNoticeTests.cs @@ -62,21 +62,11 @@ private async Task RunPipelineAsync( new("call-1", "shell_execute", args) }; - var pipelineTask = SessionToolExecutionPipeline.ExecuteToolsAsync( - executor, - toolCalls, - sessionId, - source: null, - auditLogger: null, - timeProvider: TimeProvider.System, - sessionDir: Path.GetTempPath(), - maxInlineToolResultChars: 4096, - timeout: timeout ?? TimeSpan.FromSeconds(60), - self: probe.Ref, - emitSubAgentOutput: _ => { }, - spawnChildActor: static (_, _, _) => Task.FromResult(new object()), - turnContext: InteractiveTurnContext(sessionId), - ct: TestContext.Current.CancellationToken); + var fixture = new SessionToolPipelineTestFixture(executor, toolCalls, sessionId, probe.Ref) + .WithTurnContext(InteractiveTurnContext(sessionId)) + .WithTimeout(timeout ?? TimeSpan.FromSeconds(60)); + + var pipelineTask = fixture.ExecuteAsync(TestContext.Current.CancellationToken); var completed = await probe.ExpectMsgAsync( TimeSpan.FromSeconds(5), @@ -108,6 +98,13 @@ public Task ExecuteAsync(FunctionCallContent toolCall, ToolExecutionCont } } + private sealed class RecordingAuditLogger : IToolAuditLogger + { + public List Entries { get; } = []; + + public void Log(ToolAuditEntry entry) => Entries.Add(entry); + } + // ── Timeout hint is honored exactly (no clamp, no floor) ── [Fact] @@ -220,6 +217,36 @@ public async Task Non_boolean_background_value_rejects_without_dispatch() Assert.Contains("NOT executed", content); } + [Fact] + public async Task Malformed_metadata_is_audited_as_denied() + { + var executor = new EchoExecutor(); + var audit = new RecordingAuditLogger(); + var probe = CreateTestProbe("malformed-metadata-audit"); + var sessionId = new SessionId("D1/malformed-metadata-audit"); + var toolCalls = new List + { + new("call-invalid-background", "shell_execute", new Dictionary + { + ["Command"] = "echo hi", + ["_background"] = "yes" + }) + }; + + await new SessionToolPipelineTestFixture(executor, toolCalls, sessionId, probe.Ref) + .WithTurnContext(InteractiveTurnContext(sessionId)) + .WithAudit(audit) + .ExecuteAsync(TestContext.Current.CancellationToken); + + await probe.ExpectMsgAsync( + TimeSpan.FromSeconds(3), + cancellationToken: TestContext.Current.CancellationToken); + var entry = Assert.Single(audit.Entries); + Assert.False(entry.Allowed); + Assert.Equal("invalid_meta_value", entry.DenyReason); + Assert.Equal(new ToolCallId("call-invalid-background"), entry.CallId); + } + [Fact] public async Task Non_integral_json_timeout_rejects_without_uncaught_throw() { diff --git a/src/Netclaw.Actors.Tests/Sessions/Pipelines/SessionToolPipelineTestFixture.cs b/src/Netclaw.Actors.Tests/Sessions/Pipelines/SessionToolPipelineTestFixture.cs new file mode 100644 index 000000000..9d2aad375 --- /dev/null +++ b/src/Netclaw.Actors.Tests/Sessions/Pipelines/SessionToolPipelineTestFixture.cs @@ -0,0 +1,204 @@ +// ----------------------------------------------------------------------- +// +// Copyright (C) 2026 - 2026 Petabridge, LLC +// +// ----------------------------------------------------------------------- +using Akka.Actor; +using Akka.Event; +using Microsoft.Extensions.AI; +using Netclaw.Actors.Channels; +using Netclaw.Actors.Protocol; +using Netclaw.Actors.Sessions; +using Netclaw.Actors.Sessions.Pipelines; +using Netclaw.Actors.Tools; +using Netclaw.Configuration; +using Netclaw.Tools; +using static Netclaw.Actors.Sessions.SessionProtocol; + +namespace Netclaw.Actors.Tests.Sessions.Pipelines; + +internal sealed class SessionToolPipelineTestFixture( + IToolExecutor executor, + IReadOnlyList toolCalls, + SessionId sessionId, + IActorRef replyTo) +{ + private MessageSource? _source; + private TurnContext? _turnContext; + private IToolAuditLogger _auditLogger = NullToolAuditLogger.Instance; + private TimeProvider _timeProvider = TimeProvider.System; + private string _sessionDirectory = Path.GetTempPath(); + private InlineOutputBudget _inlineOutputBudget = new(4096); + private ToolExecutionTimeout _timeout = new(TimeSpan.FromSeconds(5)); + private Action _emitSubAgentOutput = _ => { }; + private Func> _spawnChildActor + = static (_, _, _) => Task.FromResult(new object()); + private IApprovalChannel _approvalChannel = new ApprovalChannel(); + private Action _emitApprovalRequest = _ => { }; + private ToolExecutionTimeout _approvalTimeout = new(Timeout.InfiniteTimeSpan); + private BackgroundJobDispatch _backgroundJobs = new BackgroundJobDispatch.Unavailable(); + private string? _projectDirectory; + private IReadOnlyList _recentFiles = []; + private bool _setWorkingDirectoryAvailable; + private bool _streamResults; + private ModelModality _modelInputModalities = ModelModality.Text; + private IReadOnlyDictionary> _oneTimeApprovalPreSeed + = new Dictionary>(); + private IReadOnlyDictionary _decisionOverrides + = new Dictionary(); + + public SessionToolPipelineTestFixture From(MessageSource source) + { + _source = source; + return this; + } + + public SessionToolPipelineTestFixture WithTurnContext(TurnContext turnContext) + { + _turnContext = turnContext; + return this; + } + + public SessionToolPipelineTestFixture WithAudit(IToolAuditLogger auditLogger) + { + _auditLogger = auditLogger; + return this; + } + + public SessionToolPipelineTestFixture WithTimeProvider(TimeProvider timeProvider) + { + _timeProvider = timeProvider; + return this; + } + + public SessionToolPipelineTestFixture InSessionDirectory(string sessionDirectory) + { + _sessionDirectory = sessionDirectory; + return this; + } + + public SessionToolPipelineTestFixture WithInlineOutputBudget(int characters) + { + _inlineOutputBudget = new InlineOutputBudget(characters); + return this; + } + + public SessionToolPipelineTestFixture WithTimeout(TimeSpan timeout) + { + _timeout = new ToolExecutionTimeout(timeout); + return this; + } + + public SessionToolPipelineTestFixture EmittingSubAgentOutput(Action emit) + { + _emitSubAgentOutput = emit; + return this; + } + + public SessionToolPipelineTestFixture SpawningChildrenWith( + Func> spawn) + { + _spawnChildActor = spawn; + return this; + } + + public SessionToolPipelineTestFixture WithApprovals( + IApprovalChannel channel, + Action emitRequest, + TimeSpan timeout) + { + _approvalChannel = channel; + _emitApprovalRequest = emitRequest; + _approvalTimeout = new ToolExecutionTimeout(timeout); + return this; + } + + public SessionToolPipelineTestFixture WithBackgroundJobs(IActorRef manager) + { + _backgroundJobs = new BackgroundJobDispatch.Available(manager); + return this; + } + + public SessionToolPipelineTestFixture InProject(string projectDirectory) + { + _projectDirectory = projectDirectory; + return this; + } + + public SessionToolPipelineTestFixture InProject( + string projectDirectory, + IReadOnlyList recentFiles) + { + _projectDirectory = projectDirectory; + _recentFiles = recentFiles; + return this; + } + + public SessionToolPipelineTestFixture WithSetWorkingDirectoryAvailable() + { + _setWorkingDirectoryAvailable = true; + return this; + } + + public SessionToolPipelineTestFixture StreamingResults() + { + _streamResults = true; + return this; + } + + public SessionToolPipelineTestFixture AcceptingModelInput(ModelModality modalities) + { + _modelInputModalities = modalities; + return this; + } + + public SessionToolPipelineTestFixture RedrivingApprovals( + IReadOnlyDictionary> preSeed, + IReadOnlyDictionary overrides) + { + _oneTimeApprovalPreSeed = preSeed; + _decisionOverrides = overrides; + return this; + } + + public Task ExecuteAsync(CancellationToken cancellationToken) + { + var turnContext = _turnContext ?? TurnContext.FromMessageSource( + sessionId, + new TurnId("test-tool-batch"), + _source); + var runEnvironment = new SessionToolRunEnvironment + { + SessionDirectory = _sessionDirectory, + InlineOutputBudget = _inlineOutputBudget, + ModelInputModalities = _modelInputModalities, + SpawnChildActor = _spawnChildActor, + ProjectDirectory = _projectDirectory, + RecentFiles = _recentFiles + }; + var pipeline = new SessionToolExecutionPipeline( + executor, + _auditLogger, + _timeProvider, + NoLogger.Instance); + var batch = new SessionToolBatch(turnContext, runEnvironment) + { + ToolCalls = toolCalls, + DefaultTimeout = _timeout, + ReplyTo = replyTo, + EmitSubAgentOutput = _emitSubAgentOutput, + ApprovalRequests = new ToolApprovalRequests( + _approvalChannel, + _emitApprovalRequest, + _approvalTimeout), + BackgroundJobs = _backgroundJobs, + SetWorkingDirectoryAvailable = _setWorkingDirectoryAvailable, + StreamResults = _streamResults, + OneTimeApprovalPreSeed = _oneTimeApprovalPreSeed, + DecisionOverrides = _decisionOverrides, + CancellationToken = cancellationToken + }; + + return pipeline.ExecuteAsync(batch); + } +} diff --git a/src/Netclaw.Actors.Tests/Sessions/SessionToolExecutionPipelineTests.cs b/src/Netclaw.Actors.Tests/Sessions/SessionToolExecutionPipelineTests.cs index 7ab1270c8..eb262fa08 100644 --- a/src/Netclaw.Actors.Tests/Sessions/SessionToolExecutionPipelineTests.cs +++ b/src/Netclaw.Actors.Tests/Sessions/SessionToolExecutionPipelineTests.cs @@ -5,6 +5,7 @@ // ----------------------------------------------------------------------- using System.Runtime.CompilerServices; using Akka.Actor; +using Akka.Event; using Akka.Hosting; using Akka.Hosting.TestKit; using Microsoft.Extensions.AI; @@ -15,6 +16,7 @@ using Netclaw.Actors.Protocol; using Netclaw.Actors.Sessions; using Netclaw.Actors.Sessions.Pipelines; +using Netclaw.Actors.Tests.Sessions.Pipelines; using Netclaw.Actors.Tools; using Netclaw.Configuration; using Netclaw.Tests.Utilities; @@ -47,6 +49,47 @@ protected override void ConfigureAkka(AkkaConfigurationBuilder builder, IService SupportsInteractiveApproval = true }; + [Fact] + public void Batch_derives_tool_authority_from_admitted_turn() + { + var turnContext = InteractiveTurnContext(new SessionId("D1/admitted-session")) with + { + DefaultDeliveryTarget = new ChannelDeliveryTargetInfo( + "signalr", "session", "default-target", "Default"), + RequestedDeliveryTarget = new ChannelDeliveryTargetInfo( + "signalr", "session", "requested-target", "Requested") + }; + var batch = new SessionToolBatch( + turnContext, + new SessionToolRunEnvironment + { + SessionDirectory = Path.GetTempPath(), + InlineOutputBudget = new InlineOutputBudget(4096), + SpawnChildActor = static (_, _, _) => Task.FromResult(new object()) + }) + { + ToolCalls = [new FunctionCallContent("call-1", "inspect_context")], + DefaultTimeout = new ToolExecutionTimeout(TimeSpan.FromSeconds(5)), + ReplyTo = ActorRefs.Nobody, + EmitSubAgentOutput = _ => { }, + ApprovalRequests = new ToolApprovalRequests( + new ApprovalChannel(), + _ => { }, + new ToolExecutionTimeout(Timeout.InfiniteTimeSpan)), + BackgroundJobs = new BackgroundJobDispatch.Unavailable(), + CancellationToken = TestContext.Current.CancellationToken + }; + + var session = Assert.IsType(batch.RunScope.Session); + Assert.Equal(turnContext.SessionId.Value, session.SessionId); + Assert.Equal(turnContext.Audience, batch.RunScope.Audience); + Assert.Equal(turnContext.Boundary, batch.RunScope.Boundary); + Assert.Equal(turnContext.ChannelType?.ToWireValue(), batch.RunScope.ChannelType); + Assert.Equal(turnContext.SupportsInteractiveApproval, batch.RunScope.SupportsInteractiveApproval); + Assert.Equal(turnContext.DefaultDeliveryTarget, batch.RunScope.DefaultDeliveryTarget); + Assert.Equal(turnContext.RequestedDeliveryTarget, batch.RunScope.RequestedDeliveryTarget); + } + [Fact] public async Task Approval_wait_does_not_consume_tool_execution_timeout_budget() { @@ -64,24 +107,14 @@ public async Task Approval_wait_does_not_consume_tool_execution_timeout_budget() }) }; - var pipelineTask = SessionToolExecutionPipeline.ExecuteToolsAsync( - executor, - toolCalls, - sessionId, - source: null, - auditLogger: null, - timeProvider: TimeProvider.System, - sessionDir: Path.GetTempPath(), - maxInlineToolResultChars: 4096, - timeout: TimeSpan.FromSeconds(1), - self: probe.Ref, - emitSubAgentOutput: _ => { }, - spawnChildActor: static (_, _, _) => Task.FromResult(new object()), - approvalChannel: approvalChannel, - emitApprovalRequest: request => approvalRequestTcs.TrySetResult(request.Request), - approvalTimeout: Timeout.InfiniteTimeSpan, - turnContext: InteractiveTurnContext(sessionId), - ct: TestContext.Current.CancellationToken); + var pipelineTask = new SessionToolPipelineTestFixture(executor, toolCalls, sessionId, probe.Ref) + .WithTurnContext(InteractiveTurnContext(sessionId)) + .WithTimeout(TimeSpan.FromSeconds(1)) + .WithApprovals( + approvalChannel, + request => approvalRequestTcs.TrySetResult(request.Request), + Timeout.InfiniteTimeSpan) + .ExecuteAsync(TestContext.Current.CancellationToken); var approvalRequest = await approvalRequestTcs.Task.WaitAsync( TimeSpan.FromSeconds(3), @@ -119,23 +152,14 @@ public async Task Source_less_approval_required_turn_fails_closed_without_prompt }) }; - var pipelineTask = SessionToolExecutionPipeline.ExecuteToolsAsync( - executor, - toolCalls, - new SessionId("D1/source-less-approval-test"), - source: null, - auditLogger: null, - timeProvider: TimeProvider.System, - sessionDir: Path.GetTempPath(), - maxInlineToolResultChars: 4096, - timeout: TimeSpan.FromSeconds(1), - self: probe.Ref, - emitSubAgentOutput: _ => { }, - spawnChildActor: static (_, _, _) => Task.FromResult(new object()), - approvalChannel: approvalChannel, - emitApprovalRequest: request => approvals.Add(request.Request), - approvalTimeout: Timeout.InfiniteTimeSpan, - ct: TestContext.Current.CancellationToken); + var pipelineTask = new SessionToolPipelineTestFixture( + executor, toolCalls, new SessionId("D1/source-less-approval-test"), probe.Ref) + .WithTimeout(TimeSpan.FromSeconds(1)) + .WithApprovals( + approvalChannel, + request => approvals.Add(request.Request), + Timeout.InfiniteTimeSpan) + .ExecuteAsync(TestContext.Current.CancellationToken); var completed = await probe.ExpectMsgAsync( TimeSpan.FromSeconds(3), @@ -168,23 +192,14 @@ public async Task Non_interactive_turn_does_not_create_subagent_approval_bridge( ReceivedAt = DateTimeOffset.UnixEpoch }; - var pipelineTask = SessionToolExecutionPipeline.ExecuteToolsAsync( - executor, - [new FunctionCallContent("call-1", "inspect_context")], - sessionId, - source, - auditLogger: null, - timeProvider: TimeProvider.System, - sessionDir: Path.GetTempPath(), - maxInlineToolResultChars: 4096, - timeout: TimeSpan.FromSeconds(1), - self: probe.Ref, - emitSubAgentOutput: _ => { }, - spawnChildActor: static (_, _, _) => Task.FromResult(new object()), - approvalChannel: new ApprovalChannel(), - emitApprovalRequest: _ => { }, - approvalTimeout: Timeout.InfiniteTimeSpan, - ct: TestContext.Current.CancellationToken); + var pipelineTask = new SessionToolPipelineTestFixture( + executor, + [new FunctionCallContent("call-1", "inspect_context")], + sessionId, + probe.Ref) + .From(source) + .WithTimeout(TimeSpan.FromSeconds(1)) + .ExecuteAsync(TestContext.Current.CancellationToken); await probe.ExpectMsgAsync( TimeSpan.FromSeconds(3), @@ -213,24 +228,13 @@ public async Task Approve_once_does_not_reprompt_on_retry_execution() }) }; - var pipelineTask = SessionToolExecutionPipeline.ExecuteToolsAsync( - executor, - toolCalls, - sessionId, - source: null, - auditLogger: null, - timeProvider: TimeProvider.System, - sessionDir: Path.GetTempPath(), - maxInlineToolResultChars: 4096, - timeout: TimeSpan.FromSeconds(5), - self: probe.Ref, - emitSubAgentOutput: _ => { }, - spawnChildActor: static (_, _, _) => Task.FromResult(new object()), - approvalChannel: approvalChannel, - emitApprovalRequest: request => approvals.Add(request.Request), - approvalTimeout: Timeout.InfiniteTimeSpan, - turnContext: InteractiveTurnContext(sessionId), - ct: TestContext.Current.CancellationToken); + var pipelineTask = new SessionToolPipelineTestFixture(executor, toolCalls, sessionId, probe.Ref) + .WithTurnContext(InteractiveTurnContext(sessionId)) + .WithApprovals( + approvalChannel, + request => approvals.Add(request.Request), + Timeout.InfiniteTimeSpan) + .ExecuteAsync(TestContext.Current.CancellationToken); await AwaitAssertAsync(() => { @@ -272,24 +276,13 @@ public async Task Approval_request_propagates_cwd_from_approval_context() }) }; - var pipelineTask = SessionToolExecutionPipeline.ExecuteToolsAsync( - executor, - toolCalls, - sessionId, - source: null, - auditLogger: null, - timeProvider: TimeProvider.System, - sessionDir: Path.GetTempPath(), - maxInlineToolResultChars: 4096, - timeout: TimeSpan.FromSeconds(5), - self: probe.Ref, - emitSubAgentOutput: _ => { }, - spawnChildActor: static (_, _, _) => Task.FromResult(new object()), - approvalChannel: approvalChannel, - emitApprovalRequest: request => approvalRequestTcs.TrySetResult(request.Request), - approvalTimeout: Timeout.InfiniteTimeSpan, - turnContext: InteractiveTurnContext(sessionId), - ct: TestContext.Current.CancellationToken); + var pipelineTask = new SessionToolPipelineTestFixture(executor, toolCalls, sessionId, probe.Ref) + .WithTurnContext(InteractiveTurnContext(sessionId)) + .WithApprovals( + approvalChannel, + request => approvalRequestTcs.TrySetResult(request.Request), + Timeout.InfiniteTimeSpan) + .ExecuteAsync(TestContext.Current.CancellationToken); var approvalRequest = await approvalRequestTcs.Task.WaitAsync( TimeSpan.FromSeconds(3), @@ -322,24 +315,13 @@ public async Task Approval_wait_is_cancelled_by_tool_execution_token() }) }; - var pipelineTask = SessionToolExecutionPipeline.ExecuteToolsAsync( - executor, - toolCalls, - sessionId, - source: null, - auditLogger: null, - timeProvider: TimeProvider.System, - sessionDir: Path.GetTempPath(), - maxInlineToolResultChars: 4096, - timeout: TimeSpan.FromSeconds(5), - self: probe.Ref, - emitSubAgentOutput: _ => { }, - spawnChildActor: static (_, _, _) => Task.FromResult(new object()), - approvalChannel: approvalChannel, - emitApprovalRequest: request => approvalRequestTcs.TrySetResult(request.Request), - approvalTimeout: Timeout.InfiniteTimeSpan, - turnContext: InteractiveTurnContext(sessionId), - ct: executionCts.Token); + var pipelineTask = new SessionToolPipelineTestFixture(executor, toolCalls, sessionId, probe.Ref) + .WithTurnContext(InteractiveTurnContext(sessionId)) + .WithApprovals( + approvalChannel, + request => approvalRequestTcs.TrySetResult(request.Request), + Timeout.InfiniteTimeSpan) + .ExecuteAsync(executionCts.Token); var approvalRequest = await approvalRequestTcs.Task.WaitAsync( TimeSpan.FromSeconds(3), @@ -367,20 +349,10 @@ public async Task A_stalled_tool_call_times_out_without_failing_its_healthy_sibl new("call-slow", "slow_tool", new Dictionary()) }; - var pipelineTask = SessionToolExecutionPipeline.ExecuteToolsAsync( - executor, - toolCalls, - new SessionId("D1/parallel-watchdog-test"), - source: null, - auditLogger: null, - timeProvider: TimeProvider.System, - sessionDir: Path.GetTempPath(), - maxInlineToolResultChars: 4096, - timeout: TimeSpan.FromSeconds(1), - self: probe.Ref, - emitSubAgentOutput: _ => { }, - spawnChildActor: static (_, _, _) => Task.FromResult(new object()), - ct: TestContext.Current.CancellationToken); + var pipelineTask = new SessionToolPipelineTestFixture( + executor, toolCalls, new SessionId("D1/parallel-watchdog-test"), probe.Ref) + .WithTimeout(TimeSpan.FromSeconds(1)) + .ExecuteAsync(TestContext.Current.CancellationToken); // Real-time: the slow tool's per-call budget token trips ~1s in (the 1s // wall-clock budget). The ceiling stays tight so a regression — a budget @@ -407,20 +379,12 @@ public async Task Opaque_tool_stream_without_a_completion_item_surfaces_an_error var executor = new ParallelStreamingExecutor(); var probe = CreateTestProbe("no-completion-probe"); - var pipelineTask = SessionToolExecutionPipeline.ExecuteToolsAsync( - executor, - [new FunctionCallContent("call-nc", "no_completion_tool", new Dictionary())], - new SessionId("D1/no-completion-test"), - source: null, - auditLogger: null, - timeProvider: TimeProvider.System, - sessionDir: Path.GetTempPath(), - maxInlineToolResultChars: 4096, - timeout: TimeSpan.FromSeconds(5), - self: probe.Ref, - emitSubAgentOutput: _ => { }, - spawnChildActor: static (_, _, _) => Task.FromResult(new object()), - ct: TestContext.Current.CancellationToken); + var pipelineTask = new SessionToolPipelineTestFixture( + executor, + [new FunctionCallContent("call-nc", "no_completion_tool", new Dictionary())], + new SessionId("D1/no-completion-test"), + probe.Ref) + .ExecuteAsync(TestContext.Current.CancellationToken); var completed = await probe.ExpectMsgAsync( TimeSpan.FromSeconds(3), @@ -441,20 +405,14 @@ public async Task Opaque_streaming_output_does_not_extend_tool_wall_clock_budget var executor = new ChattyOpaqueExecutor(); var probe = CreateTestProbe("opaque-wall-clock-probe"); - var pipelineTask = SessionToolExecutionPipeline.ExecuteToolsAsync( - executor, - [new FunctionCallContent("call-chatty", "chatty_tool", new Dictionary())], - new SessionId("D1/opaque-wall-clock-test"), - source: null, - auditLogger: null, - timeProvider: time, - sessionDir: Path.GetTempPath(), - maxInlineToolResultChars: 4096, - timeout: TimeSpan.FromSeconds(1), - self: probe.Ref, - emitSubAgentOutput: _ => { }, - spawnChildActor: static (_, _, _) => Task.FromResult(new object()), - ct: TestContext.Current.CancellationToken); + var pipelineTask = new SessionToolPipelineTestFixture( + executor, + [new FunctionCallContent("call-chatty", "chatty_tool", new Dictionary())], + new SessionId("D1/opaque-wall-clock-test"), + probe.Ref) + .WithTimeProvider(time) + .WithTimeout(TimeSpan.FromSeconds(1)) + .ExecuteAsync(TestContext.Current.CancellationToken); await executor.ActivitySeen.Task.WaitAsync(TimeSpan.FromSeconds(3), TestContext.Current.CancellationToken); time.Advance(TimeSpan.FromSeconds(2)); @@ -478,20 +436,13 @@ public async Task Self_monitoring_tool_runs_to_completion_without_a_parent_timeo var executor = new SelfMonitoringStreamingExecutor(); var probe = CreateTestProbe("self-monitoring-probe"); - var pipelineTask = SessionToolExecutionPipeline.ExecuteToolsAsync( - executor, - [new FunctionCallContent("call-self", "spawn_agent", new Dictionary())], - new SessionId("D1/self-monitoring-test"), - source: null, - auditLogger: null, - timeProvider: TimeProvider.System, - sessionDir: Path.GetTempPath(), - maxInlineToolResultChars: 4096, - timeout: TimeSpan.FromSeconds(1), - self: probe.Ref, - emitSubAgentOutput: _ => { }, - spawnChildActor: static (_, _, _) => Task.FromResult(new object()), - ct: TestContext.Current.CancellationToken); + var pipelineTask = new SessionToolPipelineTestFixture( + executor, + [new FunctionCallContent("call-self", "spawn_agent", new Dictionary())], + new SessionId("D1/self-monitoring-test"), + probe.Ref) + .WithTimeout(TimeSpan.FromSeconds(1)) + .ExecuteAsync(TestContext.Current.CancellationToken); await executor.Started.Task.WaitAsync(TimeSpan.FromSeconds(3), TestContext.Current.CancellationToken); // Nothing has completed it and there is no timer to trip, so it stays running. @@ -519,20 +470,13 @@ public async Task Self_monitoring_tool_is_bounded_only_by_caller_cancellation() var probe = CreateTestProbe("self-monitoring-cancel-probe"); using var cts = CancellationTokenSource.CreateLinkedTokenSource(TestContext.Current.CancellationToken); - var pipelineTask = SessionToolExecutionPipeline.ExecuteToolsAsync( - executor, - [new FunctionCallContent("call-self", "spawn_agent", new Dictionary())], - new SessionId("D1/self-monitoring-cancel-test"), - source: null, - auditLogger: null, - timeProvider: TimeProvider.System, - sessionDir: Path.GetTempPath(), - maxInlineToolResultChars: 4096, - timeout: TimeSpan.FromSeconds(1), - self: probe.Ref, - emitSubAgentOutput: _ => { }, - spawnChildActor: static (_, _, _) => Task.FromResult(new object()), - ct: cts.Token); + var pipelineTask = new SessionToolPipelineTestFixture( + executor, + [new FunctionCallContent("call-self", "spawn_agent", new Dictionary())], + new SessionId("D1/self-monitoring-cancel-test"), + probe.Ref) + .WithTimeout(TimeSpan.FromSeconds(1)) + .ExecuteAsync(cts.Token); await executor.Started.Task.WaitAsync(TimeSpan.FromSeconds(3), TestContext.Current.CancellationToken); Assert.False(pipelineTask.IsCompleted); // never completes on its own @@ -555,21 +499,15 @@ public async Task Tool_model_input_file_is_materialized_as_session_media_referen var executor = new ModelInputFileExecutor(imagePath); var probe = CreateTestProbe("model-input-file-probe"); - var pipelineTask = SessionToolExecutionPipeline.ExecuteToolsAsync( - executor, - [new FunctionCallContent("call-image", FileReadTool.ToolName, new Dictionary())], - new SessionId("D1/model-input-file-test"), - source: null, - auditLogger: null, - timeProvider: TimeProvider.System, - sessionDir: dir.Path, - maxInlineToolResultChars: 4096, - timeout: TimeSpan.FromSeconds(3), - self: probe.Ref, - emitSubAgentOutput: _ => { }, - spawnChildActor: static (_, _, _) => Task.FromResult(new object()), - modelInputModalities: ModelModality.Text | ModelModality.Image, - ct: TestContext.Current.CancellationToken); + var pipelineTask = new SessionToolPipelineTestFixture( + executor, + [new FunctionCallContent("call-image", FileReadTool.ToolName, new Dictionary())], + new SessionId("D1/model-input-file-test"), + probe.Ref) + .InSessionDirectory(dir.Path) + .WithTimeout(TimeSpan.FromSeconds(3)) + .AcceptingModelInput(ModelModality.Text | ModelModality.Image) + .ExecuteAsync(TestContext.Current.CancellationToken); var completed = await probe.ExpectMsgAsync( TimeSpan.FromSeconds(3), @@ -591,22 +529,16 @@ public async Task Streaming_tool_result_persists_model_input_media_references_on var executor = new ModelInputFileExecutor(imagePath); var probe = CreateTestProbe("streaming-model-input-probe"); - var pipelineTask = SessionToolExecutionPipeline.ExecuteToolsAsync( - executor, - [new FunctionCallContent("call-image", FileReadTool.ToolName, new Dictionary())], - new SessionId("D1/streaming-model-input-file-test"), - source: null, - auditLogger: null, - timeProvider: TimeProvider.System, - sessionDir: dir.Path, - maxInlineToolResultChars: 4096, - timeout: TimeSpan.FromSeconds(3), - self: probe.Ref, - emitSubAgentOutput: _ => { }, - spawnChildActor: static (_, _, _) => Task.FromResult(new object()), - streamToolResults: true, - modelInputModalities: ModelModality.Text | ModelModality.Image, - ct: TestContext.Current.CancellationToken); + var pipelineTask = new SessionToolPipelineTestFixture( + executor, + [new FunctionCallContent("call-image", FileReadTool.ToolName, new Dictionary())], + new SessionId("D1/streaming-model-input-file-test"), + probe.Ref) + .InSessionDirectory(dir.Path) + .WithTimeout(TimeSpan.FromSeconds(3)) + .StreamingResults() + .AcceptingModelInput(ModelModality.Text | ModelModality.Image) + .ExecuteAsync(TestContext.Current.CancellationToken); var single = await probe.ExpectMsgAsync( TimeSpan.FromSeconds(3), @@ -642,7 +574,7 @@ public async Task Tool_model_input_batch_limit_is_enforced_across_registered_fil var result = SessionToolExecutionPipeline.MaterializeModelInputFiles( context, dir.Path, - logger: null, + NoLogger.Instance, batchBudget: budget); Assert.Equal(2, result.RequestedCount); @@ -658,20 +590,14 @@ public async Task Tool_model_input_file_without_matching_modality_is_skipped() var executor = new ModelInputFileExecutor(imagePath); var probe = CreateTestProbe("model-input-modality-probe"); - var pipelineTask = SessionToolExecutionPipeline.ExecuteToolsAsync( - executor, - [new FunctionCallContent("call-image", FileReadTool.ToolName, new Dictionary())], - new SessionId("D1/model-input-modality-test"), - source: null, - auditLogger: null, - timeProvider: TimeProvider.System, - sessionDir: dir.Path, - maxInlineToolResultChars: 4096, - timeout: TimeSpan.FromSeconds(3), - self: probe.Ref, - emitSubAgentOutput: _ => { }, - spawnChildActor: static (_, _, _) => Task.FromResult(new object()), - ct: TestContext.Current.CancellationToken); + var pipelineTask = new SessionToolPipelineTestFixture( + executor, + [new FunctionCallContent("call-image", FileReadTool.ToolName, new Dictionary())], + new SessionId("D1/model-input-modality-test"), + probe.Ref) + .InSessionDirectory(dir.Path) + .WithTimeout(TimeSpan.FromSeconds(3)) + .ExecuteAsync(TestContext.Current.CancellationToken); var completed = await probe.ExpectMsgAsync( TimeSpan.FromSeconds(3), @@ -690,21 +616,15 @@ public async Task Tool_model_input_file_with_mismatched_magic_is_skipped() var executor = new ModelInputFileExecutor(imagePath); var probe = CreateTestProbe("model-input-magic-probe"); - var pipelineTask = SessionToolExecutionPipeline.ExecuteToolsAsync( - executor, - [new FunctionCallContent("call-image", FileReadTool.ToolName, new Dictionary())], - new SessionId("D1/model-input-magic-test"), - source: null, - auditLogger: null, - timeProvider: TimeProvider.System, - sessionDir: dir.Path, - maxInlineToolResultChars: 4096, - timeout: TimeSpan.FromSeconds(3), - self: probe.Ref, - emitSubAgentOutput: _ => { }, - spawnChildActor: static (_, _, _) => Task.FromResult(new object()), - modelInputModalities: ModelModality.Text | ModelModality.Image, - ct: TestContext.Current.CancellationToken); + var pipelineTask = new SessionToolPipelineTestFixture( + executor, + [new FunctionCallContent("call-image", FileReadTool.ToolName, new Dictionary())], + new SessionId("D1/model-input-magic-test"), + probe.Ref) + .InSessionDirectory(dir.Path) + .WithTimeout(TimeSpan.FromSeconds(3)) + .AcceptingModelInput(ModelModality.Text | ModelModality.Image) + .ExecuteAsync(TestContext.Current.CancellationToken); var completed = await probe.ExpectMsgAsync( TimeSpan.FromSeconds(3), @@ -726,21 +646,15 @@ public async Task Tool_model_input_file_over_size_limit_is_skipped() var executor = new ModelInputFileExecutor(imagePath); var probe = CreateTestProbe("model-input-size-probe"); - var pipelineTask = SessionToolExecutionPipeline.ExecuteToolsAsync( - executor, - [new FunctionCallContent("call-image", FileReadTool.ToolName, new Dictionary())], - new SessionId("D1/model-input-size-test"), - source: null, - auditLogger: null, - timeProvider: TimeProvider.System, - sessionDir: dir.Path, - maxInlineToolResultChars: 4096, - timeout: TimeSpan.FromSeconds(3), - self: probe.Ref, - emitSubAgentOutput: _ => { }, - spawnChildActor: static (_, _, _) => Task.FromResult(new object()), - modelInputModalities: ModelModality.Text | ModelModality.Image, - ct: TestContext.Current.CancellationToken); + var pipelineTask = new SessionToolPipelineTestFixture( + executor, + [new FunctionCallContent("call-image", FileReadTool.ToolName, new Dictionary())], + new SessionId("D1/model-input-size-test"), + probe.Ref) + .InSessionDirectory(dir.Path) + .WithTimeout(TimeSpan.FromSeconds(3)) + .AcceptingModelInput(ModelModality.Text | ModelModality.Image) + .ExecuteAsync(TestContext.Current.CancellationToken); var completed = await probe.ExpectMsgAsync( TimeSpan.FromSeconds(3), diff --git a/src/Netclaw.Actors/Sessions/LlmSessionActor.cs b/src/Netclaw.Actors/Sessions/LlmSessionActor.cs index 16d2739dc..bde08b56d 100644 --- a/src/Netclaw.Actors/Sessions/LlmSessionActor.cs +++ b/src/Netclaw.Actors/Sessions/LlmSessionActor.cs @@ -57,8 +57,8 @@ public sealed class LlmSessionActor : ReceivePersistentActor, IWithTimers private readonly IReadOnlyList _contextLayers; private readonly IWorkingContextSnapshotProvider _workingContextSnapshots; private readonly IToolExecutor? _toolExecutor; + private readonly SessionToolExecutionPipeline? _toolExecutionPipeline; private readonly Tools.ToolRegistry? _toolRegistry; - private readonly IToolAuditLogger? _auditLogger; private readonly IToolApprovalService? _approvalService; private readonly ApprovalChannel _approvalChannel = new(); private readonly IMemoryExtractor _memoryExtractor; @@ -234,7 +234,6 @@ public LlmSessionActor( _subAgentLoader = tools?.SubAgentLoader; _toolExecutor = tools?.ToolExecutor; _toolRegistry = tools?.ToolRegistry; - _auditLogger = tools?.AuditLogger; _toolAccessPolicy = tools?.AccessPolicy; _approvalService = tools?.ApprovalService; _memoryExtractor = memory?.MemoryExtractor ?? NullMemoryExtractor.Instance; @@ -249,6 +248,13 @@ public LlmSessionActor( // Enrich logger with session context — all log messages automatically include SessionId _log = Context.GetLogger().WithContext(NetclawLogProperties.SessionId, _sessionId.Value); + _toolExecutionPipeline = tools is null + ? null + : new SessionToolExecutionPipeline( + tools.ToolExecutor, + tools.AuditLogger, + services.TimeProvider, + NoLogger.Instance); // Load all non-MCP tools for initial LLM calls. // MCP tools are loaded dynamically via search_tools and can be retained for a @@ -1958,12 +1964,7 @@ private void DispatchToolBatch( tc.Arguments is not null ? JsonSerializer.Serialize(tc.Arguments) : "{}"); } var self = Self; - var executor = _toolExecutor!; - var sessionId = _sessionId; - var auditLogger = _auditLogger; - var tp = _timeProvider; var sessionDir = GetSessionDirectory(); - var maxInlineToolResultChars = _config.Tuning.MaxInlineToolResultChars; // Per-call inactivity watchdogs in the tool-execution pipeline govern // tool liveness; the session ProcessingWatchdog covers only LLM calls // and compaction, so no batch tool-execution watchdog is armed here. @@ -1988,10 +1989,10 @@ await self.Ask( timeout: toolExecutionTimeout, cancellationToken: ct); - IActorRef? bgJobManager = null; var registry = ActorRegistry.For(Context.System); - if (registry.TryGet(out var mgr)) - bgJobManager = mgr; + var backgroundJobs = registry.TryGet(out var manager) + ? new BackgroundJobDispatch.Available(manager) + : (BackgroundJobDispatch)new BackgroundJobDispatch.Unavailable(); // Pre-compute set_working_directory exposure once per dispatch so the // pipeline's deny-path hint logic can run without a policy lookup. @@ -2002,21 +2003,40 @@ await self.Ask( CancelAndDisposeToolExecutionCts(); _activeToolExecutionCts = new CancellationTokenSource(); var toolExecutionCt = _activeToolExecutionCts.Token; + var turnContext = _currentTurnContext + ?? throw new InvalidOperationException("Tool batch dispatch requires admitted turn authority."); + var runEnvironment = new SessionToolRunEnvironment + { + SessionDirectory = sessionDir, + InlineOutputBudget = new InlineOutputBudget(_config.Tuning.MaxInlineToolResultChars), + ModelInputModalities = _model.InputModalities, + SpawnChildActor = spawnChildActor, + ProjectDirectory = _state.WorkingContext.ProjectDirectory, + RecentFiles = _state.WorkingContext.RecentFiles + }; + var pipeline = _toolExecutionPipeline + ?? throw new InvalidOperationException("Tool batch dispatch requires tool execution infrastructure."); + var batch = new SessionToolBatch(turnContext, runEnvironment) + { + ToolCalls = toolCalls, + DefaultTimeout = new ToolExecutionTimeout(toolExecutionTimeout), + ReplyTo = self, + EmitSubAgentOutput = emitSubAgentOutput, + ApprovalRequests = new ToolApprovalRequests( + _approvalChannel, + request => self.Tell(request), + new ToolExecutionTimeout(Timeout.InfiniteTimeSpan)), + BackgroundJobs = backgroundJobs, + SetWorkingDirectoryAvailable = setWorkingDirectoryAvailable, + StreamResults = true, + OneTimeApprovalPreSeed = oneTimeApprovalPreSeed + ?? new Dictionary>(), + DecisionOverrides = decisionOverride + ?? new Dictionary(), + CancellationToken = toolExecutionCt + }; - _ = SessionToolExecutionPipeline.ExecuteToolsAsync(executor, toolCalls, sessionId, _currentTurnSource, auditLogger, tp, sessionDir, maxInlineToolResultChars, toolExecutionTimeout, self, emitSubAgentOutput, spawnChildActor, - approvalChannel: _approvalChannel, - emitApprovalRequest: request => self.Tell(request), - approvalTimeout: Timeout.InfiniteTimeSpan, - backgroundJobManager: bgJobManager, - projectDirectory: _state.WorkingContext.ProjectDirectory, - recentFiles: _state.WorkingContext.RecentFiles, - setWorkingDirectoryAvailable: setWorkingDirectoryAvailable, - streamToolResults: true, - modelInputModalities: _model.InputModalities, - oneTimeApprovalPreSeed: oneTimeApprovalPreSeed, - decisionOverride: decisionOverride, - turnContext: _currentTurnContext, - ct: toolExecutionCt); + _ = pipeline.ExecuteAsync(batch); } private void HandleTextResponse( diff --git a/src/Netclaw.Actors/Sessions/Pipelines/SessionToolExecutionPipeline.cs b/src/Netclaw.Actors/Sessions/Pipelines/SessionToolExecutionPipeline.cs index 94351cc2a..3f41b52d6 100644 --- a/src/Netclaw.Actors/Sessions/Pipelines/SessionToolExecutionPipeline.cs +++ b/src/Netclaw.Actors/Sessions/Pipelines/SessionToolExecutionPipeline.cs @@ -4,9 +4,10 @@ // // ----------------------------------------------------------------------- using System.Diagnostics; +using System.Collections.Frozen; using Akka.Actor; +using Akka.Event; using Microsoft.Extensions.AI; -using Microsoft.Extensions.Logging; using Netclaw.Actors.Channels; using Netclaw.Actors.Jobs; using Netclaw.Actors.Protocol; @@ -64,98 +65,233 @@ public void Release(long sizeBytes) } } +internal sealed class ToolApprovalRequests +{ + public ToolApprovalRequests( + IApprovalChannel channel, + Action emitRequest, + ToolExecutionTimeout timeout) + { + ArgumentNullException.ThrowIfNull(channel); + ArgumentNullException.ThrowIfNull(emitRequest); + ArgumentNullException.ThrowIfNull(timeout); + + Channel = channel; + EmitRequest = emitRequest; + Timeout = timeout; + } + + public IApprovalChannel Channel { get; } + public Action EmitRequest { get; } + public ToolExecutionTimeout Timeout { get; } +} + +internal abstract record BackgroundJobDispatch +{ + private BackgroundJobDispatch() + { + } + + public sealed record Unavailable : BackgroundJobDispatch; + + public sealed record Available : BackgroundJobDispatch + { + public Available(IActorRef manager) + { + ArgumentNullException.ThrowIfNull(manager); + Manager = manager; + } + + public IActorRef Manager { get; } + } +} + +internal sealed class SessionToolRunEnvironment +{ + private IReadOnlyList _recentFiles = []; + + public required string SessionDirectory { get; init; } + public required InlineOutputBudget InlineOutputBudget { get; init; } + public required Func> SpawnChildActor { get; init; } + public ModelModality ModelInputModalities { get; init; } = ModelModality.Text; + public string? ProjectDirectory { get; init; } + public IReadOnlyList RecentFiles + { + get => _recentFiles; + init + { + ArgumentNullException.ThrowIfNull(value); + _recentFiles = Array.AsReadOnly(value.ToArray()); + } + } +} + +internal sealed class SessionToolBatch +{ + private static readonly IReadOnlyDictionary> NoApprovalPreSeed + = new Dictionary>().ToFrozenDictionary(); + private static readonly IReadOnlyDictionary NoDecisionOverrides + = new Dictionary().ToFrozenDictionary(); + private IReadOnlyList _toolCalls = []; + private IReadOnlyDictionary> _oneTimeApprovalPreSeed = NoApprovalPreSeed; + private IReadOnlyDictionary _decisionOverrides = NoDecisionOverrides; + + public SessionToolBatch(TurnContext turnContext, SessionToolRunEnvironment environment) + { + ArgumentNullException.ThrowIfNull(turnContext); + ArgumentNullException.ThrowIfNull(environment); + ArgumentException.ThrowIfNullOrWhiteSpace(environment.SessionDirectory); + ArgumentNullException.ThrowIfNull(environment.InlineOutputBudget); + ArgumentNullException.ThrowIfNull(environment.SpawnChildActor); + + TurnContext = turnContext; + RunScope = new ToolRunScope + { + Session = new ToolSessionScope.Bound(turnContext.SessionId.Value, environment.SessionDirectory), + Audience = turnContext.Audience, + Boundary = turnContext.Boundary, + ChannelType = turnContext.ChannelType?.ToWireValue(), + DefaultDeliveryTarget = turnContext.DefaultDeliveryTarget, + RequestedDeliveryTarget = turnContext.RequestedDeliveryTarget, + SupportsInteractiveApproval = turnContext.SupportsInteractiveApproval, + InlineOutputBudget = environment.InlineOutputBudget, + ModelInputModalities = environment.ModelInputModalities, + SpawnChildActor = environment.SpawnChildActor, + ProjectDirectory = environment.ProjectDirectory, + RecentFiles = environment.RecentFiles + }; + } + + public required IReadOnlyList ToolCalls + { + get => _toolCalls; + init + { + ArgumentNullException.ThrowIfNull(value); + _toolCalls = Array.AsReadOnly(value.ToArray()); + } + } + public TurnContext TurnContext { get; } + public ToolRunScope RunScope { get; } + public required ToolExecutionTimeout DefaultTimeout { get; init; } + public required IActorRef ReplyTo { get; init; } + public required Action EmitSubAgentOutput { get; init; } + public required ToolApprovalRequests ApprovalRequests { get; init; } + public required BackgroundJobDispatch BackgroundJobs { get; init; } + public bool SetWorkingDirectoryAvailable { get; init; } + public bool StreamResults { get; init; } + public IReadOnlyDictionary> OneTimeApprovalPreSeed + { + get => _oneTimeApprovalPreSeed; + init + { + ArgumentNullException.ThrowIfNull(value); + _oneTimeApprovalPreSeed = value.ToFrozenDictionary( + entry => entry.Key, + entry => (IReadOnlyList)Array.AsReadOnly(entry.Value.ToArray()), + StringComparer.Ordinal); + } + } + public IReadOnlyDictionary DecisionOverrides + { + get => _decisionOverrides; + init + { + ArgumentNullException.ThrowIfNull(value); + _decisionOverrides = value.ToFrozenDictionary(StringComparer.Ordinal); + } + } + public CancellationToken CancellationToken { get; init; } + + public SessionId SessionId => TurnContext.SessionId; + + public string SessionDirectory => RunScope.Session is ToolSessionScope.Bound bound + && !string.IsNullOrWhiteSpace(bound.SessionDirectory) + ? bound.SessionDirectory + : throw new InvalidOperationException("Session tool batches require a bound session directory."); + + public void Validate() + { + ArgumentNullException.ThrowIfNull(ToolCalls); + ArgumentNullException.ThrowIfNull(DefaultTimeout); + ArgumentNullException.ThrowIfNull(ReplyTo); + ArgumentNullException.ThrowIfNull(EmitSubAgentOutput); + ArgumentNullException.ThrowIfNull(ApprovalRequests); + ArgumentNullException.ThrowIfNull(BackgroundJobs); + ArgumentNullException.ThrowIfNull(OneTimeApprovalPreSeed); + ArgumentNullException.ThrowIfNull(DecisionOverrides); + } +} + /// /// Async pipeline for parallel tool execution. Runs on the thread pool and /// sends results back to the session actor via self.Tell(). /// -internal static class SessionToolExecutionPipeline +internal sealed class SessionToolExecutionPipeline { private const long MaxModelInputFileBytes = ChannelAttachmentPolicy.DefaultMaxFileBytes; private const long MaxModelInputBatchBytes = ChannelAttachmentPolicy.DefaultMaxFileBytes; - public static async Task ExecuteToolsAsync( + private readonly IToolExecutor _executor; + private readonly IToolAuditLogger _auditLogger; + private readonly TimeProvider _timeProvider; + private readonly ILoggingAdapter _logger; + + public SessionToolExecutionPipeline( IToolExecutor executor, - List toolCalls, - SessionId sessionId, - MessageSource? source, - IToolAuditLogger? auditLogger, + IToolAuditLogger auditLogger, TimeProvider timeProvider, - string sessionDir, - int maxInlineToolResultChars, - TimeSpan timeout, - IActorRef self, - Action emitSubAgentOutput, - Func> spawnChildActor, - IApprovalChannel? approvalChannel = null, - Action? emitApprovalRequest = null, - TimeSpan? approvalTimeout = null, - ILogger? logger = null, - IActorRef? backgroundJobManager = null, - string? projectDirectory = null, - IReadOnlyList? recentFiles = null, - bool setWorkingDirectoryAvailable = false, - bool streamToolResults = false, - ModelModality modelInputModalities = ModelModality.Text, - IReadOnlyDictionary>? oneTimeApprovalPreSeed = null, - IReadOnlyDictionary? decisionOverride = null, - TurnContext? turnContext = null, - CancellationToken ct = default) + ILoggingAdapter logger) + { + ArgumentNullException.ThrowIfNull(executor); + ArgumentNullException.ThrowIfNull(auditLogger); + ArgumentNullException.ThrowIfNull(timeProvider); + ArgumentNullException.ThrowIfNull(logger); + + _executor = executor; + _auditLogger = auditLogger; + _timeProvider = timeProvider; + _logger = logger; + } + + public async Task ExecuteAsync(SessionToolBatch batch) { try { + batch.Validate(); + var timeout = batch.DefaultTimeout.Value; // Execute all tool calls in parallel. Calls are not always // independent -- e.g. two file_edit calls on the same file -- so // file-mutating tools serialize their read-modify-write per target // path via FileMutationGate to avoid lost-update races here. var modelInputBudget = new ModelInputBatchBudget(MaxModelInputBatchBytes); - var tasks = toolCalls.Select(async tc => + var tasks = batch.ToolCalls.Select(async tc => { var result = await ExecuteSingleToolAsync( - executor, tc, - sessionId, - source, - auditLogger, - timeProvider, - sessionDir, - maxInlineToolResultChars, - emitSubAgentOutput, - spawnChildActor, - timeout, - ct, - approvalChannel, - emitApprovalRequest, - approvalTimeout ?? Timeout.InfiniteTimeSpan, - logger, - backgroundJobManager, - projectDirectory, - recentFiles, - setWorkingDirectoryAvailable, - modelInputModalities, - oneTimeApprovalPreSeed is not null - && oneTimeApprovalPreSeed.TryGetValue(tc.CallId, out var preSeedPatterns) + batch, + batch.OneTimeApprovalPreSeed.TryGetValue(tc.CallId, out var preSeedPatterns) ? preSeedPatterns : null, - decisionOverride is not null && decisionOverride.TryGetValue(tc.CallId, out var overrideDecision) + batch.DecisionOverrides.TryGetValue(tc.CallId, out var overrideDecision) ? overrideDecision : null, - turnContext, modelInputBudget); - if (streamToolResults) - self.Tell(new ToolExecutionSingleCompleted(result)); + if (batch.StreamResults) + batch.ReplyTo.Tell(new ToolExecutionSingleCompleted(result)); return result; }); var results = await Task.WhenAll(tasks); - if (streamToolResults) + if (batch.StreamResults) { - self.Tell(new ToolExecutionBatchCompleted()); + batch.ReplyTo.Tell(new ToolExecutionBatchCompleted()); return; } var fileAttachments = results.SelectMany(r => r.FileAttachments).ToList(); var modelInputMediaReferences = results.SelectMany(r => r.ModelInputMediaReferences).ToList(); - self.Tell(new ToolExecutionCompleted + batch.ReplyTo.Tell(new ToolExecutionCompleted { ToolResults = [.. results.Select(r => r.Message)], ModelInputMediaReferences = modelInputMediaReferences, @@ -167,52 +303,32 @@ oneTimeApprovalPreSeed is not null } catch (TimeoutException ex) { - self.Tell(new ToolExecutionFailed { Cause = ex }); + batch.ReplyTo.Tell(new ToolExecutionFailed { Cause = ex }); } catch (OperationCanceledException ex) { // The tool-execution token is cancelled both by caller (turn/user) supersede // and by the session's own timeout watchdog; surface either as a failed // batch (the watchdog message is the authoritative one). - self.Tell(new ToolExecutionFailed + batch.ReplyTo.Tell(new ToolExecutionFailed { Cause = new TimeoutException( - $"Tool execution exceeded timeout of {timeout.TotalSeconds:F0}s", + $"Tool execution exceeded timeout of {batch.DefaultTimeout.Value.TotalSeconds:F0}s", ex) }); } catch (Exception ex) { - self.Tell(new ToolExecutionFailed { Cause = ex }); + batch.ReplyTo.Tell(new ToolExecutionFailed { Cause = ex }); } } - public static async Task ExecuteSingleToolAsync( - IToolExecutor executor, + private async Task ExecuteSingleToolAsync( FunctionCallContent tc, - SessionId sessionId, - MessageSource? source, - IToolAuditLogger? auditLogger, - TimeProvider timeProvider, - string sessionDir, - int maxInlineToolResultChars, - Action emitSubAgentOutput, - Func> spawnChildActor, - TimeSpan timeout, - CancellationToken ct, - IApprovalChannel? approvalChannel = null, - Action? emitApprovalRequest = null, - TimeSpan? approvalTimeout = null, - ILogger? logger = null, - IActorRef? backgroundJobManager = null, - string? projectDirectory = null, - IReadOnlyList? recentFiles = null, - bool setWorkingDirectoryAvailable = false, - ModelModality modelInputModalities = ModelModality.Text, - IReadOnlyList? oneTimeApprovalPreSeed = null, - ApprovalDecision? decisionOverride = null, - TurnContext? turnContext = null, - ModelInputBatchBudget? modelInputBudget = null) + SessionToolBatch batch, + IReadOnlyList? oneTimeApprovalPreSeed, + ApprovalDecision? decisionOverride, + ModelInputBatchBudget modelInputBudget) { // Single execution-preflight seam, shared with the sub-agent path via // IToolExecutor.InterpretToolCall: validate the ORIGINAL arguments (parse @@ -220,10 +336,10 @@ public static async Task ExecuteSingleToolAsync( // success, extract meta + strip meta keys. Rejecting here — rather than // letting ExecuteAsync return the rejection string — is what lets the denial // be audited as Allowed=false instead of being misreported as executed. - var interpretation = executor.InterpretToolCall(tc); + var interpretation = _executor.InterpretToolCall(tc); if (interpretation.Rejection is { } rejection) { - auditLogger?.Log(BuildAuditEntry(sessionId, tc, timeProvider, TimeSpan.Zero, meta: null) with + _auditLogger.Log(BuildAuditEntry(batch.SessionId, tc, _timeProvider, TimeSpan.Zero, meta: null) with { Allowed = false, DenyReason = rejection.DenyReason @@ -244,37 +360,22 @@ public static async Task ExecuteSingleToolAsync( // The agent's per-call timeout hint is honored as requested; when absent // the inherited default (SessionConfig.ToolExecutionTimeout) applies. // ExtractFrom only yields a positive hint, so there is nothing to clamp. + var timeout = batch.DefaultTimeout.Value; if (meta?.TimeoutHintSeconds is { } hintSeconds) timeout = TimeSpan.FromSeconds(hintSeconds); var sw = Stopwatch.StartNew(); string resultText; - IParentApprovalBridge? approvalBridge = null; - if (approvalChannel is not null - && emitApprovalRequest is not null - && CanRequestInteractiveApproval(source, turnContext)) - { - approvalBridge = new ParentSessionApprovalBridge( - approvalChannel, - emitApprovalRequest, - sessionId, - tc.CallId, - turnContext?.RequesterSenderId ?? source?.SenderId, - turnContext?.RequesterPrincipal ?? source?.Principal, - turnContext?.HasAdoptedContext ?? source?.HasAdoptedContext ?? false, - turnContext?.HasThirdPartyAdoptedContext ?? source?.HasThirdPartyAdoptedContext ?? false, - turnContext?.AdoptedSpeakerIds ?? source?.AdoptedSpeakerIds ?? []); - } var completedRuns = new List(); var acceptedFindings = new List(); var outputs = new ToolExecutionOutputs(info => { if (info.IsStarted) { - emitSubAgentOutput(new SubAgentOutput + batch.EmitSubAgentOutput(new SubAgentOutput { - SessionId = sessionId, - TimestampMs = timeProvider.GetUtcNow().ToUnixTimeMilliseconds(), + SessionId = batch.SessionId, + TimestampMs = _timeProvider.GetUtcNow().ToUnixTimeMilliseconds(), AgentName = new SubAgents.AgentName(info.AgentName), Phase = Netclaw.Actors.SubAgents.SubAgentPhase.Started, ToolCount = info.ToolCount, @@ -290,7 +391,7 @@ public static async Task ExecuteSingleToolAsync( if (info.Success && info.Findings.Count == 1) { - var singleDecision = ReviewSubAgentFinding(info.Findings[0], sessionId); + var singleDecision = ReviewSubAgentFinding(info.Findings[0], batch.SessionId); decision = singleDecision.Decision.ToWireValue(); reason = singleDecision.Reason; } @@ -314,7 +415,7 @@ public static async Task ExecuteSingleToolAsync( { foreach (var finding in info.Findings) { - var findingDecision = ReviewSubAgentFinding(finding, sessionId); + var findingDecision = ReviewSubAgentFinding(finding, batch.SessionId); acceptedFindings.Add(new AcceptedSubAgentFinding { RunId = info.RunId, @@ -338,18 +439,22 @@ public static async Task ExecuteSingleToolAsync( } } }); - var context = BuildToolExecutionContext( - sessionId, - source, - sessionDir, - spawnChildActor, - approvalBridge, - projectDirectory, - recentFiles, - turnContext, - modelInputModalities, - maxInlineToolResultChars, - timeout, + var approvalBridge = CanRequestInteractiveApproval(batch.TurnContext) + ? new ParentSessionApprovalBridge( + batch.ApprovalRequests.Channel, + batch.ApprovalRequests.EmitRequest, + batch.SessionId, + tc.CallId, + batch.TurnContext.RequesterSenderId, + batch.TurnContext.RequesterPrincipal, + batch.TurnContext.HasAdoptedContext, + batch.TurnContext.HasThirdPartyAdoptedContext, + batch.TurnContext.AdoptedSpeakerIds) + : null; + var callScope = batch.RunScope with { ApprovalBridge = approvalBridge }; + var context = new ToolExecutionContext( + callScope, + new ToolExecutionTimeout(timeout), outputs); // Re-drive of an ApprovedOnce approval: the user already clicked @@ -371,7 +476,7 @@ public static async Task ExecuteSingleToolAsync( ? "Tool access denied: approval_timed_out" : $"Tool access denied: approval_denied_by_user ({tc.Name} requires interactive approval and the user declined it)"; - auditLogger?.Log(BuildAuditEntry(sessionId, tc, timeProvider, sw.Elapsed, meta) with + _auditLogger.Log(BuildAuditEntry(batch.SessionId, tc, _timeProvider, sw.Elapsed, meta) with { Allowed = false, DenyReason = resultText, @@ -393,41 +498,41 @@ public static async Task ExecuteSingleToolAsync( { if (!string.Equals(tc.Name, Tools.ShellTool.ToolName, StringComparison.Ordinal)) { - logger?.LogWarning( + _logger.Warning( "Tool {ToolName} (call {CallId}) requested background execution — " + "only shell_execute supports background mode; executing synchronously", tc.Name, tc.CallId); } - else if (backgroundJobManager is null) + else if (batch.BackgroundJobs is BackgroundJobDispatch.Unavailable) { - logger?.LogWarning( + _logger.Warning( "Tool {ToolName} (call {CallId}) requested background execution — " + "no background job manager available; executing synchronously", tc.Name, tc.CallId); } - else + else if (batch.BackgroundJobs is BackgroundJobDispatch.Available backgroundJobs) { - await executor.AuthorizeAsync(tc, context, ct); + await _executor.AuthorizeAsync(tc, context, batch.CancellationToken); sw.Stop(); return await RouteToBackgroundJobAsync( - tc, sessionId, source, auditLogger, timeProvider, - turnContext, - meta, backgroundJobManager, + tc, batch, + meta, backgroundJobs.Manager, // Honor the agent's requested timeout; when absent, no // kill timer is armed — a background job is a detached // process with no completion expectation, reaped by its // own exit, cancellation, or session passivation. meta.TimeoutHintSeconds ?? 0, - sw.Elapsed, logger, + sw.Elapsed, context.Approval.AppliedDecision, context.Approval.AppliedPattern); } } - resultText = await ExecuteToolAttemptAsync(executor, tc, context, timeout, timeProvider, ct); + resultText = await ExecuteToolAttemptAsync( + _executor, tc, context, timeout, _timeProvider, batch.CancellationToken); sw.Stop(); - auditLogger?.Log(BuildAuditEntry(sessionId, tc, timeProvider, sw.Elapsed, meta) with + _auditLogger.Log(BuildAuditEntry(batch.SessionId, tc, _timeProvider, sw.Elapsed, meta) with { Allowed = true, ApprovalDecision = context.Approval.AppliedDecision, @@ -435,14 +540,13 @@ public static async Task ExecuteSingleToolAsync( }); } catch (ToolApprovalRequiredException approvalEx) - when (approvalChannel is not null && emitApprovalRequest is not null) { - if (!CanRequestInteractiveApproval(source, turnContext)) + if (!CanRequestInteractiveApproval(batch.TurnContext)) { sw.Stop(); resultText = $"Tool requires approval but no interactive approval requester is available: {approvalEx.ApprovalContext.ToolName}"; - auditLogger?.Log(BuildAuditEntry(sessionId, tc, timeProvider, sw.Elapsed, meta) with + _auditLogger.Log(BuildAuditEntry(batch.SessionId, tc, _timeProvider, sw.Elapsed, meta) with { Allowed = false, DenyReason = "interactive_approval_unavailable" @@ -459,25 +563,24 @@ public static async Task ExecuteSingleToolAsync( // Mid-turn approval pause: emit request to channel, block on TCS var ctx = approvalEx.ApprovalContext; - var approvalWaitTimeout = approvalTimeout ?? Timeout.InfiniteTimeSpan; - var waitTask = approvalChannel.WaitForApprovalAsync( + var waitTask = batch.ApprovalRequests.Channel.WaitForApprovalAsync( new ToolCallId(tc.CallId), - approvalWaitTimeout, - ct); + batch.ApprovalRequests.Timeout.Value, + batch.CancellationToken); - emitApprovalRequest(new ToolInteractionRequestDispatch(new ToolInteractionRequest + batch.ApprovalRequests.EmitRequest(new ToolInteractionRequestDispatch(new ToolInteractionRequest { - SessionId = sessionId, + SessionId = batch.SessionId, Kind = "approval", CallId = new ToolCallId(tc.CallId), ToolName = new ToolName(ctx.ToolName), DisplayText = ctx.DisplayText, - RequesterSenderId = turnContext?.RequesterSenderId ?? source?.SenderId, - RequesterPrincipal = turnContext?.RequesterPrincipal ?? source?.Principal, - HasAdoptedContext = turnContext?.HasAdoptedContext ?? source?.HasAdoptedContext ?? false, - HasThirdPartyAdoptedContext = turnContext?.HasThirdPartyAdoptedContext ?? source?.HasThirdPartyAdoptedContext ?? false, - AdoptedSpeakerIds = turnContext?.AdoptedSpeakerIds ?? source?.AdoptedSpeakerIds ?? [], - PersistedAdoptedContext = turnContext?.HasAdoptedContext ?? source?.HasAdoptedContext ?? false, + RequesterSenderId = batch.TurnContext.RequesterSenderId, + RequesterPrincipal = batch.TurnContext.RequesterPrincipal, + HasAdoptedContext = batch.TurnContext.HasAdoptedContext, + HasThirdPartyAdoptedContext = batch.TurnContext.HasThirdPartyAdoptedContext, + AdoptedSpeakerIds = batch.TurnContext.AdoptedSpeakerIds, + PersistedAdoptedContext = batch.TurnContext.HasAdoptedContext, Patterns = ctx.Patterns, CandidateVerbs = ctx.CandidateVerbs, Candidates = ctx.Candidates ?? [], @@ -508,29 +611,29 @@ or ApprovalDecision.ApprovedAlways sw = Stopwatch.StartNew(); if (meta is { Background: true } && string.Equals(tc.Name, Tools.ShellTool.ToolName, StringComparison.Ordinal) - && backgroundJobManager is not null) + && batch.BackgroundJobs is BackgroundJobDispatch.Available backgroundJobs) { - await executor.AuthorizeAsync(tc, context, ct); + await _executor.AuthorizeAsync(tc, context, batch.CancellationToken); sw.Stop(); return await RouteToBackgroundJobAsync( - tc, sessionId, source, auditLogger, timeProvider, - turnContext, - meta, backgroundJobManager, + tc, batch, + meta, backgroundJobs.Manager, // Honor the agent's requested timeout; when absent, no // kill timer is armed — a background job is a detached // process with no completion expectation, reaped by its // own exit, cancellation, or session passivation. meta.TimeoutHintSeconds ?? 0, - sw.Elapsed, logger, + sw.Elapsed, decision.ToString(), string.Join(", ", ctx.Patterns)); } - resultText = await ExecuteToolAttemptAsync(executor, tc, context, timeout, timeProvider, ct); + resultText = await ExecuteToolAttemptAsync( + _executor, tc, context, timeout, _timeProvider, batch.CancellationToken); sw.Stop(); var patternStr = string.Join(", ", ctx.Patterns); - auditLogger?.Log(BuildAuditEntry(sessionId, tc, timeProvider, sw.Elapsed, meta) with + _auditLogger.Log(BuildAuditEntry(batch.SessionId, tc, _timeProvider, sw.Elapsed, meta) with { Allowed = true, ApprovalDecision = decision.ToString(), @@ -555,14 +658,14 @@ or ApprovalDecision.ApprovedAlways cwd: context.Approval.Cwd, sessionDirectory: context.SessionDirectory, projectDirectory: context.ProjectDirectory, - setWorkingDirectoryAvailable: setWorkingDirectoryAvailable); + setWorkingDirectoryAvailable: batch.SetWorkingDirectoryAvailable); resultText = string.IsNullOrEmpty(hint) ? reason : $"{reason}\n{hint}"; // Denied audit entries should describe the exact blocked units // the user saw in the prompt. Broader reusable approval entries // are only relevant when B/C is granted. var deniedPatternStr = string.Join(", ", ctx.Patterns); - auditLogger?.Log(BuildAuditEntry(sessionId, tc, timeProvider, sw.Elapsed, meta) with + _auditLogger.Log(BuildAuditEntry(batch.SessionId, tc, _timeProvider, sw.Elapsed, meta) with { Allowed = false, DenyReason = reason, @@ -571,30 +674,18 @@ or ApprovalDecision.ApprovedAlways }); } } - catch (ToolApprovalRequiredException approvalEx) - { - // No approval channel available — treat as denied - sw.Stop(); - resultText = $"Tool requires approval but no approval channel is available: {approvalEx.ApprovalContext.ToolName}"; - - auditLogger?.Log(BuildAuditEntry(sessionId, tc, timeProvider, sw.Elapsed, meta) with - { - Allowed = false, - DenyReason = "no_approval_channel" - }); - } catch (ToolAccessDeniedException ex) { sw.Stop(); resultText = $"Tool access denied: {ex.DenyReason}"; - auditLogger?.Log(BuildAuditEntry(sessionId, tc, timeProvider, sw.Elapsed, meta) with + _auditLogger.Log(BuildAuditEntry(batch.SessionId, tc, _timeProvider, sw.Elapsed, meta) with { Allowed = false, DenyReason = ex.DenyReason }); } - catch (OperationCanceledException) when (ct.IsCancellationRequested) + catch (OperationCanceledException) when (batch.CancellationToken.IsCancellationRequested) { // Caller (turn/user) cancellation is not a tool failure. Self-monitoring // tools are bounded only by ct, so this is the normal cancel path; let it @@ -607,15 +698,15 @@ or ApprovalDecision.ApprovedAlways sw.Stop(); resultText = $"Error executing tool: {ex.Message}"; - auditLogger?.Log(BuildAuditEntry(sessionId, tc, timeProvider, sw.Elapsed, meta) with + _auditLogger.Log(BuildAuditEntry(batch.SessionId, tc, _timeProvider, sw.Elapsed, meta) with { Allowed = false, DenyReason = $"tool_execution_error:{ex.GetType().Name}" }); } - modelInputBudget ??= new ModelInputBatchBudget(MaxModelInputBatchBytes); - var modelInputMaterialization = MaterializeModelInputFiles(context, sessionDir, logger, modelInputBudget); + var modelInputMaterialization = MaterializeModelInputFiles( + context, batch.SessionDirectory, _logger, modelInputBudget); // No inline clamp here: DispatchingToolExecutor already bounds every tool // result to the inline budget N (and spills the overflow). Clamping again // would re-window the already-windowed+steered result. @@ -783,18 +874,13 @@ internal static SubAgentFindingReviewResult ReviewSubAgentFinding( return new(SubAgentFindingReviewDecision.Accepted, null); } - private static async Task RouteToBackgroundJobAsync( + private async Task RouteToBackgroundJobAsync( FunctionCallContent tc, - SessionId sessionId, - MessageSource? source, - IToolAuditLogger? auditLogger, - TimeProvider timeProvider, - TurnContext? turnContext, + SessionToolBatch batch, ToolCallMeta meta, IActorRef backgroundJobManager, int timeoutSeconds, TimeSpan duration, - ILogger? logger, string? approvalDecision = null, string? approvalPattern = null) { @@ -816,11 +902,8 @@ private static async Task RouteToBackgroundJobAsync( // A background job inherits the submitting turn's authority context. // There is no safe default — defaulting a missing context to Personal // would silently escalate the job's audience. - var audience = turnContext?.Audience ?? source?.Audience; - var boundary = turnContext?.Boundary ?? source?.Boundary; - var channelType = turnContext?.ChannelType ?? source?.ChannelType; - var senderId = turnContext?.RequesterSenderId ?? source?.SenderId; - if (audience is null || boundary is null || channelType is null) + var channelType = batch.TurnContext.ChannelType; + if (channelType is null) throw new InvalidOperationException( "Background-job submission requires turn authority context; trust context cannot be defaulted."); @@ -828,13 +911,13 @@ private static async Task RouteToBackgroundJobAsync( { Command = command, WorkingDirectory = workingDirectory, - SessionId = sessionId, + SessionId = batch.SessionId, Rationale = meta.Rationale ?? "background shell execution", - Audience = audience.Value, - Boundary = boundary.Value, + Audience = batch.TurnContext.Audience, + Boundary = batch.TurnContext.Boundary, OriginChannelType = channelType.Value, TimeoutSeconds = timeoutSeconds, - SenderId = senderId + SenderId = batch.TurnContext.RequesterSenderId }; try @@ -842,11 +925,11 @@ private static async Task RouteToBackgroundJobAsync( var started = await backgroundJobManager.Ask( startCmd, TimeSpan.FromSeconds(30)); - logger?.LogInformation( + _logger.Info( "Background job {JobId} submitted for shell command (session {SessionId})", - started.JobId.Value, sessionId.Value); + started.JobId.Value, batch.SessionId.Value); - auditLogger?.Log(BuildAuditEntry(sessionId, tc, timeProvider, duration, meta) with + _auditLogger.Log(BuildAuditEntry(batch.SessionId, tc, _timeProvider, duration, meta) with { Allowed = true, ApprovalDecision = approvalDecision, @@ -870,18 +953,18 @@ private static async Task RouteToBackgroundJobAsync( JobId = started.JobId, Command = command, Rationale = startCmd.Rationale, - StartedAtMs = timeProvider.GetUtcNow().ToUnixTimeMilliseconds(), - Audience = audience.Value, - Boundary = boundary.Value, + StartedAtMs = _timeProvider.GetUtcNow().ToUnixTimeMilliseconds(), + Audience = batch.TurnContext.Audience, + Boundary = batch.TurnContext.Boundary, OutputLogPath = started.OutputLogPath }; return new ToolCallResult(resultMessage, [], [], [], [], jobInfo); } catch (Exception ex) { - logger?.LogError(ex, "Failed to submit background job for {ToolName}", tc.Name); + _logger.Error(ex, "Failed to submit background job for {ToolName}", tc.Name); - auditLogger?.Log(BuildAuditEntry(sessionId, tc, timeProvider, duration, meta) with + _auditLogger.Log(BuildAuditEntry(batch.SessionId, tc, _timeProvider, duration, meta) with { Allowed = false, DenyReason = $"background_job_submission_failed:{ex.GetType().Name}", @@ -903,7 +986,7 @@ private static async Task RouteToBackgroundJobAsync( internal static ModelInputMaterializationResult MaterializeModelInputFiles( ToolExecutionContext context, string sessionDir, - ILogger? logger, + ILoggingAdapter logger, ModelInputBatchBudget? batchBudget = null) { if (context.Outputs.ModelInputFiles.Count == 0) @@ -916,7 +999,7 @@ internal static ModelInputMaterializationResult MaterializeModelInputFiles( catch (Exception ex) when (ex is IOException or UnauthorizedAccessException) { var mediaDir = Path.Combine(sessionDir, SessionDirectoryHelper.MediaSubdirectory); - logger?.LogWarning(ex, "Failed to create model input media directory: {Path}", mediaDir); + logger.Warning(ex, "Failed to create model input media directory: {Path}", mediaDir); return new ModelInputMaterializationResult([], context.Outputs.ModelInputFiles.Count); } @@ -934,13 +1017,13 @@ internal static ModelInputMaterializationResult MaterializeModelInputFiles( var mimeType = file.MimeType; if (!SessionMediaStore.TryGetSupportedModelInput(mimeType, out var mediaModality, out var requiredModelModality)) { - logger?.LogWarning("Model input file MIME type is not supported, skipping: {MimeType}", mimeType); + logger.Warning("Model input file MIME type is not supported, skipping: {MimeType}", mimeType); continue; } if (!context.ModelInputModalities.HasFlag(requiredModelModality)) { - logger?.LogWarning( + logger.Warning( "Model input file requires unavailable modality {Modality}, skipping: {Path}", requiredModelModality, file.FilePath); @@ -949,26 +1032,26 @@ internal static ModelInputMaterializationResult MaterializeModelInputFiles( if (!File.Exists(file.FilePath)) { - logger?.LogWarning("Model input file not found, skipping: {Path}", file.FilePath); + logger.Warning("Model input file not found, skipping: {Path}", file.FilePath); continue; } var info = new FileInfo(file.FilePath); if (info.Length <= 0) { - logger?.LogWarning("Model input file is empty, skipping: {Path}", file.FilePath); + logger.Warning("Model input file is empty, skipping: {Path}", file.FilePath); continue; } if (info.Length > MaxModelInputFileBytes) { - logger?.LogWarning("Model input file exceeds size limit, skipping: {Path}", file.FilePath); + logger.Warning("Model input file exceeds size limit, skipping: {Path}", file.FilePath); continue; } if (!batchBudget.TryReserve(info.Length)) { - logger?.LogWarning("Model input file would exceed batch size limit, skipping: {Path}", file.FilePath); + logger.Warning("Model input file would exceed batch size limit, skipping: {Path}", file.FilePath); continue; } @@ -976,7 +1059,7 @@ internal static ModelInputMaterializationResult MaterializeModelInputFiles( if (!IsFileMagicCompatible(file.FilePath, mimeType)) { - logger?.LogWarning( + logger.Warning( "Model input file MIME type does not match detected bytes, skipping: {Path}", file.FilePath); batchBudget.Release(reservedBytes); @@ -997,7 +1080,7 @@ internal static ModelInputMaterializationResult MaterializeModelInputFiles( // gap drives the model-input handoff warning, so this is not silent. batchBudget.Release(reservedBytes); reservedBytes = 0; - logger?.LogWarning("Model input image could not be bounded, skipping: {Path}", file.FilePath); + logger.Warning("Model input image could not be bounded, skipping: {Path}", file.FilePath); continue; } @@ -1016,7 +1099,7 @@ internal static ModelInputMaterializationResult MaterializeModelInputFiles( { if (reservedBytes > 0) batchBudget.Release(reservedBytes); - logger?.LogWarning(ex, "Failed to materialize model input file: {Path}", file.FilePath); + logger.Warning(ex, "Failed to materialize model input file: {Path}", file.FilePath); } } @@ -1047,51 +1130,8 @@ private static bool IsFileMagicCompatible(string path, MimeType mimeType) return string.Equals(MimeTypeCatalog.Normalize(detected), mimeType.Value, StringComparison.OrdinalIgnoreCase); } - private static ToolExecutionContext BuildToolExecutionContext( - SessionId sessionId, - MessageSource? source, - string sessionDir, - Func> spawnChildActor, - IParentApprovalBridge? approvalBridge, - string? projectDirectory, - IReadOnlyList? recentFiles, - TurnContext? turnContext, - ModelModality modelInputModalities, - int maxInlineToolResultChars, - TimeSpan timeout, - ToolExecutionOutputs outputs) - { - // This legacy fallback disappears when TurnContext becomes a required - // batch member in Stage 2. Until then it is resolved once at the seam, - // never independently by downstream policy code. - var runScope = new ToolRunScope - { - Session = new ToolSessionScope.Bound(sessionId.Value, sessionDir), - Audience = turnContext?.Audience ?? source?.Audience ?? TrustAudience.Public, - InlineOutputBudget = new InlineOutputBudget(maxInlineToolResultChars), - Boundary = turnContext?.Boundary ?? source?.Boundary, - ChannelType = turnContext?.ChannelType?.ToWireValue() - ?? (source is null ? null : source.ChannelType.ToWireValue()), - DefaultDeliveryTarget = turnContext?.DefaultDeliveryTarget ?? source?.DefaultDeliveryTarget, - RequestedDeliveryTarget = turnContext?.RequestedDeliveryTarget ?? source?.RequestedDeliveryTarget, - SupportsInteractiveApproval = turnContext?.SupportsInteractiveApproval - ?? source?.ChannelType.SupportsInteractiveApproval(), - ModelInputModalities = modelInputModalities, - SpawnChildActor = spawnChildActor, - ApprovalBridge = approvalBridge, - ProjectDirectory = projectDirectory, - RecentFiles = recentFiles ?? [], - }; - return new ToolExecutionContext(runScope, new ToolExecutionTimeout(timeout), outputs); - } - - private static bool CanRequestInteractiveApproval(MessageSource? source, TurnContext? turnContext) - { - if (turnContext is not null) - return turnContext.SupportsInteractiveApproval && turnContext.HasApprovalRequester; - - return source is not null && source.ChannelType.SupportsInteractiveApproval(); - } + private static bool CanRequestInteractiveApproval(TurnContext turnContext) + => turnContext.SupportsInteractiveApproval && turnContext.HasApprovalRequester; private static ToolAuditEntry BuildAuditEntry( SessionId sessionId, diff --git a/src/Netclaw.Actors/Sessions/SessionDependencies.cs b/src/Netclaw.Actors/Sessions/SessionDependencies.cs index e1f21e80a..75c0f10c6 100644 --- a/src/Netclaw.Actors/Sessions/SessionDependencies.cs +++ b/src/Netclaw.Actors/Sessions/SessionDependencies.cs @@ -29,7 +29,7 @@ public sealed record SessionServices( /// public sealed record SessionToolServices( IToolExecutor ToolExecutor, - IToolAuditLogger? AuditLogger, + IToolAuditLogger AuditLogger, ToolRegistry ToolRegistry, ToolAccessPolicy? AccessPolicy, TrustContextDeriver? TrustDeriver, diff --git a/src/Netclaw.Actors/SubAgents/SubAgentActor.cs b/src/Netclaw.Actors/SubAgents/SubAgentActor.cs index c0e2ea888..aed0e39d9 100644 --- a/src/Netclaw.Actors/SubAgents/SubAgentActor.cs +++ b/src/Netclaw.Actors/SubAgents/SubAgentActor.cs @@ -1403,7 +1403,7 @@ private static ModelInputMaterializationResult MaterializeModelInputFiles( return SessionToolExecutionPipeline.MaterializeModelInputFiles( toolContext, toolContext.SessionDirectory, - logger: null, + NoLogger.Instance, modelInputBudget); } diff --git a/src/Netclaw.Actors/Tools/IToolExecutor.cs b/src/Netclaw.Actors/Tools/IToolExecutor.cs index d5cddaac7..d6b4e7088 100644 --- a/src/Netclaw.Actors/Tools/IToolExecutor.cs +++ b/src/Netclaw.Actors/Tools/IToolExecutor.cs @@ -142,3 +142,21 @@ public interface IToolAuditLogger { void Log(ToolAuditEntry entry); } + +/// +/// Explicit audit sink for deployments that do not configure durable tool auditing. +/// Keeps the execution pipeline's audit dependency required without manufacturing +/// nullable branches at every allow and deny path. +/// +public sealed class NullToolAuditLogger : IToolAuditLogger +{ + public static NullToolAuditLogger Instance { get; } = new(); + + private NullToolAuditLogger() + { + } + + public void Log(ToolAuditEntry entry) + { + } +} diff --git a/src/Netclaw.Daemon/Program.cs b/src/Netclaw.Daemon/Program.cs index 0eec0db14..ec488a61a 100644 --- a/src/Netclaw.Daemon/Program.cs +++ b/src/Netclaw.Daemon/Program.cs @@ -965,7 +965,7 @@ static void ConfigureDaemonServices( services.AddSingleton(sp => new SessionToolServices( sp.GetRequiredService(), - sp.GetService(), + sp.GetService() ?? NullToolAuditLogger.Instance, sp.GetRequiredService(), sp.GetService(), sp.GetService(), From 532fc9958787773e48b6706448e12c1213077a86 Mon Sep 17 00:00:00 2001 From: Aaron Stannard Date: Tue, 14 Jul 2026 22:26:05 +0000 Subject: [PATCH 3/4] fix(tools): remove unused batch timeout local --- .../Sessions/Pipelines/SessionToolExecutionPipeline.cs | 1 - 1 file changed, 1 deletion(-) diff --git a/src/Netclaw.Actors/Sessions/Pipelines/SessionToolExecutionPipeline.cs b/src/Netclaw.Actors/Sessions/Pipelines/SessionToolExecutionPipeline.cs index 3f41b52d6..77c6ab8c9 100644 --- a/src/Netclaw.Actors/Sessions/Pipelines/SessionToolExecutionPipeline.cs +++ b/src/Netclaw.Actors/Sessions/Pipelines/SessionToolExecutionPipeline.cs @@ -259,7 +259,6 @@ public async Task ExecuteAsync(SessionToolBatch batch) try { batch.Validate(); - var timeout = batch.DefaultTimeout.Value; // Execute all tool calls in parallel. Calls are not always // independent -- e.g. two file_edit calls on the same file -- so // file-mutating tools serialize their read-modify-write per target From b6177b84995234b0f90b35fba68af4ec3147fb5f Mon Sep 17 00:00:00 2001 From: Aaron Stannard Date: Tue, 14 Jul 2026 22:32:02 +0000 Subject: [PATCH 4/4] fix(tools): require explicit audit logger registration --- src/Netclaw.Actors.Tests/Sessions/LlmSessionTestExtensions.cs | 3 ++- src/Netclaw.Daemon/Program.cs | 3 ++- 2 files changed, 4 insertions(+), 2 deletions(-) diff --git a/src/Netclaw.Actors.Tests/Sessions/LlmSessionTestExtensions.cs b/src/Netclaw.Actors.Tests/Sessions/LlmSessionTestExtensions.cs index eaf453f0b..1eb6ead62 100644 --- a/src/Netclaw.Actors.Tests/Sessions/LlmSessionTestExtensions.cs +++ b/src/Netclaw.Actors.Tests/Sessions/LlmSessionTestExtensions.cs @@ -42,9 +42,10 @@ public static IServiceCollection AddLlmSessionCompositeRecords(this IServiceColl if (services.Any(d => d.ServiceType == typeof(IToolExecutor))) { + services.TryAddSingleton(NullToolAuditLogger.Instance); services.TryAddSingleton(sp => new SessionToolServices( sp.GetRequiredService(), - sp.GetService() ?? NullToolAuditLogger.Instance, + sp.GetRequiredService(), sp.GetRequiredService(), sp.GetService(), sp.GetService(), diff --git a/src/Netclaw.Daemon/Program.cs b/src/Netclaw.Daemon/Program.cs index ec488a61a..f1772c2c7 100644 --- a/src/Netclaw.Daemon/Program.cs +++ b/src/Netclaw.Daemon/Program.cs @@ -739,6 +739,7 @@ static void ConfigureDaemonServices( toolAccessPolicy, sp.GetService(), sp.GetRequiredService>())); + services.AddSingleton(NullToolAuditLogger.Instance); // Operational notification webhooks var notificationsConfig = configuration.GetSection("Notifications") @@ -965,7 +966,7 @@ static void ConfigureDaemonServices( services.AddSingleton(sp => new SessionToolServices( sp.GetRequiredService(), - sp.GetService() ?? NullToolAuditLogger.Instance, + sp.GetRequiredService(), sp.GetRequiredService(), sp.GetService(), sp.GetService(),