From fddcc3fefe320263819c002091efd0c24dcbf19f Mon Sep 17 00:00:00 2001 From: rcourtman Date: Thu, 16 Jul 2026 11:39:23 +0100 Subject: [PATCH] Honor Patrol lifecycle tool ordering --- .../v6/internal/subsystems/ai-runtime.md | 14 +- internal/ai/chat/agentic.go | 43 +++++- internal/ai/chat/agentic_additional_test.go | 124 ++++++++++++++++++ 3 files changed, 172 insertions(+), 9 deletions(-) diff --git a/docs/release-control/v6/internal/subsystems/ai-runtime.md b/docs/release-control/v6/internal/subsystems/ai-runtime.md index adbaa20f6..01323e31d 100644 --- a/docs/release-control/v6/internal/subsystems/ai-runtime.md +++ b/docs/release-control/v6/internal/subsystems/ai-runtime.md @@ -276,8 +276,8 @@ the message history as untrusted evidence. It explicitly defines returned Claude Code tool activity. This separation must not collapse back into a user prompt that labels Pulse's own system instruction untrusted or activates a coding/plan-mode persona. Neither route receives Pulse MCP configuration. The -agent returns tool arguments as encoded JSON; Pulse parses them and rejects unknown tool names, -duplicate or empty call IDs, malformed argument objects, and violations of +agent returns tool arguments as encoded JSON; Pulse parses them and rejects +unknown tool names, duplicate or empty call IDs, malformed argument objects, and violations of `none` or `required` tool choice before the existing provider-neutral tool loop can execute anything. The normal registry, profile, approval, protected resource, action, and verification contracts remain the only infrastructure @@ -324,6 +324,16 @@ The configured Z.ai `/api/coding/paas/` endpoint is likewise reported as `coding_plan_allowance`, with unknown monetary cost and no per-run metered-API dollar budget; the standard `/api/paas/` endpoint remains `metered_api`. +The provider-neutral agentic loop runs independent tool calls in parallel, but +parallelism must not erase a same-turn Patrol lifecycle dependency. When a +provider returns `patrol_get_findings` together with +`patrol_report_finding`, `patrol_assess_finding`, or +`patrol_resolve_finding`, Pulse executes that batch sequentially in the +provider's original order. It does not reorder an invalid write-before-read +batch, and it retains parallel execution for independent reads and independent +finding writes. This makes the existing read-before-write guard deterministic +without requiring the model to understand an internal goroutine schedule. + ## Canonical Files 1. `internal/ai/` diff --git a/internal/ai/chat/agentic.go b/internal/ai/chat/agentic.go index e9eece35f..cdd1e1591 100644 --- a/internal/ai/chat/agentic.go +++ b/internal/ai/chat/agentic.go @@ -364,6 +364,25 @@ func isPatrolFindingLifecycleWrite(toolName string) bool { } } +// requiresOrderedPatrolFindingLifecycleExecution identifies a same-turn +// read-before-write dependency. Independent reads and independent finding +// writes may still run in parallel, but a lifecycle write cannot race the +// patrol_get_findings call that establishes its deduplication/assessment +// precondition. The provider's original call order remains authoritative. +func requiresOrderedPatrolFindingLifecycleExecution(toolCalls []providers.ToolCall) bool { + hasFindingsRead := false + hasLifecycleWrite := false + for _, tc := range toolCalls { + switch { + case strings.TrimSpace(tc.Name) == agentcapabilities.PatrolGetFindingsToolName: + hasFindingsRead = true + case isPatrolFindingLifecycleWrite(tc.Name): + hasLifecycleWrite = true + } + } + return hasFindingsRead && hasLifecycleWrite +} + func applySuccessfulToolFSM(fsm *SessionFSM, toolKind ToolKind, toolName string) bool { if fsm == nil { return false @@ -1726,9 +1745,10 @@ func (a *AgenticLoop) executeWithTools(ctx context.Context, sessionID string, me pendingExec = append(pendingExec, pendingToolExec{tc: tc, toolKind: toolKind}) } - // --- Phase 2: Execute pending tools (parallel if multiple) --- - // Tool execution is stateless I/O — safe to parallelize. - // Cap concurrency at 4 to avoid overwhelming infrastructure. + // --- Phase 2: Execute pending tools --- + // Independent calls run in parallel with concurrency capped at four. + // Ordered Patrol finding lifecycle batches run sequentially so their + // read-before-write precondition cannot race within one provider turn. execResults := make([]parallelToolResult, len(pendingExec)) if len(pendingExec) > 0 { for _, pe := range pendingExec { @@ -1755,7 +1775,8 @@ func (a *AgenticLoop) executeWithTools(ctx context.Context, sessionID string, me emitWorkflowState(callback, "execute", executeMessage, sessionFSMState(fsm), workflowTool) } - if len(pendingExec) > 1 { + orderedPatrolLifecycle := requiresOrderedPatrolFindingLifecycleExecution(toolCalls) + if len(pendingExec) > 1 && !orderedPatrolLifecycle { log.Info(). Int("tool_count", len(pendingExec)). Str("session_id", sessionID). @@ -1775,9 +1796,17 @@ func (a *AgenticLoop) executeWithTools(ctx context.Context, sessionID string, me }(j, pe.tc) } wg.Wait() - } else if len(pendingExec) == 1 { - r, e := a.executeToolSafely(ctx, pendingExec[0].tc.ID, pendingExec[0].tc.Name, pendingExec[0].tc.Input) - execResults[0] = parallelToolResult{Result: r, Err: e} + } else { + if orderedPatrolLifecycle && len(pendingExec) > 1 { + log.Debug(). + Int("tool_count", len(pendingExec)). + Str("session_id", sessionID). + Msg("[AgenticLoop] Executing ordered Patrol finding lifecycle batch") + } + for j, pe := range pendingExec { + r, e := a.executeToolSafely(ctx, pe.tc.ID, pe.tc.Name, pe.tc.Input) + execResults[j] = parallelToolResult{Result: r, Err: e} + } } // --- Phase 3: Post-process results in original order (sequential) --- diff --git a/internal/ai/chat/agentic_additional_test.go b/internal/ai/chat/agentic_additional_test.go index 556d5fdcc..bed8fb0de 100644 --- a/internal/ai/chat/agentic_additional_test.go +++ b/internal/ai/chat/agentic_additional_test.go @@ -5,13 +5,36 @@ import ( "encoding/json" "errors" "strings" + "sync" "testing" + "time" "github.com/rcourtman/pulse-go-rewrite/internal/agentcapabilities" "github.com/rcourtman/pulse-go-rewrite/internal/ai/providers" "github.com/rcourtman/pulse-go-rewrite/internal/ai/tools" ) +func TestRequiresOrderedPatrolFindingLifecycleExecution(t *testing.T) { + tests := []struct { + name string + calls []providers.ToolCall + want bool + }{ + {name: "independent reads stay parallel", calls: []providers.ToolCall{{Name: agentcapabilities.PulseQueryToolName}, {Name: agentcapabilities.PulseReadToolName}}}, + {name: "independent finding reports stay parallel", calls: []providers.ToolCall{{Name: agentcapabilities.PatrolReportFindingToolName}, {Name: agentcapabilities.PatrolReportFindingToolName}}}, + {name: "findings read before assessment is ordered", calls: []providers.ToolCall{{Name: agentcapabilities.PatrolGetFindingsToolName}, {Name: agentcapabilities.PatrolAssessFindingToolName}}, want: true}, + {name: "findings read before report is ordered", calls: []providers.ToolCall{{Name: agentcapabilities.PatrolGetFindingsToolName}, {Name: agentcapabilities.PatrolReportFindingToolName}}, want: true}, + {name: "provider order remains relevant even when reversed", calls: []providers.ToolCall{{Name: agentcapabilities.PatrolResolveFindingToolName}, {Name: agentcapabilities.PatrolGetFindingsToolName}}, want: true}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := requiresOrderedPatrolFindingLifecycleExecution(tt.calls); got != tt.want { + t.Fatalf("requiresOrderedPatrolFindingLifecycleExecution() = %v, want %v", got, tt.want) + } + }) + } +} + func TestAgenticLoop_Setters(t *testing.T) { loop := &AgenticLoop{} loop.SetMaxTurns(7) @@ -350,6 +373,41 @@ type repairTestPatrolFindingCreator struct { created []tools.PatrolFindingInput } +type orderedPatrolFindingCreator struct { + mu sync.Mutex + checked bool + assessments []tools.PatrolFindingAssessmentInput +} + +func (c *orderedPatrolFindingCreator) CreateFinding(tools.PatrolFindingInput) (string, bool, error) { + return "finding-existing", false, nil +} + +func (c *orderedPatrolFindingCreator) ResolveFinding(string, string) error { return nil } + +func (c *orderedPatrolFindingCreator) GetActiveFindings(string, string) []tools.PatrolFindingInfo { + // Make the historical parallel-execution race deterministic: an assessment + // started alongside this read observes checked=false and is rejected. + time.Sleep(25 * time.Millisecond) + c.mu.Lock() + c.checked = true + c.mu.Unlock() + return []tools.PatrolFindingInfo{{ID: "finding-existing", Severity: "warning", Category: "reliability", ResourceID: "app-container-existing", ResourceName: "existing", ResourceType: "app-container", Title: "Container unhealthy"}} +} + +func (c *orderedPatrolFindingCreator) HasCheckedFindings() bool { + c.mu.Lock() + defer c.mu.Unlock() + return c.checked +} + +func (c *orderedPatrolFindingCreator) AssessFinding(input tools.PatrolFindingAssessmentInput) error { + c.mu.Lock() + defer c.mu.Unlock() + c.assessments = append(c.assessments, input) + return nil +} + func (c *repairTestPatrolFindingCreator) CreateFinding(input tools.PatrolFindingInput) (string, bool, error) { c.created = append(c.created, input) return "finding-" + input.ResourceID, true, nil @@ -385,6 +443,72 @@ func (s *stubStreamingProvider) ListModels(ctx context.Context) ([]providers.Mod return nil, nil } +func TestAgenticLoopOrdersSameTurnPatrolFindingReadBeforeAssessment(t *testing.T) { + provider := &stubStreamingProvider{} + executor := tools.NewPulseToolExecutor(tools.ExecutorConfig{}) + executor.ApplyExecutionProfile(tools.ProfilePatrolDetection) + creator := &orderedPatrolFindingCreator{} + executor.SetPatrolFindingCreator(creator) + loop := NewAgenticLoop(provider, executor, "full Patrol prompt") + loop.SetExecutionProfile(tools.ProfilePatrolDetection) + loop.SetMaxTurns(2) + + providerCalls := 0 + provider.chatStream = func(_ context.Context, _ providers.ChatRequest, callback providers.StreamCallback) error { + providerCalls++ + switch providerCalls { + case 1: + getCall := providers.ToolCall{ID: "get-findings", Name: agentcapabilities.PatrolGetFindingsToolName, Input: map[string]interface{}{}} + assessCall := providers.ToolCall{ID: "assess-finding", Name: agentcapabilities.PatrolAssessFindingToolName, Input: map[string]interface{}{ + "finding_id": "finding-existing", + "verdict": "present", + "evidence": "Current provider state remains unhealthy.", + "reason": "The current health check still fails.", + }} + for _, call := range []providers.ToolCall{getCall, assessCall} { + callback(providers.StreamEvent{Type: "tool_start", Data: providers.ToolStartEvent{ID: call.ID, Name: call.Name, Input: call.Input}}) + } + callback(providers.StreamEvent{Type: "done", Data: providers.DoneEvent{ToolCalls: []providers.ToolCall{getCall, assessCall}}}) + case 2: + callback(providers.StreamEvent{Type: "content", Data: providers.ContentEvent{Text: "Finding remains present."}}) + callback(providers.StreamEvent{Type: "done", Data: providers.DoneEvent{}}) + default: + t.Fatalf("unexpected provider call %d", providerCalls) + } + return nil + } + + available := []providers.Tool{ + {Name: agentcapabilities.PatrolGetFindingsToolName}, + {Name: agentcapabilities.PatrolAssessFindingToolName}, + } + var toolEnds []ToolEndData + _, err := loop.ExecuteWithTools(context.Background(), "ordered-finding-lifecycle", []Message{{Role: "user", Content: "Reassess the active finding."}}, available, func(event StreamEvent) { + if event.Type != "tool_end" { + return + } + var data ToolEndData + if err := json.Unmarshal(event.Data, &data); err != nil { + t.Fatalf("decode tool_end: %v", err) + } + toolEnds = append(toolEnds, data) + }) + if err != nil { + t.Fatalf("ordered Patrol finding lifecycle failed: %v", err) + } + if providerCalls != 2 { + t.Fatalf("provider calls = %d, want lifecycle turn and summary", providerCalls) + } + if len(toolEnds) != 2 || !toolEnds[0].Success || !toolEnds[1].Success { + t.Fatalf("tool results = %+v, want ordered successful read and assessment", toolEnds) + } + creator.mu.Lock() + defer creator.mu.Unlock() + if len(creator.assessments) != 1 || creator.assessments[0].FindingID != "finding-existing" { + t.Fatalf("assessments = %+v, want one existing-finding verdict", creator.assessments) + } +} + func TestAgenticLoopRepairsRejectedSiblingAfterMixedPatrolFindingBatch(t *testing.T) { provider := &stubStreamingProvider{} executor := tools.NewPulseToolExecutor(tools.ExecutorConfig{})