mirror of
https://github.com/rcourtman/Pulse.git
synced 2026-10-19 18:44:53 +00:00
Allow bounded subscription-agent preflights
This commit is contained in:
parent
2bc48774c2
commit
b84f650e77
9 changed files with 189 additions and 21 deletions
|
|
@ -273,11 +273,16 @@ duplicate or empty call IDs, malformed argument objects, and violations of
|
|||
can execute anything. The normal registry, profile, approval, protected
|
||||
resource, action, and verification contracts remain the only infrastructure
|
||||
authority. Calls are serialized per local subscription agent and bounded by
|
||||
the configured request timeout and prompt/output size limits. The native
|
||||
provider boundary accepts the canonical qualified Pulse model identity from
|
||||
shared callers, strips only its own subscription-provider prefix before CLI
|
||||
execution, and rejects foreign provider prefixes rather than forwarding an
|
||||
invalid or cross-provider model name to the local agent.
|
||||
the configured request timeout and prompt/output size limits, with a two-minute
|
||||
minimum request allowance for local subscription turns. Patrol tool-call
|
||||
preflight owns one route-aware outer deadline: API providers retain the
|
||||
30-second budget, while local subscription agents receive a bounded two-minute
|
||||
budget for CLI startup and complete structured-output assembly. A caller's
|
||||
earlier cancellation still wins. The native provider boundary accepts the
|
||||
canonical qualified Pulse model identity from shared callers, strips only its
|
||||
own subscription-provider prefix before CLI execution, and rejects foreign
|
||||
provider prefixes rather than forwarding an invalid or cross-provider model
|
||||
name to the local agent.
|
||||
Patrol consumes the provider streaming interface, so the adapter projects each
|
||||
fully validated CLI turn into canonical buffered `content`, `tool_start`, and
|
||||
`done` events. It must emit nothing before the complete CLI response passes the
|
||||
|
|
|
|||
|
|
@ -1503,7 +1503,12 @@ payload shape change when the portal presents compact client rows.
|
|||
provider-test and Patrol-preflight endpoints remain the authority for
|
||||
binary, login, model-access, and structured-tool readiness. No API response
|
||||
may expose CLI credential material, and no failure may silently substitute
|
||||
the corresponding metered API provider.
|
||||
the corresponding metered API provider. The manual Patrol-preflight handler
|
||||
and settings-triggered asynchronous preflight must delegate deadline
|
||||
ownership to the AI runtime so route policy cannot drift at the API edge:
|
||||
API providers retain the 30-second preflight budget, local subscription
|
||||
agents receive the bounded two-minute preflight budget, and an earlier
|
||||
request cancellation still wins.
|
||||
Legacy Assistant SSE routes in this handler that still use the older
|
||||
execute envelope, including `/api/ai/execute/stream` and
|
||||
`/api/ai/investigate-alert`, must preserve their existing top-level
|
||||
|
|
|
|||
|
|
@ -42,6 +42,17 @@ type PatrolPreflightResult struct {
|
|||
// outside preflight has no operational meaning.
|
||||
const patrolPreflightToolName = "verify_pulse_patrol"
|
||||
|
||||
const defaultPatrolPreflightTimeout = 30 * time.Second
|
||||
|
||||
func patrolPreflightTimeout(provider string) time.Duration {
|
||||
switch strings.TrimSpace(provider) {
|
||||
case config.AIProviderCodexSubscription, config.AIProviderClaudeSubscription:
|
||||
return providers.SubscriptionAgentMinimumRequestTimeout
|
||||
default:
|
||||
return defaultPatrolPreflightTimeout
|
||||
}
|
||||
}
|
||||
|
||||
// patrolPreflightCache holds the most recent PatrolPreflightResult plus
|
||||
// the wall-clock time it was recorded. Surfaced through the AI settings
|
||||
// response so the UI can render a "last verified" indicator without
|
||||
|
|
@ -81,14 +92,11 @@ func (s *Service) recordPatrolPreflight(result PatrolPreflightResult, at time.Ti
|
|||
// background so callers (notably the settings save handler) don't block
|
||||
// on a 5-10 second LLM round-trip. The result populates the preflight
|
||||
// cache and the next /api/settings/ai poll surfaces it through the
|
||||
// patrol_preflight snapshot. The 30s timeout matches the manual
|
||||
// preflight handler so a stuck provider can't pin the goroutine
|
||||
// indefinitely.
|
||||
// patrol_preflight snapshot. RunPatrolToolPreflight owns the route-aware
|
||||
// deadline so the asynchronous and manual entrypoints cannot drift.
|
||||
func (s *Service) TriggerPatrolPreflightAsync(provider, model string) {
|
||||
go func() {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
||||
defer cancel()
|
||||
s.RunPatrolToolPreflight(ctx, provider, model)
|
||||
s.RunPatrolToolPreflight(context.Background(), provider, model)
|
||||
}()
|
||||
}
|
||||
|
||||
|
|
@ -176,6 +184,8 @@ func (s *Service) RunPatrolToolPreflight(ctx context.Context, providerName, mode
|
|||
parsedProvider, parsedModel := config.ParseModelString(modelStr)
|
||||
result.Provider = parsedProvider
|
||||
result.Model = parsedModel
|
||||
preflightCtx, cancel := context.WithTimeout(ctx, patrolPreflightTimeout(parsedProvider))
|
||||
defer cancel()
|
||||
|
||||
provider, err := providers.NewForModel(cfg, modelStr)
|
||||
if err != nil {
|
||||
|
|
@ -216,7 +226,7 @@ func (s *Service) RunPatrolToolPreflight(ctx context.Context, providerName, mode
|
|||
req.ToolChoice = &providers.ToolChoice{Type: providers.ToolChoiceRequired}
|
||||
}
|
||||
|
||||
resp, err := provider.Chat(ctx, req)
|
||||
resp, err := provider.Chat(preflightCtx, req)
|
||||
result.DurationMs = time.Since(started).Milliseconds()
|
||||
|
||||
if err != nil {
|
||||
|
|
|
|||
|
|
@ -6,6 +6,9 @@ import (
|
|||
"errors"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"runtime"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
|
@ -44,6 +47,80 @@ func newPatrolPreflightTestService(t *testing.T, model string, handler http.Hand
|
|||
return &patrolPreflightTestService{svc: svc, server: server}
|
||||
}
|
||||
|
||||
func TestPatrolPreflightTimeoutIsRouteAware(t *testing.T) {
|
||||
tests := []struct {
|
||||
provider string
|
||||
want time.Duration
|
||||
}{
|
||||
{provider: config.AIProviderOpenAI, want: 30 * time.Second},
|
||||
{provider: config.AIProviderAnthropic, want: 30 * time.Second},
|
||||
{provider: config.AIProviderCodexSubscription, want: 2 * time.Minute},
|
||||
{provider: config.AIProviderClaudeSubscription, want: 2 * time.Minute},
|
||||
{provider: "", want: 30 * time.Second},
|
||||
{provider: "future-provider", want: 30 * time.Second},
|
||||
}
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.provider, func(t *testing.T) {
|
||||
if got := patrolPreflightTimeout(tt.provider); got != tt.want {
|
||||
t.Fatalf("patrolPreflightTimeout(%q) = %s, want %s", tt.provider, got, tt.want)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestRunPatrolToolPreflightSubscriptionRoutePreservesCallerCancellation(t *testing.T) {
|
||||
if runtime.GOOS == "windows" {
|
||||
t.Skip("fake subscription CLI uses a POSIX shell script")
|
||||
}
|
||||
binDir := t.TempDir()
|
||||
claudePath := filepath.Join(binDir, "claude")
|
||||
if err := os.WriteFile(claudePath, []byte(`#!/bin/sh
|
||||
while :; do :; done
|
||||
`), 0700); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Setenv("PATH", binDir+string(os.PathListSeparator)+os.Getenv("PATH"))
|
||||
|
||||
svc := NewService(config.NewConfigPersistence(t.TempDir()), nil)
|
||||
svc.cfg = &config.AIConfig{
|
||||
Enabled: true,
|
||||
Model: "claude-subscription:opus",
|
||||
PatrolModel: "claude-subscription:opus",
|
||||
ClaudeSubscriptionEnabled: true,
|
||||
RequestTimeoutSeconds: 1,
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Millisecond)
|
||||
defer cancel()
|
||||
started := time.Now()
|
||||
result := svc.RunPatrolToolPreflight(ctx, "", "")
|
||||
if result.Success {
|
||||
t.Fatalf("expected caller cancellation to fail subscription preflight, got %+v", result)
|
||||
}
|
||||
if elapsed := time.Since(started); elapsed > 500*time.Millisecond {
|
||||
t.Fatalf("subscription caller cancellation took %s", elapsed)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRunPatrolToolPreflightPreservesCallerCancellation(t *testing.T) {
|
||||
h := newPatrolPreflightTestService(t, "openai:gpt-4o-mini", func(w http.ResponseWriter, r *http.Request) {
|
||||
time.Sleep(200 * time.Millisecond)
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
_, _ = w.Write([]byte(`{"id":"late","model":"gpt-4o-mini","choices":[{"finish_reason":"stop","message":{"role":"assistant","content":"late"}}]}`))
|
||||
})
|
||||
defer h.close()
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond)
|
||||
defer cancel()
|
||||
started := time.Now()
|
||||
result := h.svc.RunPatrolToolPreflight(ctx, "", "")
|
||||
if result.Success {
|
||||
t.Fatalf("expected caller cancellation to fail preflight, got %+v", result)
|
||||
}
|
||||
if elapsed := time.Since(started); elapsed > time.Second {
|
||||
t.Fatalf("caller cancellation took %s", elapsed)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRunPatrolToolPreflight_Success_ToolCallObserved(t *testing.T) {
|
||||
// Provider accepts the request and the model emits a tool call —
|
||||
// the green-path outcome we want operators to see.
|
||||
|
|
|
|||
|
|
@ -3,6 +3,7 @@ package providers
|
|||
import (
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/rcourtman/pulse-go-rewrite/internal/config"
|
||||
)
|
||||
|
|
@ -475,6 +476,37 @@ func TestNewForProvider_AnthropicOAuthUnsupported(t *testing.T) {
|
|||
}
|
||||
}
|
||||
|
||||
func TestNewForProvider_SubscriptionAgentsReceiveMinimumRequestAllowance(t *testing.T) {
|
||||
cfg := &config.AIConfig{
|
||||
Enabled: true,
|
||||
RequestTimeoutSeconds: 30,
|
||||
CodexSubscriptionEnabled: true,
|
||||
ClaudeSubscriptionEnabled: true,
|
||||
}
|
||||
tests := []struct {
|
||||
provider string
|
||||
model string
|
||||
}{
|
||||
{provider: config.AIProviderCodexSubscription, model: "gpt-5.6-luna"},
|
||||
{provider: config.AIProviderClaudeSubscription, model: "opus"},
|
||||
}
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.provider, func(t *testing.T) {
|
||||
provider, err := NewForProvider(cfg, tt.provider, tt.model)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
client, ok := provider.(*SubscriptionAgentClient)
|
||||
if !ok {
|
||||
t.Fatalf("provider = %T, want *SubscriptionAgentClient", provider)
|
||||
}
|
||||
if client.timeout != 2*time.Minute {
|
||||
t.Fatalf("timeout = %s, want 2m", client.timeout)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestNewForModel(t *testing.T) {
|
||||
cfg := &config.AIConfig{
|
||||
Enabled: true,
|
||||
|
|
|
|||
|
|
@ -24,6 +24,12 @@ const (
|
|||
SubscriptionAgentCodex SubscriptionAgent = "codex-subscription"
|
||||
SubscriptionAgentClaude SubscriptionAgent = "claude-subscription"
|
||||
|
||||
// SubscriptionAgentMinimumRequestTimeout accounts for local CLI startup,
|
||||
// subscription-plan routing, and non-streaming structured-output assembly.
|
||||
// Caller cancellation still wins, and Patrol preflight applies its own
|
||||
// route-aware outer deadline.
|
||||
SubscriptionAgentMinimumRequestTimeout = 2 * time.Minute
|
||||
|
||||
maxSubscriptionAgentPromptBytes = 4 << 20
|
||||
maxSubscriptionAgentOutputBytes = 8 << 20
|
||||
)
|
||||
|
|
@ -86,7 +92,18 @@ func (b *cappedBuffer) Write(p []byte) (int, error) {
|
|||
}
|
||||
|
||||
func NewSubscriptionAgentClient(agent SubscriptionAgent, model string, timeout time.Duration) *SubscriptionAgentClient {
|
||||
return &SubscriptionAgentClient{agent: agent, model: strings.TrimSpace(model), timeout: timeout}
|
||||
return &SubscriptionAgentClient{
|
||||
agent: agent,
|
||||
model: strings.TrimSpace(model),
|
||||
timeout: subscriptionAgentRequestTimeout(timeout),
|
||||
}
|
||||
}
|
||||
|
||||
func subscriptionAgentRequestTimeout(configured time.Duration) time.Duration {
|
||||
if configured < SubscriptionAgentMinimumRequestTimeout {
|
||||
return SubscriptionAgentMinimumRequestTimeout
|
||||
}
|
||||
return configured
|
||||
}
|
||||
|
||||
func (c *SubscriptionAgentClient) Name() string { return string(c.agent) }
|
||||
|
|
|
|||
|
|
@ -40,6 +40,16 @@ func TestCappedBufferBoundsChildOutput(t *testing.T) {
|
|||
}
|
||||
}
|
||||
|
||||
func TestSubscriptionAgentRequestTimeout(t *testing.T) {
|
||||
if got := subscriptionAgentRequestTimeout(30 * time.Second); got != SubscriptionAgentMinimumRequestTimeout {
|
||||
t.Fatalf("short configured timeout = %s, want %s", got, SubscriptionAgentMinimumRequestTimeout)
|
||||
}
|
||||
configured := 3 * time.Minute
|
||||
if got := subscriptionAgentRequestTimeout(configured); got != configured {
|
||||
t.Fatalf("long configured timeout = %s, want %s", got, configured)
|
||||
}
|
||||
}
|
||||
|
||||
func TestNormalizeSubscriptionAgentModel(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
|
|
|
|||
|
|
@ -6076,13 +6076,10 @@ func (h *AISettingsHandler) HandlePatrolPreflight(w http.ResponseWriter, r *http
|
|||
return
|
||||
}
|
||||
|
||||
// Tight timeout: preflight is a single round-trip, not a Patrol pass.
|
||||
// 30s matches the per-provider connection-test budget; in practice
|
||||
// this completes in well under 10s for any healthy provider.
|
||||
ctx, cancel := context.WithTimeout(r.Context(), 30*time.Second)
|
||||
defer cancel()
|
||||
|
||||
preflight := aiService.RunPatrolToolPreflight(ctx, body.Provider, body.Model)
|
||||
// The service owns the route-aware deadline. API providers keep the tight
|
||||
// preflight budget while local subscription agents receive enough time for
|
||||
// bounded CLI startup and structured-output assembly.
|
||||
preflight := aiService.RunPatrolToolPreflight(r.Context(), body.Provider, body.Model)
|
||||
|
||||
response := aiPatrolPreflightResponse{
|
||||
Success: preflight.Success && preflight.ToolCallObserved,
|
||||
|
|
|
|||
|
|
@ -3608,6 +3608,21 @@ func TestAISettingsHandler_PatrolPreflight_RejectsInvalidProviderName(t *testing
|
|||
assert.Equal(t, http.StatusBadRequest, rec.Code)
|
||||
}
|
||||
|
||||
func TestAISettingsHandler_PatrolPreflight_DelegatesDeadlineToAIRuntime(t *testing.T) {
|
||||
source, err := os.ReadFile("ai_handlers.go")
|
||||
require.NoError(t, err)
|
||||
text := string(source)
|
||||
start := strings.Index(text, "func (h *AISettingsHandler) HandlePatrolPreflight(")
|
||||
require.NotEqual(t, -1, start)
|
||||
end := strings.Index(text[start:], "\nfunc ")
|
||||
if end < 0 {
|
||||
end = len(text) - start
|
||||
}
|
||||
body := text[start : start+end]
|
||||
require.Contains(t, body, "RunPatrolToolPreflight(r.Context(), body.Provider, body.Model)")
|
||||
require.NotContains(t, body, "context.WithTimeout")
|
||||
}
|
||||
|
||||
// TestOrchestratorAndChatAdaptersMapTheSameMessageFields keeps the deliberate
|
||||
// GetMessages mirror between orchestratorChatAdapter (ai_handlers.go) and
|
||||
// chatServiceAdapter (chat_service_adapter.go) honest: both convert the same
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue